Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions agent/redact.py
Original file line number Diff line number Diff line change
Expand Up @@ -997,6 +997,44 @@ def _has_http_method_substring(text: str) -> bool:
return any(method in upper for method in _HTTP_METHOD_SUBSTRINGS)


# Env-var name suffixes that mark a value as a credential. The regex passes
# above mask tokens by *shape* (vendor prefixes like ``sk-``); this suffix
# list drives the exact-value pass below, which masks opaque values that
# carry no recognizable prefix (e.g. ``MY_SERVICE_TOKEN=abc123randomstring``).
_CREDENTIAL_VALUE_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY", "_PASSWORD")


def _known_secret_values() -> set[str]:
"""Return the exact values of credential-named env vars, when non-trivial.

Used by :func:`mask_known_secret_values` to strip a secret's literal
value out of text even when it has no recognizable vendor prefix. Values
shorter than 6 chars are skipped — masking ``KEY=true`` or ``TOKEN=1``
would mangle ordinary prose while adding no real protection.
"""
values: set[str] = set()
for name, value in os.environ.items():
if value and len(value) >= 6 and name.upper().endswith(_CREDENTIAL_VALUE_SUFFIXES):
values.add(value)
return values


def mask_known_secret_values(text: str) -> str:
"""Mask the exact values of known credential env vars wherever they appear.

Complements the shape-based regex passes: a secret value echoed into a
status line, warning, or error message is replaced with ``***`` even when
it doesn't match any vendor prefix. Best-effort — if a value is not
currently known to the process it cannot be masked here (the regex passes
still catch prefix-shaped tokens).
"""
if not text:
return text
for value in _known_secret_values():
text = text.replace(value, "***")
return text


class RedactingFormatter(logging.Formatter):
"""Log formatter that redacts secrets from all log messages."""

Expand Down
49 changes: 45 additions & 4 deletions hermes_cli/env_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -635,14 +635,55 @@ def _apply_external_secret_sources(home_path: Path) -> None:
file=sys.stderr,
)
if src.result.error:
print(f" {src.label}: {src.result.error}", file=sys.stderr)
print(
f" {src.label}: {_mask_secret_text(src.result.error, home_path)}",
file=sys.stderr,
)
hint = _remediation_hint(src.name, src.result.error_kind, cfg)
if hint:
print(f" {src.label}: → {hint}", file=sys.stderr)
print(
f" {src.label}: → {_mask_secret_text(hint, home_path)}",
file=sys.stderr,
)
for warn in src.result.warnings:
print(f" {src.label}: {warn}", file=sys.stderr)
print(
f" {src.label}: {_mask_secret_text(warn, home_path)}",
file=sys.stderr,
)
for conflict in report.conflicts:
print(f" Secret sources: {conflict}", file=sys.stderr)
print(
f" Secret sources: {_mask_secret_text(conflict, home_path)}",
file=sys.stderr,
)


def _mask_secret_text(text: str, home_path: str | os.PathLike) -> str:
"""Mask known secret values out of status text before it reaches stderr.

Uses the exact values applied from external secret sources for THIS home
(``_SECRET_SOURCE_VALUES_BY_HOME[home_key]`` — the authoritative set for
what this status line is about) plus the generic credential-env scan in
``agent.redact``. Error, hint, warning, and conflict lines can carry a
secret *value* (a backend echoing it, a remediation hint quoting it);
names are already suppressed on the applied-count line. Snapshot lookup
is scoped to ``home_path`` on purpose: snapshots are per-home (status
output for one profile must not depend on another profile's values).

Every non-empty value from the home's snapshot is masked — no minimum
length filter, because short external-source ``*_PASSWORD`` / ``*_TOKEN``
values (e.g. ``"ab"``) are exactly the ones a backend echo would leak.
Best-effort: never raises, never blocks startup.
"""
if not text:
return text
from agent.redact import mask_known_secret_values

home_key = str(Path(home_path).resolve())
masked = text
for value in _SECRET_SOURCE_VALUES_BY_HOME.get(home_key, {}).values():
if value and value in masked:
masked = masked.replace(value, "***")
return mask_known_secret_values(masked)


def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict) -> str:
Expand Down
152 changes: 152 additions & 0 deletions tests/test_env_loader_secret_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,158 @@ def test_apply_external_secret_sources_status_line_suppresses_secret_names(
assert "LEAK_THIS_TOKEN" not in err


def test_status_warning_with_secret_value_is_masked(tmp_path, monkeypatch, capsys):
"""A source warning that echoes a secret VALUE must be masked in stderr —
the merged name-suppression fix covers names, but values are the
exfiltration risk."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
monkeypatch.delenv("LEAK_THIS_API_KEY", raising=False)
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" bitwarden:\n"
" enabled: true\n"
" project_id: test-project\n"
" access_token_env: BWS_ACCESS_TOKEN\n",
encoding="utf-8",
)

import agent.secret_sources.bitwarden as bw_module

monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
monkeypatch.setattr(
bw_module,
"fetch_bitwarden_secrets",
lambda **_kw: (
{"LEAK_THIS_API_KEY": "sk-super-secret-value-123"},
["bws returned suspicious value sk-super-secret-value-123"],
),
)

from agent.secret_sources import registry as reg_module

reg_module._reset_registry_for_tests()

env_loader._apply_external_secret_sources(tmp_path)

err = capsys.readouterr().err
assert "applied 1 secret" in err
assert "sk-super-secret-value-123" not in err


def test_status_error_with_secret_value_is_masked(tmp_path, monkeypatch, capsys):
"""A source error that embeds a known secret VALUE must be masked."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
# The value is known to Hermes (applied in a prior run / .env) and the
# backend error echoes it.
monkeypatch.setenv("LEAK_THIS_API_KEY", "sk-error-value-456")
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" bitwarden:\n"
" enabled: true\n"
" project_id: test-project\n"
" access_token_env: BWS_ACCESS_TOKEN\n",
encoding="utf-8",
)

import agent.secret_sources.bitwarden as bw_module

monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))

def _raise_with_value(**_kw):
raise RuntimeError("Bitwarden rejected token sk-error-value-456")

monkeypatch.setattr(bw_module, "fetch_bitwarden_secrets", _raise_with_value)

from agent.secret_sources import registry as reg_module

reg_module._reset_registry_for_tests()

env_loader._apply_external_secret_sources(tmp_path)

err = capsys.readouterr().err
assert "Bitwarden rejected token" in err # diagnostic preserved
assert "sk-error-value-456" not in err


def test_status_warning_with_short_secret_value_is_masked(tmp_path, monkeypatch, capsys):
"""A warning echoing a SHORT applied value must be masked too.

The generic env scan in agent.redact skips values shorter than 6 chars
(``_known_secret_values``), so a short external-source ``*_TOKEN`` /
``*_PASSWORD`` echoed by a backend is the exact leak the snapshot pass
must cover — no minimum-length filter on the authoritative set.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
monkeypatch.delenv("LEAK_THIS_TOKEN", raising=False)
(tmp_path / "config.yaml").write_text(
"secrets:\n"
" bitwarden:\n"
" enabled: true\n"
" project_id: test-project\n"
" access_token_env: BWS_ACCESS_TOKEN\n",
encoding="utf-8",
)

import agent.secret_sources.bitwarden as bw_module

monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
monkeypatch.setattr(
bw_module,
"fetch_bitwarden_secrets",
lambda **_kw: (
{"LEAK_THIS_TOKEN": "ab"},
["bws echoed value ab back in the warning"],
),
)

from agent.secret_sources import registry as reg_module

reg_module._reset_registry_for_tests()

env_loader._apply_external_secret_sources(tmp_path)

err = capsys.readouterr().err
assert "bws echoed value" in err # diagnostic preserved
assert "bws echoed value ***" in err # short value masked
assert "value ab" not in err


def test_mask_secret_text_scoped_to_own_home_snapshot(tmp_path):
"""Status masking for one home must not use another home's snapshot.

Snapshots are intentionally per-home (``_SECRET_SOURCE_VALUES_BY_HOME``);
the authoritative set for a status line is the resolved home's own
values. A cross-home value in the text must survive untouched so one
profile's status output never depends on — or leaks — another profile's
secrets.
"""
home_a = tmp_path / "profile-a"
home_b = tmp_path / "profile-b"
home_a.mkdir()
home_b.mkdir()
env_loader._SECRET_SOURCE_VALUES_BY_HOME[str(home_a.resolve())] = {
"SHARED_API_KEY": "value-a"
}
env_loader._SECRET_SOURCE_VALUES_BY_HOME[str(home_b.resolve())] = {
"SHARED_API_KEY": "value-b"
}

# Home A's status line: A's own value masked, B's value left alone.
out_a = env_loader._mask_secret_text(
"warning quoting value-a but not value-b", home_a
)
assert "warning quoting *** but not value-b" == out_a

# Home B's status line: B's own value masked, A's value left alone.
out_b = env_loader._mask_secret_text(
"warning quoting value-b but not value-a", home_b
)
assert "warning quoting *** but not value-a" == out_b


def test_external_secret_values_are_isolated_between_homes(tmp_path, monkeypatch):
"""A later apply for the same key must not mutate an earlier home snapshot."""
from agent.secret_scope import build_profile_secret_scope
Expand Down
Loading