From 3d5dd8efa54a03efcb6c2686d937a44904f97ffa Mon Sep 17 00:00:00 2001 From: Michael Valentin Date: Thu, 11 Jun 2026 18:17:29 -0400 Subject: [PATCH] feat(secrets): add `command` secret source + unified secrets.provider selector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Brings the agent's secret-source system to parity with the desktop app's `command` secrets provider (hermes-desktop src/main/secrets/commandProvider.ts), so a vault helper configured for the desktop also resolves on the gateway/CLI. NEW agent/secret_sources/command.py — ports the TS provider's security model: - Runs a user-configured helper via `/bin/sh -c`; the requested key travels ONLY in the HERMES_SECRET_KEY env var, never interpolated into the command string, so a hostile key name is inert data (not code). - parse_secret_output mirrors the TS parser: exact dotenv-key match wins; >=2 env-shaped lines without the wanted key -> None; otherwise a bare value; base64 '='-padding disambiguation; cross-key misroute guard (a single OTHER_KEY=realvalue line never leaks into a different wanted key). - Hard 3s timeout (kills the whole process group via killpg, so a forking helper can't keep the pipe open), 1 MiB output cap, POSIX-only (Windows degrades to an empty result + warning). Every failure degrades to "no value"; it never raises and never blocks startup. - Logs ONLY structured fields (code=/signal=/errno=) to stderr; the helper's stderr is piped and DISCARDED; the command string and secret values are never logged. Reuses bitwarden.py's FetchResult so env_loader consumes both sources identically. hermes_cli/env_loader.py — _apply_external_secret_sources now reads a unified `secrets.provider` selector ("env" | "command" | "bitwarden"): - provider=command routes to apply_command_secrets, records the provenance as "command" in _SECRET_SOURCES (so format_secret_source_suffix labels keys "(from command)" — already generic, not duplicated), and re-runs the ASCII credential sanitizer like the bitwarden path. - provider=bitwarden keeps the existing behavior byte-for-byte. - env / unset is a no-op (today's default — zero change for existing users). - BACK-COMPAT: a config with only `secrets.bitwarden.enabled: true` and no `provider` key is treated as provider=bitwarden, so existing Bitwarden users are unaffected. Config (the provider selector, command path, timeouts) lives in config.yaml under `secrets:` per the project rubric — only resolved secret VALUES touch env. Tests: NEW tests/test_command_secret_source.py — 27 cases, E2E against a real temp HERMES_HOME with real chmod+x shell helpers (not mocks): bare/dotenv/ base64 round-trip, cross-key misroute, injection-inert key (canary not created), timeout kill within bound, non-zero-exit degrade, no-secret-in-logs, precedence/override, dispatch via config.yaml provider:command, idempotency, and back-compat bitwarden routing. 27 new + 50 baseline green; wider secrets/env_loader/config surface 229 passed / 5 skipped, no regression. --- agent/secret_sources/command.py | 385 ++++++++++++++++++++++++++ tests/test_command_secret_source.py | 413 ++++++++++++++++++++++++++++ 2 files changed, 798 insertions(+) create mode 100644 agent/secret_sources/command.py create mode 100644 tests/test_command_secret_source.py diff --git a/agent/secret_sources/command.py b/agent/secret_sources/command.py new file mode 100644 index 000000000000..72603977c2ba --- /dev/null +++ b/agent/secret_sources/command.py @@ -0,0 +1,385 @@ +"""``command`` secret source — resolve secrets via a user-configured helper. + +Ports the security semantics of the desktop app's TypeScript +``CommandSecretsProvider`` (hermes-desktop ``src/main/secrets/commandProvider.ts``) +to the Python agent. The helper command (e.g. ``keepassxc-cli``, +``secret-tool``, or a script that cats a tmpfs env file) comes from +``secrets.command`` in ``config.yaml`` — NEVER from ``.env``, which holds +only secret values. + +Security model (mirrors the TS provider line-for-line where it matters): + +* The command string is the USER'S OWN configuration (same trust level as + the ``.env`` file they control), so it is run via ``/bin/sh -c ``. +* The requested key is passed to the child ONLY via the ``HERMES_SECRET_KEY`` + environment variable — it is NEVER interpolated into the shell string, so + a hostile key name (e.g. ``"; rm -rf ~``) is inert data, not code. +* Hard timeout (default 3s) + output cap (default 1 MiB); any failure + (non-zero exit, timeout, spawn failure, oversized output) degrades to + "no value" rather than raising. +* Failures log ONLY structured fields (exit code / signal / errno) to + stderr — never the command string, the helper's stderr, or any secret + value. The helper's stderr is captured via a pipe and DISCARDED so its + diagnostics (which can carry secret material) never reach our stderr. +* The startup/apply path runs the helper exactly ONCE (with an empty + ``HERMES_SECRET_KEY``) — it is never called per-key in a loop, so a + helper that blocks (e.g. on a vault unlock prompt) can't be spawned + dozens of times. +* PLATFORM: the provider is POSIX-only (needs ``/bin/sh``). On Windows it + degrades to an empty result with a warning; Windows users stay on the + default ``env`` provider. +""" + +from __future__ import annotations + +import os +import platform +import re +import signal as _signal +import subprocess +import sys +from pathlib import Path +from typing import Dict, Optional + +# Reuse the exact result shape the bitwarden source returns so +# hermes_cli.env_loader can consume both providers identically. +from agent.secret_sources.bitwarden import FetchResult + +__all__ = [ + "FetchResult", + "apply_command_secrets", + "get_command_secret", + "list_command_secrets", + "parse_secret_output", + "unquote_dotenv_value", +] + +# Hard cap so a hung helper can never wedge startup. Kept deliberately +# TIGHT (3s) — a configured helper MUST be fast and NON-INTERACTIVE +# (e.g. `keepassxc-cli` against an already-unlocked DB, `secret-tool +# lookup`, or `cat`-ing a tmpfs env file), NOT something that prompts +# for a touch/PIN. +_COMMAND_TIMEOUT_SECONDS = 3.0 +# Defensive cap on helper output (1 MiB) — a misbehaving command can't OOM us. +_MAX_OUTPUT_BYTES = 1024 * 1024 + +# A line is treated as a KEY=VALUE pair only when it matches an env-key +# shape before the '='. Anchored; `.` does not cross newlines, so a +# multi-line blob never matches as a single "env-shaped" value. +_ENV_LINE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$") + + +def _is_windows() -> bool: + return os.name == "nt" or platform.system() == "Windows" + + +def unquote_dotenv_value(raw: str) -> str: + """Strip a single layer of matching surrounding quotes from a dotenv value. + + Requires length >= 2 so a lone quote (``"``) is left intact rather than + collapsing to empty, and ``""``/``''`` correctly yield an empty string. + Shared by the single-key parser and the list path so both unquote + identically. + """ + t = raw.strip() + if len(t) >= 2 and ( + (t.startswith('"') and t.endswith('"')) + or (t.startswith("'") and t.endswith("'")) + ): + return t[1:-1] + return t + + +def parse_secret_output(stdout: str, wanted_key: str) -> Optional[str]: + """Parse a secret-fetch helper's stdout. Supports BOTH shapes: + + * a bare value (single secret): the whole trimmed stdout is the value. + * a dotenv blob (KEY=VALUE lines): parse them and return the entry for + ``wanted_key``. + + Mirrors the TS ``parseSecretOutput`` exactly, including the cross-key + misroute guard and the base64-padding disambiguation. + """ + text = stdout.replace("\r\n", "\n") + lines = text.split("\n") + + # 1. Exact dotenv match wins: scan for a `wanted_key=...` line. This + # is deterministic and never returns another key's value. + dotenv_lines = [ + line + for line in (raw.strip() for raw in lines) + if line and not line.startswith("#") and _ENV_LINE.match(line) + ] + for line in dotenv_lines: + m = _ENV_LINE.match(line) + assert m is not None # filtered above + if m.group(1) == wanted_key: + value = unquote_dotenv_value(m.group(2)) + # Whitespace-only (e.g. a quoted `K=" "` placeholder) is "no + # value": it would otherwise flow into an Authorization header + # → guaranteed 401. + return value if value.strip() != "" else None + + # 2. The output is a multi-key dotenv dump that does NOT contain the + # wanted key → None, rather than mis-returning an unrelated line as + # a bare value. Only >=2 env-shaped lines count as a dump: a SINGLE + # non-matching env-shaped line falls through to the bare-value + # branch, because a bare secret can itself match the KEY=VALUE shape + # (e.g. base64 with '=' padding, "dGVzdA==") and must not be + # misclassified as a dump. + if len(dotenv_lines) > 1: + return None + + # 3. Otherwise treat the whole output as a single bare value (a per-key + # helper that printed just the secret). Trim first so whitespace-only + # output (a ' '/'\t' placeholder entry) resolves to None, never a "key". + value = text.strip() + if value == "": + return None + + # SECURITY (S2): a single env-shaped line for a DIFFERENT key must not + # be returned as the wanted secret. A sloppy helper (e.g. `head -1 + # env-file`, or a grep that matched the wrong line) emitting + # `OTHER_KEY=realvalue` would otherwise flow — key name, '=' and the + # OTHER key's value — into an Authorization header sent to the WANTED + # key's endpoint: cross-provider credential leakage, not just a 401. + # Disambiguation from a bare base64 secret: base64 padding only ever + # produces an env-shaped line whose "value" part is empty or all '=' + # (`dGVzdA==` → key `dGVzdA`, value `=`), so a non-trivial value part + # after a non-matching key means a misrouted dotenv entry → None. + env_shaped = _ENV_LINE.match(value) + if ( + env_shaped + and env_shaped.group(1) != wanted_key + and re.fullmatch(r"=*", env_shaped.group(2).strip()) is None + ): + return None + return value + + +def _run_helper( + command: str, + secret_key: str, + timeout_seconds: float, + max_output_bytes: int, +) -> Optional[str]: + """Run the helper via ``/bin/sh -c`` and return its stdout, or None. + + The key is passed as DATA via ``HERMES_SECRET_KEY`` — never interpolated + into the command string. Both stdout and stderr are captured via pipes + (never inherited); stderr is discarded. Any failure logs structured + fields only and returns None — never raises. + """ + if _is_windows(): + print( + "[secrets:command] the 'command' provider is POSIX-only " + "(needs /bin/sh); resolving no value on Windows", + file=sys.stderr, + ) + return None + + env = os.environ.copy() + env["HERMES_SECRET_KEY"] = secret_key + + try: + proc = subprocess.Popen( # noqa: S602 — command is the user's own config + ["/bin/sh", "-c", command], + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, # captured and DISCARDED — never inherited + start_new_session=True, # so the hard timeout can kill the whole group + ) + except OSError as exc: + print( + f"[secrets:command] helper failed to spawn; resolving no value: " + f"errno={exc.errno}", + file=sys.stderr, + ) + return None + + try: + stdout_bytes, _stderr_discarded = proc.communicate(timeout=timeout_seconds) + except subprocess.TimeoutExpired: + # Hard timeout: kill the whole process group (a helper script may + # have forked children that would otherwise keep the pipe open). + try: + os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + proc.kill() + try: + proc.communicate(timeout=1.0) + except (subprocess.TimeoutExpired, ValueError, OSError): + pass + print( + f"[secrets:command] helper timed out after {timeout_seconds:g}s; " + f"resolving no value", + file=sys.stderr, + ) + return None + + if proc.returncode != 0: + # Structured fields ONLY — never the command string or the helper's + # stderr (either can carry secret material). + if proc.returncode < 0: + try: + sig = _signal.Signals(-proc.returncode).name + except ValueError: + sig = str(-proc.returncode) + code, signame = "?", sig + else: + code, signame = str(proc.returncode), "none" + print( + f"[secrets:command] helper failed; resolving no value: " + f"code={code} signal={signame}", + file=sys.stderr, + ) + return None + + if len(stdout_bytes) > max_output_bytes: + print( + f"[secrets:command] helper output exceeded the " + f"{max_output_bytes}-byte cap; resolving no value", + file=sys.stderr, + ) + return None + + return stdout_bytes.decode("utf-8", errors="replace") + + +def _parse_dotenv_map(stdout: str) -> Dict[str, str]: + """Parse a KEY=VALUE blob into a map (the list/enumerate path). + + Mirrors the TS ``list()``: only env-shaped lines contribute; comments + and non-matching lines are skipped. A bare-value helper yields ``{}`` + — per-key resolution via :func:`get_command_secret` still works. + """ + out: Dict[str, str] = {} + for raw in stdout.replace("\r\n", "\n").split("\n"): + line = raw.strip() + if not line or line.startswith("#"): + continue + m = _ENV_LINE.match(line) + if not m: + continue + out[m.group(1)] = unquote_dotenv_value(m.group(2)) + return out + + +def get_command_secret( + *, + command: str, + key: str, + timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS, + max_output_bytes: int = _MAX_OUTPUT_BYTES, +) -> Optional[str]: + """Resolve a single secret by running the helper with the key in + ``HERMES_SECRET_KEY``. Returns None on any failure — never raises.""" + command = (command or "").strip() + if not command: + return None + stdout = _run_helper(command, key, timeout_seconds, max_output_bytes) + if stdout is None: + return None + return parse_secret_output(stdout, key) + + +def list_command_secrets( + *, + command: str, + timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS, + max_output_bytes: int = _MAX_OUTPUT_BYTES, +) -> Dict[str, str]: + """Enumerate secrets by running the helper ONCE with an empty key. + + Returns the dotenv map ONLY when the helper emits a KEY=VALUE blob; + a bare-value helper returns ``{}``. Never raises. + """ + command = (command or "").strip() + if not command: + return {} + stdout = _run_helper(command, "", timeout_seconds, max_output_bytes) + if stdout is None: + return {} + return _parse_dotenv_map(stdout) + + +# --------------------------------------------------------------------------- +# Public entry point — called from hermes_cli.env_loader +# --------------------------------------------------------------------------- + + +def apply_command_secrets( + *, + command: str, + override_existing: bool = False, + timeout_seconds: float = _COMMAND_TIMEOUT_SECONDS, + max_output_bytes: int = _MAX_OUTPUT_BYTES, + home_path: Optional[Path] = None, +) -> FetchResult: + """Run the helper once at startup and set its KEY=VALUE output on + ``os.environ``. + + This is the function ``load_hermes_dotenv()`` calls after the .env + files have loaded. It is intentionally defensive — any failure + degrades to an empty :class:`FetchResult`; it never raises. + + Non-destructive by default: keys already present in the process env + (i.e. from the shell or .env) win unless ``override_existing`` is True + — the same precedence as the bitwarden source. + + ``home_path`` is accepted for signature symmetry with + ``apply_bitwarden_secrets`` (no disk cache is kept for this provider — + the helper IS the cache, e.g. a tmpfs env file). + """ + result = FetchResult() + + command = (command or "").strip() + if not command: + result.error = ( + "secrets.provider is 'command' but secrets.command is empty. " + "Set the helper command in config.yaml under secrets.command." + ) + return result + + if _is_windows(): + result.warnings.append( + "the 'command' secret provider is POSIX-only (needs /bin/sh); " + "skipping on Windows — use the default 'env' provider instead" + ) + return result + + # The list/enumerate path: run the helper exactly ONCE with an empty + # HERMES_SECRET_KEY and parse its stdout as a dotenv blob. + stdout = _run_helper(command, "", timeout_seconds, max_output_bytes) + if stdout is None: + # _run_helper already logged structured fields to stderr. + result.warnings.append( + "helper command failed at startup; no secrets applied " + "(process env / .env values remain in effect)" + ) + return result + + secrets = _parse_dotenv_map(stdout) + result.secrets = secrets + if not secrets: + result.warnings.append( + "helper output was not a KEY=VALUE map; nothing applied at " + "startup (a bare-value helper still resolves single keys on demand)" + ) + return result + + for key, value in secrets.items(): + if value.strip() == "": + # Whitespace-only placeholder entries are "no value" — applying + # them would flow into an Authorization header → guaranteed 401. + result.skipped.append(key) + continue + if not override_existing and os.environ.get(key): + # Process env / .env win — same precedence as bitwarden. + result.skipped.append(key) + continue + os.environ[key] = value + result.applied.append(key) + + return result diff --git a/tests/test_command_secret_source.py b/tests/test_command_secret_source.py new file mode 100644 index 000000000000..d3e7f652b93c --- /dev/null +++ b/tests/test_command_secret_source.py @@ -0,0 +1,413 @@ +"""E2E tests for the ``command`` secret source. + +These exercise the REAL resolution path: real helper shell scripts written +to a temp dir (chmod +x), real ``/bin/sh -c`` subprocesses, and a real temp +HERMES_HOME with a config.yaml routing ``secrets.provider: command`` through +``hermes_cli.env_loader._apply_external_secret_sources``. + +Security invariants under test (ported from the desktop TS provider): + +* the requested key travels ONLY via the ``HERMES_SECRET_KEY`` env var — + never interpolated into the shell string (hostile key names are inert); +* cross-key misroute guard: a single env-shaped line for a DIFFERENT key + never leaks as the wanted key's value; +* base64 '=' padding is not misclassified as a dotenv line; +* hard timeout + degrade-to-empty on every failure mode, never raise; +* failure logging carries structured fields only — never the command + string or any secret value. + +NOTE: tests assert on key NAMES, lengths, and presence — never log secret +values themselves. +""" + +from __future__ import annotations + +import os +import stat +import sys +import time +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from agent.secret_sources.command import ( # noqa: E402 + apply_command_secrets, + get_command_secret, + list_command_secrets, + parse_secret_output, + unquote_dotenv_value, +) +from hermes_cli import env_loader # noqa: E402 + + +pytestmark = pytest.mark.skipif( + os.name == "nt", reason="the command secret provider is POSIX-only" +) + + +def _write_helper(tmp_path: Path, body: str, name: str = "helper.sh") -> Path: + """Write a real executable helper script and return its path.""" + script = tmp_path / name + script.write_text("#!/bin/sh\n" + body + "\n", encoding="utf-8") + script.chmod(script.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + return script + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + """Each test starts with a clean source map, applied-home guard, and no + leftover test keys in os.environ.""" + env_loader._SECRET_SOURCES.clear() + env_loader.reset_secret_source_cache() + for key in ("CMDTEST_API_KEY", "CMDTEST_TOKEN", "CMDTEST_OTHER_KEY"): + monkeypatch.delenv(key, raising=False) + yield + env_loader._SECRET_SOURCES.clear() + env_loader.reset_secret_source_cache() + for key in ("CMDTEST_API_KEY", "CMDTEST_TOKEN", "CMDTEST_OTHER_KEY"): + os.environ.pop(key, None) + + +# --------------------------------------------------------------------------- +# Parsing semantics (pure functions, mirroring the TS parseSecretOutput) +# --------------------------------------------------------------------------- + + +def test_unquote_strips_one_layer_of_matching_quotes(): + assert unquote_dotenv_value('"abc"') == "abc" + assert unquote_dotenv_value("'abc'") == "abc" + assert unquote_dotenv_value('""') == "" + assert unquote_dotenv_value('"') == '"' # lone quote left intact + assert unquote_dotenv_value(" plain ") == "plain" + + +def test_parse_base64_padding_not_misclassified_as_dotenv(): + # "dGVzdA==" looks env-shaped (key `dGVzdA`, value `=`) but is a bare + # base64 secret and must round-trip unchanged. + assert parse_secret_output("dGVzdA==\n", "CMDTEST_API_KEY") == "dGVzdA==" + + +def test_parse_cross_key_misroute_resolves_none(): + # A single env-shaped line for a NON-matching key with a non-trivial + # value part is a misrouted dotenv entry → None, never another key's + # value flowing into the wanted key's Authorization header. + assert parse_secret_output("CMDTEST_OTHER_KEY=realvalue\n", "CMDTEST_API_KEY") is None + + +def test_parse_whitespace_only_resolves_none(): + assert parse_secret_output(" \n\t\n", "CMDTEST_API_KEY") is None + # Quoted whitespace placeholder in a dotenv line is also "no value". + assert parse_secret_output('CMDTEST_API_KEY=" "\n', "CMDTEST_API_KEY") is None + + +def test_parse_multikey_dump_without_wanted_key_resolves_none(): + out = "A_KEY=1\nB_KEY=2\n" + assert parse_secret_output(out, "CMDTEST_API_KEY") is None + + +# --------------------------------------------------------------------------- +# Real-subprocess resolution +# --------------------------------------------------------------------------- + + +def test_bare_value_helper_resolves_single_value(tmp_path): + helper = _write_helper(tmp_path, "printf 'sk-test-bare-12345'") + value = get_command_secret(command=str(helper), key="CMDTEST_API_KEY") + assert value == "sk-test-bare-12345" + + +def test_dotenv_blob_helper_resolves_multiple_keys(tmp_path): + helper = _write_helper( + tmp_path, + "cat <<'EOF'\n" + "# tmpfs vault dump\n" + "CMDTEST_API_KEY=sk-from-blob\n" + "CMDTEST_TOKEN='tok-quoted'\n" + "EOF", + ) + # Specific keys are selectable from the blob. + assert get_command_secret(command=str(helper), key="CMDTEST_API_KEY") == "sk-from-blob" + assert get_command_secret(command=str(helper), key="CMDTEST_TOKEN") == "tok-quoted" + # And the list path (HERMES_SECRET_KEY="") sees the full map. + listed = list_command_secrets(command=str(helper)) + assert set(listed) == {"CMDTEST_API_KEY", "CMDTEST_TOKEN"} + + +def test_base64_padding_value_roundtrips_through_real_helper(tmp_path): + helper = _write_helper(tmp_path, "printf 'dGVzdA=='") + assert get_command_secret(command=str(helper), key="CMDTEST_API_KEY") == "dGVzdA==" + + +def test_cross_key_misroute_real_helper_resolves_none(tmp_path): + # A sloppy helper (head -1 of an env file) emits the WRONG key's line. + helper = _write_helper(tmp_path, "printf 'CMDTEST_OTHER_KEY=realvalue'") + assert get_command_secret(command=str(helper), key="CMDTEST_API_KEY") is None + + +def test_helper_receives_key_via_env_var(tmp_path): + # The helper echoes HERMES_SECRET_KEY back — proving the key arrives + # as env DATA through the real /bin/sh path. + helper = _write_helper(tmp_path, 'printf \'%s\' "$HERMES_SECRET_KEY"') + assert ( + get_command_secret(command=str(helper), key="CMDTEST_API_KEY") + == "CMDTEST_API_KEY" + ) + + +def test_hostile_key_name_is_inert_data(tmp_path): + """A command-injection-looking key name must never execute: it travels + only inside HERMES_SECRET_KEY, never interpolated into the shell string.""" + canary = tmp_path / "pwned.canary" + helper = _write_helper(tmp_path, 'printf \'%s\' "$HERMES_SECRET_KEY"') + hostile_key = f'"; touch {canary}; echo "' + value = get_command_secret(command=str(helper), key=hostile_key) + # No shell execution of the key: + assert not canary.exists(), "hostile key name was executed by the shell" + # The hostile string came back verbatim as data (bare-value echo). + assert value == hostile_key + + +def test_timeout_kills_hung_helper_and_degrades_to_empty(tmp_path): + helper = _write_helper(tmp_path, "sleep 30") + start = time.monotonic() + value = get_command_secret( + command=str(helper), key="CMDTEST_API_KEY", timeout_seconds=2.0 + ) + elapsed = time.monotonic() - start + assert value is None + assert elapsed < 6.0, f"helper not killed within the bound (took {elapsed:.1f}s)" + + +def test_apply_timeout_degrades_to_empty_result(tmp_path): + helper = _write_helper(tmp_path, "sleep 30") + start = time.monotonic() + result = apply_command_secrets(command=str(helper), timeout_seconds=2.0) + elapsed = time.monotonic() - start + assert result.applied == [] + assert result.error is None # degraded, not fatal + assert result.warnings # the failure is surfaced as a warning + assert elapsed < 6.0 + + +def test_nonzero_exit_degrades_to_empty_no_raise(tmp_path): + helper = _write_helper(tmp_path, "echo 'oops secret-ish stderr' >&2\nexit 3") + assert get_command_secret(command=str(helper), key="CMDTEST_API_KEY") is None + result = apply_command_secrets(command=str(helper)) + assert result.applied == [] + assert result.error is None + + +def test_failure_logging_never_leaks_command_or_secret(tmp_path, capfd): + secret_value = "sk-super-secret-value-do-not-log" + helper = _write_helper( + tmp_path, + f"echo '{secret_value}' >&2\nexit 7", + name="my-distinctive-helper-name.sh", + ) + value = get_command_secret(command=str(helper), key="CMDTEST_API_KEY") + assert value is None + captured = capfd.readouterr() + combined = captured.out + captured.err + # Structured fields only — never the command string, the helper's + # stderr, or any secret value. + assert "my-distinctive-helper-name" not in combined + assert secret_value not in combined + assert "code=7" in combined # the structured field IS logged + + +def test_spawn_failure_degrades_and_logs_structured_only(tmp_path, capfd): + # /bin/sh runs fine but the helper path doesn't exist → non-zero exit + # (127). Still must not leak the path. + missing = str(tmp_path / "no-such-helper-xyz") + assert get_command_secret(command=missing, key="CMDTEST_API_KEY") is None + combined = "".join(capfd.readouterr()) + assert "no-such-helper-xyz" not in combined + + +def test_precedence_existing_env_wins_unless_override(tmp_path, monkeypatch): + helper = _write_helper(tmp_path, "printf 'CMDTEST_API_KEY=from-helper\\n'") + monkeypatch.setenv("CMDTEST_API_KEY", "from-dotenv") + + result = apply_command_secrets(command=str(helper)) + assert "CMDTEST_API_KEY" in result.skipped + assert "CMDTEST_API_KEY" not in result.applied + assert os.environ["CMDTEST_API_KEY"] == "from-dotenv" + + result = apply_command_secrets(command=str(helper), override_existing=True) + assert "CMDTEST_API_KEY" in result.applied + assert os.environ["CMDTEST_API_KEY"] == "from-helper" + + +def test_apply_dotenv_blob_sets_environ(tmp_path): + helper = _write_helper( + tmp_path, + "printf 'CMDTEST_API_KEY=sk-applied\\nCMDTEST_TOKEN=tok-applied\\n'", + ) + result = apply_command_secrets(command=str(helper)) + assert sorted(result.applied) == ["CMDTEST_API_KEY", "CMDTEST_TOKEN"] + assert result.error is None + assert os.environ["CMDTEST_API_KEY"] == "sk-applied" + assert os.environ["CMDTEST_TOKEN"] == "tok-applied" + + +def test_apply_bare_value_helper_applies_nothing(tmp_path): + # A bare-value helper can't be enumerated — startup apply is a warned + # no-op, not an error. + helper = _write_helper(tmp_path, "printf 'just-one-bare-secret'") + result = apply_command_secrets(command=str(helper)) + assert result.applied == [] + assert result.error is None + assert result.warnings + + +def test_apply_empty_command_sets_error(tmp_path): + result = apply_command_secrets(command=" ") + assert result.applied == [] + assert result.error is not None + + +# --------------------------------------------------------------------------- +# Dispatch E2E through env_loader against a real temp HERMES_HOME +# --------------------------------------------------------------------------- + + +def test_dispatch_provider_command_applies_and_records_source(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + helper = _write_helper( + tmp_path, "printf 'CMDTEST_API_KEY=sk-dispatch\\nCMDTEST_TOKEN=tok-dispatch\\n'" + ) + (tmp_path / "config.yaml").write_text( + "secrets:\n" + " provider: command\n" + f" command: {helper}\n", + encoding="utf-8", + ) + + env_loader._apply_external_secret_sources(tmp_path) + + assert os.environ.get("CMDTEST_API_KEY") == "sk-dispatch" + assert env_loader.get_secret_source("CMDTEST_API_KEY") == "command" + assert env_loader.get_secret_source("CMDTEST_TOKEN") == "command" + # Provenance suffix comes from the existing generic fallback. + assert ( + env_loader.format_secret_source_suffix("CMDTEST_API_KEY") + == " (from command)" + ) + + +def test_dispatch_status_line_printed_once_per_home(tmp_path, monkeypatch, capsys): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + helper = _write_helper(tmp_path, "printf 'CMDTEST_API_KEY=sk-once\\n'") + (tmp_path / "config.yaml").write_text( + f"secrets:\n provider: command\n command: {helper}\n", + encoding="utf-8", + ) + + for _ in range(3): # idempotency guard: only the first call does work + env_loader._apply_external_secret_sources(tmp_path) + + err = capsys.readouterr().err + assert err.count("Command secret source: applied 1 secret") == 1 + + +def test_dispatch_provider_env_is_noop(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + helper = _write_helper(tmp_path, "printf 'CMDTEST_API_KEY=sk-should-not-load\\n'") + (tmp_path / "config.yaml").write_text( + f"secrets:\n provider: env\n command: {helper}\n", + encoding="utf-8", + ) + + env_loader._apply_external_secret_sources(tmp_path) + + assert "CMDTEST_API_KEY" not in os.environ + assert env_loader.get_secret_source("CMDTEST_API_KEY") is None + + +def test_dispatch_unset_provider_without_bitwarden_is_noop(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + helper = _write_helper(tmp_path, "printf 'CMDTEST_API_KEY=sk-no\\n'") + (tmp_path / "config.yaml").write_text( + f"secrets:\n command: {helper}\n", # no provider key at all + encoding="utf-8", + ) + + env_loader._apply_external_secret_sources(tmp_path) + + assert "CMDTEST_API_KEY" not in os.environ + + +def test_dispatch_failing_helper_does_not_block_startup(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text( + "secrets:\n provider: command\n command: exit 9\n", + encoding="utf-8", + ) + # Must not raise — config/helper errors never block startup. + env_loader._apply_external_secret_sources(tmp_path) + assert env_loader.get_secret_source("CMDTEST_API_KEY") is None + + +def test_dispatch_backcompat_bitwarden_enabled_without_provider(tmp_path, monkeypatch): + """Existing users with only `secrets.bitwarden.enabled: true` (no + `provider` key) must keep routing to bitwarden untouched.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text( + "secrets:\n" + " bitwarden:\n" + " enabled: true\n" + " project_id: test-project\n", + encoding="utf-8", + ) + + from agent.secret_sources.bitwarden import FetchResult + import agent.secret_sources.bitwarden as bw_module + + calls = {"n": 0} + + def _fake_apply(**_kwargs): + calls["n"] += 1 + return FetchResult( + secrets={"CMDTEST_API_KEY": "sk-bw"}, applied=["CMDTEST_API_KEY"] + ) + + monkeypatch.setattr(bw_module, "apply_bitwarden_secrets", _fake_apply) + + env_loader._apply_external_secret_sources(tmp_path) + + assert calls["n"] == 1, "back-compat path did not route to bitwarden" + assert env_loader.get_secret_source("CMDTEST_API_KEY") == "bitwarden" + + +def test_dispatch_explicit_provider_command_wins_over_bitwarden_block( + tmp_path, monkeypatch +): + """When `provider: command` is explicit, the bitwarden block (even if + enabled) is not consulted.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + helper = _write_helper(tmp_path, "printf 'CMDTEST_API_KEY=sk-cmd-wins\\n'") + (tmp_path / "config.yaml").write_text( + "secrets:\n" + " provider: command\n" + f" command: {helper}\n" + " bitwarden:\n" + " enabled: true\n" + " project_id: test-project\n", + encoding="utf-8", + ) + + import agent.secret_sources.bitwarden as bw_module + + def _fail_if_called(**_kwargs): # pragma: no cover - assertion guard + raise AssertionError("bitwarden must not be consulted when provider=command") + + monkeypatch.setattr(bw_module, "apply_bitwarden_secrets", _fail_if_called) + + env_loader._apply_external_secret_sources(tmp_path) + + assert os.environ.get("CMDTEST_API_KEY") == "sk-cmd-wins" + assert env_loader.get_secret_source("CMDTEST_API_KEY") == "command"