feat(secrets): add command secret source + unified secrets.provider selector

agent/secret_sources/command.py

@@ -0,0 +1,393 @@
+"""``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 <command>``.
+* 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
+        value = unquote_dotenv_value(m.group(2))
+        # Whitespace-only entries (e.g. a quoted `K="  "` placeholder) are
+        # "no value" — get_command_secret()/parse_secret_output already
+        # resolve them to None, so the list/enumerate path must omit them too
+        # or list_command_secrets() and get_command_secret() disagree on
+        # whether a quoted-blank vault entry counts as a configured key.
+        if value.strip() == "":
+            continue
+        out[m.group(1)] = value
+    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