diff --git a/agent/secret_sources/__init__.py b/agent/secret_sources/__init__.py index e1564058ad11..28330b5fc21c 100644 --- a/agent/secret_sources/__init__.py +++ b/agent/secret_sources/__init__.py @@ -10,4 +10,6 @@ - ``bitwarden`` — Bitwarden Secrets Manager (`bws` CLI). See ``agent.secret_sources.bitwarden`` for the integration and ``hermes_cli.secrets_cli`` for the user-facing setup wizard. + - ``protonpass`` — Proton Pass ``pass://`` secret references (`pass-cli`). + See ``agent.secret_sources.protonpass`` for the integration. """ diff --git a/agent/secret_sources/protonpass.py b/agent/secret_sources/protonpass.py new file mode 100644 index 000000000000..df601d59bf1c --- /dev/null +++ b/agent/secret_sources/protonpass.py @@ -0,0 +1,737 @@ +"""Proton Pass (`pass-cli`) secret source. + +Resolve provider credentials from Proton Pass ``pass://vault/item/field`` +references at process startup so they don't have to live in plaintext in +``~/.hermes/.env``. + +Design summary +-------------- + +* Users map environment-variable names to Proton Pass secret references in + ``secrets.protonpass.env``:: + + secrets: + protonpass: + enabled: true + env: + OPENAI_API_KEY: "pass://Private/OpenAI/api key" + ANTHROPIC_API_KEY: "pass://Private/Anthropic/credential" + +* After ``.env`` loads, each reference is resolved and injected into + ``os.environ`` (the same point in startup as the Bitwarden / 1Password + sources). Resolution shells out to ``pass-cli run --no-masking -- printenv + `` with the reference placed in the child's environment: ``run`` + substitutes the ``pass://`` URI for the real value before exec'ing the + command, and ``--no-masking`` keeps that value from being replaced with + ```` on stdout. This uses only documented + behaviour and sidesteps the decorated/uncertain output of ``item view``. +* Authentication uses Proton Pass's persistent session model. Unlike the + 1Password CLI's per-invocation service-account token, ``pass-cli`` logs in + once (``pass-cli login``, which reads ``PROTON_PASS_PERSONAL_ACCESS_TOKEN``) + and stores a session in a platform-specific keyring. We rely on an existing + session when one is present and only attempt a login — using the configured + personal-access-token env var — when a resolve fails for an auth-shaped + reason. Hermes never downloads ``pass-cli``. +* Failures NEVER block startup. A missing ``pass-cli`` binary, a login + failure, a bad reference, or an empty value each surface a one-line warning + and Hermes continues with whatever credentials ``.env`` already had. + +The atomic-write / ``0600`` / TTL cache mechanics are kept self-contained in +this module (mirroring ``bitwarden.py`` on ``main``) so the backend merges +without depending on the shared ``agent.secret_sources._cache`` substrate +introduced in the open 1Password PR (#36896); it can be rebased onto that +substrate once it lands. The disk file holds only resolved secret *values*; +auth material is fingerprinted, never stored. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import re +import shutil +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Configuration constants +# --------------------------------------------------------------------------- + +# How long to wait for a single `pass-cli` subprocess, in seconds. +_PASS_RUN_TIMEOUT = 30 + +# Default env var the `pass-cli login` command reads for non-interactive +# personal-access-token auth. Users can point `personal_access_token_env` at a +# different name; we always export the value to the child as +# PROTON_PASS_PERSONAL_ACCESS_TOKEN, which is what `pass-cli` itself looks for. +_DEFAULT_TOKEN_ENV = "PROTON_PASS_PERSONAL_ACCESS_TOKEN" + +# Env var name `pass-cli` reads its personal access token from. +_PASS_TOKEN_ENV = "PROTON_PASS_PERSONAL_ACCESS_TOKEN" + +# Internal env var we set the pass:// reference under before invoking +# `pass-cli run -- -c `. `run` resolves the reference +# found in this variable and the wrapped command echoes the resolved value +# back to us. We wrap the *current* Python interpreter rather than a shell +# builtin like `printenv` (POSIX-only) so resolution works identically on +# Windows, macOS, and Linux. +_RESOLVE_SENTINEL = "HERMES_PASS_CLI_RESOLVE" + +# Echo script run under the current interpreter: write the (now-resolved) +# sentinel value to stdout verbatim, with no added newline so secrets with +# meaningful trailing whitespace survive. +_ECHO_SCRIPT = ( + "import os,sys;sys.stdout.write(os.environ.get(%r,''))" % _RESOLVE_SENTINEL +) + +# `run` masks resolved secrets in child stdout/stderr by default; we pass +# --no-masking to read the value, so this marker should never appear, but we +# treat it as an empty/failed resolve defensively. +_CONCEALED_MARKER = "" + +# Strip whole ANSI CSI sequences (colour, cursor moves, line erases) from any +# `pass-cli` diagnostic we surface — not just the lone ESC byte — so a control +# sequence can't reposition the cursor or hide text after a redaction marker. +_ANSI_CSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") + +# stderr fragments that indicate an auth/session problem rather than a bad +# reference — these are the cases where attempting a `pass-cli login` and +# retrying is worthwhile. +_AUTH_ERROR_HINTS = ( + "not logged in", + "not signed in", + "no session", + "session expired", + "session has expired", + "unauthorized", + "unauthenticated", + "authentication", + "please log in", + "please login", + "log in first", +) + +# Env vars the `pass-cli` child actually needs. We build a minimal allowlisted +# env rather than copying all of os.environ (which, post-dotenv, holds every +# provider credential) into the child — tighter blast radius if `pass-cli` or +# anything it execs ever misbehaves. HOME / XDG_* / DBUS are needed so the +# persistent session keyring is reachable; the token is added only for login. +_PASS_ENV_ALLOWLIST = ( + "PATH", + "HOME", + "USERPROFILE", + "APPDATA", + "LOCALAPPDATA", + "SystemRoot", + "TMPDIR", + "TMP", + "TEMP", + "XDG_CONFIG_HOME", + "XDG_DATA_HOME", + "XDG_CACHE_HOME", + "XDG_RUNTIME_DIR", + "DBUS_SESSION_BUS_ADDRESS", +) + + +# --------------------------------------------------------------------------- +# Result + cache dataclasses (self-contained; mirror bitwarden.py on main) +# --------------------------------------------------------------------------- + + +@dataclass +class FetchResult: + """Outcome of a single Proton Pass pull. + + ``error`` is set only for *fatal* conditions (nothing was fetched); + non-fatal issues go into ``warnings`` so the values that did resolve are + still applied. ``ok`` is the convenience inverse of ``error``. + """ + + secrets: Dict[str, str] = field(default_factory=dict) + applied: List[str] = field(default_factory=list) # set into os.environ + skipped: List[str] = field(default_factory=list) # already set / protected + warnings: List[str] = field(default_factory=list) # non-fatal issues + error: Optional[str] = None # fatal: nothing fetched + binary_path: Optional[Path] = None + + @property + def ok(self) -> bool: + return self.error is None + + +@dataclass +class _CachedFetch: + secrets: Dict[str, str] + fetched_at: float + + def is_fresh(self, ttl_seconds: float) -> bool: + if ttl_seconds <= 0: + return False + return (time.time() - self.fetched_at) < ttl_seconds + + +def is_valid_env_name(name: str) -> bool: + """Return True if ``name`` is a usable POSIX environment-variable name. + + Must be non-empty, start with a letter or underscore, and contain only + alphanumerics and underscores. Used to drop secret names that couldn't be + exported (e.g. ``"has spaces"`` or ``"1LEADING_DIGIT"``). + """ + if not name: + return False + if not (name[0].isalpha() or name[0] == "_"): + return False + return all(c.isalnum() or c == "_" for c in name) + + +# --------------------------------------------------------------------------- +# Cache +# --------------------------------------------------------------------------- + +# In-process cache. The key folds in str(home_path) so a HERMES_HOME switch +# inside one long-lived process (e.g. the gateway) can't return another +# profile's secrets from L1. +_CacheKey = Tuple[str, str, str] # (auth_fp, home, refs_fp) +_CACHE: Dict[_CacheKey, _CachedFetch] = {} + +_DISK_CACHE_BASENAME = "protonpass_cache.json" + + +def _disk_cache_path(home_path: Optional[Path] = None) -> Path: + """Path to the on-disk cache (exposed for tests and direct callers).""" + if home_path is None: + home_path = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) + return home_path / "cache" / _DISK_CACHE_BASENAME + + +def _disk_key_str(cache_key: _CacheKey) -> str: + """Serialize a cache key for on-disk storage, omitting home_path. + + The disk file is already partitioned by home (it lives under + ``/cache/``), so the path provides the home dimension. + """ + auth_fp, _home, refs_fp = cache_key + return f"{auth_fp}|{refs_fp}" + + +def _read_disk_cache( + cache_key: _CacheKey, ttl_seconds: float, home_path: Optional[Path] = None +) -> Optional[_CachedFetch]: + """Return a fresh cached entry for ``cache_key`` from disk, or None. + + Best-effort: any I/O or parse error, a key mismatch, or a stale entry all + return None so the caller re-fetches. + """ + if ttl_seconds <= 0: + return None + path = _disk_cache_path(home_path) + try: + with open(path, "r", encoding="utf-8") as f: + payload = json.load(f) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(payload, dict): + return None + if payload.get("key") != _disk_key_str(cache_key): + return None + secrets = payload.get("secrets") + fetched_at = payload.get("fetched_at") + if not isinstance(secrets, dict) or not isinstance(fetched_at, (int, float)): + return None + typed: Dict[str, str] = { + k: v for k, v in secrets.items() if isinstance(k, str) and isinstance(v, str) + } + entry = _CachedFetch(secrets=typed, fetched_at=float(fetched_at)) + if not entry.is_fresh(ttl_seconds): + return None + return entry + + +def _write_disk_cache( + cache_key: _CacheKey, + entry: _CachedFetch, + ttl_seconds: float, + home_path: Optional[Path] = None, +) -> None: + """Persist ``entry`` for ``cache_key`` atomically at mode ``0600``. + + No-op when ``ttl_seconds <= 0`` (caching genuinely off) or on any I/O error + — the next invocation just re-fetches. + """ + if ttl_seconds <= 0: + return + path = _disk_cache_path(home_path) + try: + cache_dir = path.parent + cache_dir.mkdir(parents=True, exist_ok=True) + # mkdir's mode is umask-subject; chmod the dir to 0700 so cache + # metadata isn't exposed if HERMES_HOME is ever made traversable. + try: + os.chmod(cache_dir, 0o700) + except OSError: + pass + payload = { + "key": _disk_key_str(cache_key), + "secrets": entry.secrets, + "fetched_at": entry.fetched_at, + } + fd, tmp = tempfile.mkstemp( + prefix=".protonpass_cache_", suffix=".tmp", dir=str(cache_dir) + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(payload, f) + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + except OSError: + pass # best-effort — a disk-cache miss next invocation is fine + + +# --------------------------------------------------------------------------- +# Reference validation + fingerprinting +# --------------------------------------------------------------------------- + + +def _validate_references( + references: Optional[Dict[str, str]], +) -> Tuple[Dict[str, str], List[str]]: + """Return ``(valid_refs, warnings)`` from an ``env`` mapping. + + A reference is kept only if its target env-var name is a valid POSIX name + and the value is a stripped ``pass://…`` reference string. Everything else + produces a warning and is dropped (never fatal). + """ + valid: Dict[str, str] = {} + warnings: List[str] = [] + for name, ref in (references or {}).items(): + if not is_valid_env_name(name): + warnings.append(f"Skipping {name!r}: not a valid env-var name") + continue + if not isinstance(ref, str): + warnings.append(f"Skipping {name!r}: reference is not a string") + continue + cleaned = ref.strip() + if not cleaned.startswith("pass://"): + warnings.append( + f"Skipping {name!r}: {ref!r} is not a pass:// secret reference" + ) + continue + valid[name] = cleaned + return valid, warnings + + +def _auth_fingerprint(token_env: str) -> str: + """SHA-256 prefix over the auth material a login would use. + + Folds in the configured personal-access-token value. Rotating the token to + a different identity therefore changes the cache key, so a value cached + under a previous identity is never served under a new one. The session + itself lives in an external keyring (not in os.environ), so it can't be + fingerprinted here — the token is the stable identity proxy. Never logged + or displayed; the raw token never leaves this hash. + """ + material = f"token={os.environ.get(token_env, '')}" + return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16] + + +def _refs_fingerprint(references: Dict[str, str]) -> str: + """SHA-256 prefix over the configured name→reference mapping.""" + material = "\n".join(f"{name}={references[name]}" for name in sorted(references)) + return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16] + + +# --------------------------------------------------------------------------- +# Binary discovery +# --------------------------------------------------------------------------- + + +def find_pass_cli(binary_path: str = "") -> Optional[Path]: + """Resolve a usable ``pass-cli`` binary, or None. + + When ``binary_path`` is set it is used verbatim and PATH is NOT consulted — + pinning an absolute path is a way to avoid trusting whatever ``pass-cli`` + shows up first on ``PATH``. A pinned-but-missing path returns None (the + caller surfaces a clear error) rather than silently falling back. + """ + if binary_path: + pinned = Path(binary_path) + if pinned.exists() and os.access(pinned, os.X_OK): + return pinned + return None + found = shutil.which("pass-cli") + return Path(found) if found else None + + +# --------------------------------------------------------------------------- +# `pass-cli` invocation +# --------------------------------------------------------------------------- + + +def _scrub(text: str) -> str: + """Remove ANSI control sequences and trim, for safe message surfacing.""" + return _ANSI_CSI_RE.sub("", text).replace("\x1b", "").strip() + + +def _looks_like_auth_error(stderr: str) -> bool: + low = stderr.lower() + return any(hint in low for hint in _AUTH_ERROR_HINTS) + + +def _pass_child_env(token_value: str = "") -> Dict[str, str]: + """Build a minimal allowlisted environment for the ``pass-cli`` child. + + ``token_value`` is injected as PROTON_PASS_PERSONAL_ACCESS_TOKEN only for + login; resolve calls pass an empty token and rely on the persistent + session. + """ + env: Dict[str, str] = {} + for key in _PASS_ENV_ALLOWLIST: + val = os.environ.get(key) + if val is not None: + env[key] = val + if token_value: + env[_PASS_TOKEN_ENV] = token_value + env["NO_COLOR"] = "1" + return env + + +class _AuthError(RuntimeError): + """A resolve failed for an auth/session reason — a login retry may help.""" + + +def _pass_login(pass_cli: Path, token_value: str) -> None: + """Establish a Proton Pass session from a personal access token. + + Raises :class:`RuntimeError` on failure. Runs fully non-interactively: + the token is passed via the child environment (never argv, so it stays out + of the process list) and stdin is closed so a missing/invalid token fails + fast instead of blocking on an interactive prompt. + """ + if not token_value: + raise RuntimeError( + "no personal access token available to log in (set the env var named " + "by secrets.protonpass.personal_access_token_env)" + ) + cmd = [str(pass_cli), "login"] + try: + proc = subprocess.run( # noqa: S603 — pass-cli path is user-trusted, argv list + cmd, + env=_pass_child_env(token_value), + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=_PASS_RUN_TIMEOUT, + stdin=subprocess.DEVNULL, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + f"pass-cli login timed out after {_PASS_RUN_TIMEOUT}s" + ) from exc + except OSError as exc: + raise RuntimeError(f"failed to invoke pass-cli: {exc}") from exc + + if proc.returncode != 0: + err = _scrub(proc.stderr or "")[:200] + raise RuntimeError(err or f"pass-cli login exited {proc.returncode}") + + +def _run_pass_resolve(pass_cli: Path, reference: str) -> str: + """Resolve a single ``pass://`` reference to its value. + + Uses ``pass-cli run --no-masking -- -c `` with the + reference placed in the child env: ``run`` substitutes the resolved value + into the env var and the wrapped interpreter echoes it back. Raises + :class:`_AuthError` on an auth-shaped failure (so the caller can attempt a + login + retry) and :class:`RuntimeError` on any other failure — including a + ``returncode 0`` with empty/concealed output, which would otherwise + silently clobber a good ``.env``/shell credential. + """ + cmd = [ + str(pass_cli), + "run", + "--no-masking", + "--", + sys.executable, + "-c", + _ECHO_SCRIPT, + ] + child_env = _pass_child_env() + child_env[_RESOLVE_SENTINEL] = reference + + try: + proc = subprocess.run( # noqa: S603 — pass-cli path is user-trusted, argv list + cmd, + env=child_env, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=_PASS_RUN_TIMEOUT, + stdin=subprocess.DEVNULL, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + f"pass-cli run timed out after {_PASS_RUN_TIMEOUT}s for {reference!r}" + ) from exc + except OSError as exc: + raise RuntimeError(f"failed to invoke pass-cli: {exc}") from exc + + if proc.returncode != 0: + err = _scrub(proc.stderr or "")[:200] + msg = ( + f"pass-cli run failed for {reference!r}: {err}" + if err + else f"pass-cli run exited {proc.returncode} for {reference!r}" + ) + if _looks_like_auth_error(err): + raise _AuthError(msg) + raise RuntimeError(msg) + + # The echo script writes the value verbatim with no added newline, so we + # take stdout as-is — a value with meaningful internal/trailing whitespace + # survives intact. An empty/whitespace-only or still-concealed value is + # treated as empty: applying it would silently clobber a good credential + # with effectively nothing. + value = proc.stdout or "" + if not value.strip() or value.strip() == _CONCEALED_MARKER: + raise RuntimeError(f"pass-cli returned an empty value for {reference!r}") + return value + + +# --------------------------------------------------------------------------- +# Fetch +# --------------------------------------------------------------------------- + + +def _resolve_batch( + pass_cli: Path, references: Dict[str, str] +) -> Tuple[Dict[str, str], List[str], bool]: + """Resolve ``references`` once. Returns ``(secrets, warnings, auth_failed)``. + + ``auth_failed`` is True if any reference failed for an auth/session reason, + signalling the caller to try a login and re-run the batch. + """ + secrets: Dict[str, str] = {} + warnings: List[str] = [] + auth_failed = False + for name in sorted(references): + try: + secrets[name] = _run_pass_resolve(pass_cli, references[name]) + except _AuthError as exc: + auth_failed = True + warnings.append(str(exc)) + except RuntimeError as exc: + warnings.append(str(exc)) + return secrets, warnings, auth_failed + + +def fetch_protonpass_secrets( + *, + references: Dict[str, str], + token_env: str = _DEFAULT_TOKEN_ENV, + binary: Optional[Path] = None, + binary_path: str = "", + use_cache: bool = True, + cache_ttl_seconds: float = 300, + home_path: Optional[Path] = None, +) -> Tuple[Dict[str, str], List[str]]: + """Resolve ``references`` (name → ``pass://…``) to ``(secrets, warnings)``. + + Raises :class:`RuntimeError` only when no ``pass-cli`` binary is available — + a fatal "can't fetch anything" condition. Per-reference failures (bad + reference, empty value, persistent auth failure) are collected as warnings + and the reference is dropped, so one bad entry never sinks the rest. + + On an auth-shaped failure we attempt a single ``pass-cli login`` using the + configured token env var and re-resolve the still-missing references. + + Only a complete, error-free pull is cached, so a transient auth failure + isn't frozen in for the whole TTL window. + """ + valid, warnings = _validate_references(references) + if not valid: + return {}, warnings + + cache_key: _CacheKey = ( + _auth_fingerprint(token_env), + str(home_path) if home_path is not None else "", + _refs_fingerprint(valid), + ) + + if use_cache: + cached = _CACHE.get(cache_key) + if cached and cached.is_fresh(cache_ttl_seconds): + return dict(cached.secrets), warnings + disk_cached = _read_disk_cache(cache_key, cache_ttl_seconds, home_path) + if disk_cached is not None: + # Promote into L1 so later fetches in this process skip the disk read. + _CACHE[cache_key] = disk_cached + return dict(disk_cached.secrets), warnings + + pass_cli = binary or find_pass_cli(binary_path) + if pass_cli is None: + raise RuntimeError( + "pass-cli not found. Install the Proton Pass CLI " + "(https://protonpass.github.io/pass-cli/) or set " + "secrets.protonpass.binary_path to its absolute location." + ) + + secrets, batch_warnings, auth_failed = _resolve_batch(pass_cli, valid) + + # One login + retry of the still-missing references on an auth failure. + if auth_failed and len(secrets) < len(valid): + token_value = os.environ.get(token_env, "").strip() + try: + _pass_login(pass_cli, token_value) + remaining = {n: r for n, r in valid.items() if n not in secrets} + retry_secrets, retry_warnings, _ = _resolve_batch(pass_cli, remaining) + secrets.update(retry_secrets) + # Replace the auth warnings with the retry's outcome — anything still + # missing after a successful login is a real, reportable failure. + batch_warnings = retry_warnings + except RuntimeError as exc: + batch_warnings.append(f"pass-cli login failed: {exc}") + + warnings.extend(batch_warnings) + + # Cache only a complete, error-free pull. + if use_cache and len(secrets) == len(valid) and not batch_warnings and secrets: + entry = _CachedFetch(secrets=dict(secrets), fetched_at=time.time()) + _CACHE[cache_key] = entry + _write_disk_cache(cache_key, entry, cache_ttl_seconds, home_path) + + return secrets, warnings + + +# --------------------------------------------------------------------------- +# Public entry point — called from hermes_cli.env_loader +# --------------------------------------------------------------------------- + + +def apply_protonpass_secrets( + *, + enabled: bool, + env: Optional[Dict[str, str]] = None, + personal_access_token_env: str = _DEFAULT_TOKEN_ENV, + binary_path: str = "", + override_existing: bool = True, + cache_ttl_seconds: float = 300, + home_path: Optional[Path] = None, +) -> FetchResult: + """Resolve configured ``pass://`` references and set them on ``os.environ``. + + Called by ``load_hermes_dotenv()`` after the .env files have loaded. + Intentionally defensive — any failure returns a :class:`FetchResult` with + ``error`` set (or surfaces warnings); it never raises. + + Parameters mirror the ``secrets.protonpass.*`` config keys so the caller + can splat the dict in. References already satisfied by the current + environment (when ``override_existing`` is false) are skipped *before* + fetching, so ``pass-cli`` is never invoked for a value that would be + discarded. + """ + result = FetchResult() + + if not enabled: + return result + + valid, warnings = _validate_references(env) + result.warnings.extend(warnings) + + # Skip-before-fetch: never resolve a reference we'd only throw away. + refs_to_fetch: Dict[str, str] = {} + for name, ref in valid.items(): + if name == personal_access_token_env: + # Never let a resolved secret clobber the very token used to auth. + result.skipped.append(name) + continue + if not override_existing and os.environ.get(name): + result.skipped.append(name) + continue + refs_to_fetch[name] = ref + + if not refs_to_fetch: + return result + + binary = find_pass_cli(binary_path) + result.binary_path = binary + if binary is None: + if binary_path: + result.error = ( + f"secrets.protonpass.binary_path ({binary_path!r}) is not an " + "executable pass-cli binary." + ) + else: + result.error = ( + "secrets.protonpass.enabled is true but pass-cli was not found " + "on PATH. Install it (https://protonpass.github.io/pass-cli/) " + "or set secrets.protonpass.binary_path." + ) + return result + + try: + secrets, fetch_warnings = fetch_protonpass_secrets( + references=refs_to_fetch, + token_env=personal_access_token_env, + binary=binary, + cache_ttl_seconds=cache_ttl_seconds, + home_path=home_path, + ) + except RuntimeError as exc: + result.error = str(exc) + return result + + result.secrets = secrets + result.warnings.extend(fetch_warnings) + + for name, value in secrets.items(): + # The token-var and override guards already filtered refs_to_fetch, but + # re-check defensively in case the fetch layer ever returns extras. + if name == personal_access_token_env: + if name not in result.skipped: + result.skipped.append(name) + continue + if not override_existing and os.environ.get(name): + if name not in result.skipped: + result.skipped.append(name) + continue + os.environ[name] = value + result.applied.append(name) + + return result + + +# --------------------------------------------------------------------------- +# Test hook — used by hermetic tests to flush the cache between cases. +# --------------------------------------------------------------------------- + + +def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None: + """Clear in-process AND disk caches. + + Tests can pass ``home_path`` to scope the disk cleanup to a tmpdir. + Without it we fall back to the same default resolution as the writer. + """ + _CACHE.clear() + try: + _disk_cache_path(home_path).unlink() + except (FileNotFoundError, OSError): + pass diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 06e3ad5d7b9f..cfde1511d2e9 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2872,6 +2872,32 @@ def _ensure_hermes_home_managed(home: Path): # `hermes secrets bitwarden setup`. "server_url": "", }, + "protonpass": { + # Master switch. When false, pass-cli is never invoked — same as + # not having this section at all. + "enabled": False, + # Mapping of env-var name -> pass://vault/item/field reference. + # Resolved at startup and injected into os.environ. Entries whose + # name isn't a valid env-var name, or whose value isn't a pass:// + # reference, are skipped with a warning. + "env": {}, + # Name of the env var holding the Proton Pass personal access + # token. Used to (re)establish a `pass-cli` session when one isn't + # already present. This is the one bootstrap secret; it lives in + # ~/.hermes/.env (or your shell) and never in config.yaml. + "personal_access_token_env": "PROTON_PASS_PERSONAL_ACCESS_TOKEN", + # Absolute path to pass-cli. When set it is used verbatim and PATH + # is NOT consulted. Empty means resolve `pass-cli` from PATH. + "binary_path": "", + # Seconds to cache resolved secrets in-process and on disk. 0 + # disables both cache layers (nothing is written to disk). + "cache_ttl_seconds": 300, + # When True, resolved values overwrite existing env vars. Default + # True because the point of a central store is rotation — if .env + # had the final say, rotating in Proton Pass wouldn't take effect + # until you also cleared the matching .env line. + "override_existing": True, + }, }, # Paste collapse thresholds (TUI + CLI). @@ -2908,7 +2934,7 @@ def _ensure_hermes_home_managed(home: Path): # Config schema version - bump this when adding new required fields - "_config_version": 30, + "_config_version": 31, } # ============================================================================= @@ -5230,6 +5256,27 @@ def migrate_config(interactive: bool = True, quiet: bool = False) -> Dict[str, A "(LLM consolidation is now opt-in; pruning stays on)" ) + # ── Version 30 → 31: seed secrets.protonpass (Proton Pass source) ── + # The runtime deep-merge already supplies the default disabled section, but + # seed it so the new backend is visible/editable in config.yaml alongside + # secrets.bitwarden. Only add it when a secrets section exists and lacks it + # — never clobber a value the user already set. + if current_ver < 31: + config = read_raw_config() + raw_secrets = config.get("secrets") + if isinstance(raw_secrets, dict) and "protonpass" not in raw_secrets: + raw_secrets["protonpass"] = copy.deepcopy( + DEFAULT_CONFIG["secrets"]["protonpass"] + ) + config["secrets"] = raw_secrets + save_config(config) + results["config_added"].append("secrets.protonpass (disabled)") + if not quiet: + print( + " ✓ Seeded secrets.protonpass (Proton Pass pass:// source, " + "disabled by default)" + ) + # ── Post-migration: disable exfiltration-shaped MCP stdio entries ── # Users can hand-edit mcp_servers, and older installs may already contain a # malicious entry. Preserve the stanza for auditability but mark it diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index c7d507d8c2f3..4f060d826553 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging import os import sys from pathlib import Path @@ -9,6 +10,8 @@ from dotenv import load_dotenv from utils import atomic_replace +logger = logging.getLogger(__name__) + # Env var name suffixes that indicate credential values. These are the # only env vars whose values we sanitize on load — we must not silently @@ -78,6 +81,8 @@ def format_secret_source_suffix(env_var: str) -> str: return "" if source == "bitwarden": return " (from Bitwarden)" + if source == "protonpass": + return " (from Proton Pass)" # Generic fallback — future-proofing for additional secret sources # (e.g. 1Password, HashiCorp Vault) without having to update every # call site. @@ -307,8 +312,22 @@ def _apply_external_secret_sources(home_path: Path) -> None: except Exception: # noqa: BLE001 — config errors must not block startup return - bw_cfg = (cfg or {}).get("bitwarden") or {} - if not bw_cfg.get("enabled"): + if not isinstance(cfg, dict): + return + + # Each backend runs inside its own broad guard: a failure in one source (a + # crash, or a malformed config section) must neither abort startup nor stop + # the other sources from loading. + for apply_backend in (_apply_bitwarden, _apply_protonpass): + try: + apply_backend(cfg, home_path) + except Exception: # noqa: BLE001 — secret-source failures never block startup + logger.debug("%s failed", apply_backend.__name__, exc_info=True) + + +def _apply_bitwarden(cfg: dict, home_path: Path) -> None: + bw_cfg = cfg.get("bitwarden") or {} + if not isinstance(bw_cfg, dict) or not bw_cfg.get("enabled"): return try: @@ -321,39 +340,77 @@ def _apply_external_secret_sources(home_path: Path) -> None: access_token_env=bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN"), project_id=bw_cfg.get("project_id", ""), override_existing=bool(bw_cfg.get("override_existing", False)), - cache_ttl_seconds=float(bw_cfg.get("cache_ttl_seconds", 300)), + cache_ttl_seconds=_coerce_ttl(bw_cfg.get("cache_ttl_seconds", 300)), auto_install=bool(bw_cfg.get("auto_install", True)), server_url=str(bw_cfg.get("server_url", "") or "").strip(), home_path=home_path, ) + _record_secret_source_result("bitwarden", "Bitwarden Secrets Manager", result) + + +def _apply_protonpass(cfg: dict, home_path: Path) -> None: + pp_cfg = cfg.get("protonpass") or {} + if not isinstance(pp_cfg, dict) or not pp_cfg.get("enabled"): + return + + try: + from agent.secret_sources.protonpass import apply_protonpass_secrets + except ImportError: + return + env_map = pp_cfg.get("env") + env_map = env_map if isinstance(env_map, dict) else {} + result = apply_protonpass_secrets( + enabled=True, + env=env_map, + personal_access_token_env=pp_cfg.get( + "personal_access_token_env", "PROTON_PASS_PERSONAL_ACCESS_TOKEN" + ), + binary_path=str(pp_cfg.get("binary_path", "") or "").strip(), + override_existing=bool(pp_cfg.get("override_existing", True)), + cache_ttl_seconds=_coerce_ttl(pp_cfg.get("cache_ttl_seconds", 300)), + home_path=home_path, + ) + _record_secret_source_result("protonpass", "Proton Pass", result) + + +def _coerce_ttl(value: object, default: float = 300) -> float: + """Coerce a config TTL to float without ever raising. + + A stray ``cache_ttl_seconds: "abc"`` (or a YAML list) must not crash + startup — fall back to the default instead. + """ + try: + return float(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return default + + +def _record_secret_source_result(label: str, display: str, result) -> None: + """Record applied keys + print a one-line status for a backend result. + + Centralized so every backend labels its origin identically — the setup / + ``hermes model`` flows read ``_SECRET_SOURCES`` to show e.g. + "(from Proton Pass)" next to a detected credential instead of an + unexplained "credentials ✓". + """ if result.applied: - # Re-run the ASCII sanitization pass: BSM values are user-supplied - # and might have the same copy-paste corruption as a manually - # edited .env (see #6843). + # Re-run the ASCII sanitization pass: externally-sourced values are + # user-supplied and might have the same copy-paste corruption as a + # manually edited .env (see #6843). _sanitize_loaded_credentials() - # Remember where these came from so the setup / `hermes model` - # flows can label detected credentials with "(from Bitwarden)" — - # otherwise users see "credentials ✓" with no hint that the value - # came from BSM rather than .env. for name in result.applied: - _SECRET_SOURCES[name] = "bitwarden" + _SECRET_SOURCES[name] = label print( - f" Bitwarden Secrets Manager: applied {len(result.applied)} " + f" {display}: applied {len(result.applied)} " f"secret{'s' if len(result.applied) != 1 else ''} " f"({', '.join(sorted(result.applied))})", file=sys.stderr, ) if result.error: - print( - f" Bitwarden Secrets Manager: {result.error}", - file=sys.stderr, - ) + print(f" {display}: {result.error}", file=sys.stderr) for warn in result.warnings: - print( - f" Bitwarden Secrets Manager: {warn}", - file=sys.stderr, - ) + print(f" {display}: {warn}", file=sys.stderr) def _load_secrets_config(home_path: Path) -> dict: diff --git a/tests/test_env_loader_secret_sources.py b/tests/test_env_loader_secret_sources.py index 91c9d4c6e4f5..eb681e893715 100644 --- a/tests/test_env_loader_secret_sources.py +++ b/tests/test_env_loader_secret_sources.py @@ -173,3 +173,91 @@ def _fake_apply(**_kwargs): env_loader.reset_secret_source_cache() env_loader._apply_external_secret_sources(tmp_path) assert call_count["n"] == 2 + + +def test_format_secret_source_suffix_protonpass_uses_proper_name(): + env_loader._SECRET_SOURCES["ANTHROPIC_API_KEY"] = "protonpass" + assert ( + env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY") + == " (from Proton Pass)" + ) + + +def test_apply_external_secret_sources_records_protonpass_origin(tmp_path, monkeypatch): + """End-to-end: applied Proton Pass keys land in ``_SECRET_SOURCES``.""" + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + config_path = tmp_path / "config.yaml" + config_path.write_text( + "secrets:\n" + " protonpass:\n" + " enabled: true\n" + " env:\n" + " ANTHROPIC_API_KEY: pass://Private/Anthropic/credential\n", + encoding="utf-8", + ) + + from agent.secret_sources.protonpass import FetchResult + + def _fake_apply(**_kwargs): + return FetchResult( + secrets={"ANTHROPIC_API_KEY": "sk-ant-test"}, + applied=["ANTHROPIC_API_KEY"], + ) + + import agent.secret_sources.protonpass as pp_module + + monkeypatch.setattr(pp_module, "apply_protonpass_secrets", _fake_apply) + + env_loader._apply_external_secret_sources(tmp_path) + + assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "protonpass" + assert ( + env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY") + == " (from Proton Pass)" + ) + + +def test_apply_external_secret_sources_runs_protonpass_even_if_bitwarden_disabled( + tmp_path, monkeypatch +): + """A disabled Bitwarden section must not short-circuit Proton Pass. + + This is the regression guard for the refactor that split the single + inlined Bitwarden block into per-backend handlers: the old early + ``return`` on ``bitwarden.enabled == false`` would have skipped every + other source. + """ + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + config_path = tmp_path / "config.yaml" + config_path.write_text( + "secrets:\n" + " bitwarden:\n" + " enabled: false\n" + " protonpass:\n" + " enabled: true\n" + " env:\n" + " OPENAI_API_KEY: pass://Private/OpenAI/api key\n", + encoding="utf-8", + ) + + from agent.secret_sources.protonpass import FetchResult + + called = {"n": 0} + + def _fake_apply(**_kwargs): + called["n"] += 1 + return FetchResult( + secrets={"OPENAI_API_KEY": "sk-test"}, + applied=["OPENAI_API_KEY"], + ) + + import agent.secret_sources.protonpass as pp_module + + monkeypatch.setattr(pp_module, "apply_protonpass_secrets", _fake_apply) + + env_loader._apply_external_secret_sources(tmp_path) + + assert called["n"] == 1 + assert env_loader.get_secret_source("OPENAI_API_KEY") == "protonpass" diff --git a/tests/test_protonpass_secrets.py b/tests/test_protonpass_secrets.py new file mode 100644 index 000000000000..2ecfd011b3de --- /dev/null +++ b/tests/test_protonpass_secrets.py @@ -0,0 +1,544 @@ +"""Hermetic tests for the Proton Pass (`pass-cli`) secret source. + +We never invoke the real ``pass-cli`` binary: ``subprocess.run`` is mocked so +the suite stays fast and offline-safe. A live resolve is exercised manually +outside of pytest. +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from unittest import mock + +import pytest + + +# Make the worktree importable without depending on the installed wheel. +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from agent.secret_sources import protonpass as pp # noqa: E402 + + +@pytest.fixture(autouse=True) +def _reset_caches(): + pp._reset_cache_for_tests() + yield + pp._reset_cache_for_tests() + + +@pytest.fixture(autouse=True) +def _clean_pass_env(monkeypatch): + """Start every test from a known Proton Pass auth state.""" + monkeypatch.delenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", raising=False) + yield + + +def _ok(value: str): + return mock.Mock(returncode=0, stdout=value, stderr="") + + +def _err(code: int, stderr: str): + return mock.Mock(returncode=code, stdout="", stderr=stderr) + + +def _is_login(cmd) -> bool: + return "login" in cmd + + +def _ref_of(kwargs) -> str: + """The pass:// reference a resolve invocation was asked to resolve.""" + return kwargs["env"][pp._RESOLVE_SENTINEL] + + +def _resolver(values): + """A fake subprocess.run that resolves references from ``values``. + + ``values`` maps ``pass://…`` reference → stdout string. + """ + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + ref = _ref_of(kwargs) + return _ok(values[ref]) + + return fake_run + + +# --------------------------------------------------------------------------- +# Reference validation +# --------------------------------------------------------------------------- + + +def test_validate_references_filters_bad_names_and_refs(): + refs = { + "OPENAI_API_KEY": "pass://Private/OpenAI/api key", + "1BAD_NAME": "pass://Private/x/y", # bad env name + "HAS SPACE": "pass://Private/x/y", # bad env name + "NOT_A_REF": "https://example.com", # not pass:// + "WHITESPACE": " pass://Private/z/field ", # stripped + kept + } + valid, warnings = pp._validate_references(refs) + assert valid == { + "OPENAI_API_KEY": "pass://Private/OpenAI/api key", + "WHITESPACE": "pass://Private/z/field", + } + assert len(warnings) == 3 + + +# --------------------------------------------------------------------------- +# fetch_protonpass_secrets +# --------------------------------------------------------------------------- + + +def test_fetch_happy_path(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + values = { + "pass://Private/OpenAI/api key": "sk-abc", + "pass://Private/Anthropic/credential": "sk-ant-xyz", + } + monkeypatch.setattr(pp.subprocess, "run", _resolver(values)) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={ + "OPENAI_API_KEY": "pass://Private/OpenAI/api key", + "ANTHROPIC_API_KEY": "pass://Private/Anthropic/credential", + }, + binary=fake, + use_cache=False, + ) + assert secrets == {"OPENAI_API_KEY": "sk-abc", "ANTHROPIC_API_KEY": "sk-ant-xyz"} + assert warnings == [] + + +def test_fetch_uses_run_no_masking_and_option_terminator(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + captured = {} + + def fake_run(cmd, **kwargs): + captured["cmd"] = cmd + captured["env"] = kwargs["env"] + return _ok("value") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + cmd = captured["cmd"] + assert cmd[:3] == [str(fake), "run", "--no-masking"] + # `--` must precede the wrapped command so a crafted ref can't be a flag. + assert "--" in cmd + # The wrapped command is the current interpreter echoing the resolved + # value — cross-platform, no POSIX-only `printenv`. + assert cmd[cmd.index("--") + 1 :] == [sys.executable, "-c", pp._ECHO_SCRIPT] + # The reference is passed via the child env, not on argv. + assert captured["env"][pp._RESOLVE_SENTINEL] == "pass://V/I/F" + + +def test_fetch_empty_rc0_does_not_clobber(monkeypatch, tmp_path): + """returncode 0 with empty stdout must surface as a warning, not a value.""" + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr(pp.subprocess, "run", lambda *a, **k: _ok(" \n")) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + assert secrets == {} + assert any("empty value" in w for w in warnings) + + +def test_fetch_concealed_value_rejected(monkeypatch, tmp_path): + """A still-masked value must never be applied as a real secret.""" + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr( + pp.subprocess, "run", lambda *a, **k: _ok(pp._CONCEALED_MARKER + "\n") + ) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + assert secrets == {} + assert any("empty value" in w for w in warnings) + + +def test_fetch_read_failure_becomes_warning_and_scrubs_ansi(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr( + pp.subprocess, + "run", + lambda *a, **k: _err(1, "\x1b[31m[ERROR] no vault access\x1b[0m"), + ) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + assert secrets == {} + assert len(warnings) == 1 + assert "\x1b" not in warnings[0] + assert "[31m" not in warnings[0] + assert "no vault access" in warnings[0] + + +def test_fetch_one_bad_one_good(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + ref = _ref_of(kwargs) + return _ok("good-value") if ref == "pass://V/good/f" else _err(1, "no access") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={"GOOD": "pass://V/good/f", "BAD": "pass://V/bad/f"}, + binary=fake, + use_cache=False, + ) + assert secrets == {"GOOD": "good-value"} + assert len(warnings) == 1 + + +def test_fetch_auth_failure_triggers_login_and_retry(monkeypatch, tmp_path): + """An auth-shaped failure should drive one login + a successful retry.""" + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "pst_token") + state = {"logged_in": False, "logins": 0} + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + state["logged_in"] = True + state["logins"] += 1 + # The token must be injected for login, never on argv. + assert kwargs["env"].get("PROTON_PASS_PERSONAL_ACCESS_TOKEN") == "pst_token" + assert "pst_token" not in cmd + return _ok("") + if not state["logged_in"]: + return _err(1, "Error: not logged in") + return _ok("resolved-after-login") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + + secrets, warnings = pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + assert secrets == {"K": "resolved-after-login"} + assert state["logins"] == 1 + assert warnings == [] + + +def test_fetch_missing_binary_raises(monkeypatch): + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": None) + with pytest.raises(RuntimeError, match="pass-cli not found"): + pp.fetch_protonpass_secrets(references={"K": "pass://V/I/F"}, use_cache=False) + + +def test_fetch_child_env_is_allowlisted_and_tokenless_on_resolve(monkeypatch, tmp_path): + """The resolve child must NOT inherit provider creds or the PAT.""" + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setenv("OPENAI_API_KEY", "leak-me") + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "pst_token") + captured = {} + + def fake_run(cmd, **kwargs): + captured["env"] = kwargs["env"] + return _ok("v") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, binary=fake, use_cache=False + ) + env = captured["env"] + assert "OPENAI_API_KEY" not in env # not inherited + # The token is only for login; a resolve relies on the persistent session. + assert "PROTON_PASS_PERSONAL_ACCESS_TOKEN" not in env + assert env.get("NO_COLOR") == "1" + + +# --------------------------------------------------------------------------- +# Caching +# --------------------------------------------------------------------------- + + +def test_inprocess_cache_hit(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + calls = {"n": 0} + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + calls["n"] += 1 + return _ok("v") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp._reset_cache_for_tests(tmp_path) + for _ in range(2): + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=60, + binary=fake, home_path=tmp_path, + ) + assert calls["n"] == 1 # second call served from L1 cache + + +def test_disk_cache_roundtrip_and_no_token_on_disk(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "pst_supersecret") + calls = {"n": 0} + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + calls["n"] += 1 + return _ok("resolved") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp._reset_cache_for_tests(tmp_path) + + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=300, + binary=fake, home_path=tmp_path, + ) + assert calls["n"] == 1 + + cache_path = pp._disk_cache_path(tmp_path) + assert cache_path.exists() + assert (os.stat(cache_path).st_mode & 0o777) == 0o600 + text = cache_path.read_text() + assert "pst_supersecret" not in text # token never on disk + payload = json.loads(text) + assert payload["secrets"] == {"K": "resolved"} + + # Simulate a fresh process: clear only the in-process cache. + pp._CACHE.clear() + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=300, + binary=fake, home_path=tmp_path, + ) + assert calls["n"] == 1 # served from disk, pass-cli not re-invoked + + +def test_ttl_zero_disables_both_layers(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + calls = {"n": 0} + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + calls["n"] += 1 + return _ok("v") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp._reset_cache_for_tests(tmp_path) + + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=0, + binary=fake, home_path=tmp_path, + ) + assert not pp._disk_cache_path(tmp_path).exists() # nothing written at TTL 0 + pp._CACHE.clear() + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=0, + binary=fake, home_path=tmp_path, + ) + assert calls["n"] == 2 # never cached + + +def test_token_change_invalidates_cache(monkeypatch, tmp_path): + """A different personal access token must not reuse a cached value.""" + fake = tmp_path / "pass-cli" + fake.write_text("") + calls = {"n": 0} + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + calls["n"] += 1 + return _ok("v") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp._reset_cache_for_tests(tmp_path) + + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "pst_A") + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=300, + binary=fake, home_path=tmp_path, + ) + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "pst_B") + pp._CACHE.clear() + pp.fetch_protonpass_secrets( + references={"K": "pass://V/I/F"}, cache_ttl_seconds=300, + binary=fake, home_path=tmp_path, + ) + assert calls["n"] == 2 # cache key changed → refetch + + +def test_partial_failure_not_cached(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + + def fake_run(cmd, **kwargs): + if _is_login(cmd): + return _ok("") + ref = _ref_of(kwargs) + return _ok("v") if ref == "pass://V/good/f" else _err(1, "fail") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + pp._reset_cache_for_tests(tmp_path) + pp.fetch_protonpass_secrets( + references={"G": "pass://V/good/f", "B": "pass://V/bad/f"}, + cache_ttl_seconds=300, binary=fake, home_path=tmp_path, + ) + assert not pp._disk_cache_path(tmp_path).exists() + + +def test_reset_cache_clears_disk(tmp_path): + cache_path = pp._disk_cache_path(tmp_path) + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text("{}") + assert cache_path.exists() + pp._reset_cache_for_tests(tmp_path) + assert not cache_path.exists() + pp._reset_cache_for_tests(tmp_path) # idempotent + + +# --------------------------------------------------------------------------- +# find_pass_cli +# --------------------------------------------------------------------------- + + +def test_find_pass_cli_pinned_path_not_on_path(tmp_path, monkeypatch): + pinned = tmp_path / "pass-cli" + pinned.write_text("") + pinned.chmod(0o755) + monkeypatch.setattr(pp.shutil, "which", lambda name: "/usr/bin/pass-cli") + assert pp.find_pass_cli(str(pinned)) == pinned + + +def test_find_pass_cli_pinned_missing_returns_none(tmp_path, monkeypatch): + monkeypatch.setattr(pp.shutil, "which", lambda name: "/usr/bin/pass-cli") + assert pp.find_pass_cli(str(tmp_path / "nope")) is None + + +# --------------------------------------------------------------------------- +# apply_protonpass_secrets +# --------------------------------------------------------------------------- + + +def test_apply_disabled_returns_empty(): + result = pp.apply_protonpass_secrets(enabled=False, env={"K": "pass://V/I/F"}) + assert result.ok + assert not result.applied + + +def test_apply_missing_binary_sets_error(monkeypatch): + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": None) + result = pp.apply_protonpass_secrets(enabled=True, env={"K": "pass://V/I/F"}) + assert not result.ok + assert "pass-cli" in result.error + + +def test_apply_sets_env(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": fake) + monkeypatch.setattr(pp.subprocess, "run", lambda *a, **k: _ok("resolved-val")) + monkeypatch.delenv("MY_PP_KEY", raising=False) + + result = pp.apply_protonpass_secrets( + enabled=True, env={"MY_PP_KEY": "pass://V/I/F"}, cache_ttl_seconds=0, + ) + assert result.ok + assert result.applied == ["MY_PP_KEY"] + assert os.environ["MY_PP_KEY"] == "resolved-val" + + +def test_apply_skips_before_fetch_when_not_overriding(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": fake) + monkeypatch.setenv("MY_PP_KEY", "from-env") + calls = {"n": 0} + + def fake_run(*a, **k): + calls["n"] += 1 + return _ok("from-proton") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + + result = pp.apply_protonpass_secrets( + enabled=True, env={"MY_PP_KEY": "pass://V/I/F"}, + override_existing=False, cache_ttl_seconds=0, + ) + assert "MY_PP_KEY" in result.skipped + assert os.environ["MY_PP_KEY"] == "from-env" + assert calls["n"] == 0 # never even called pass-cli for a value we'd discard + + +def test_apply_never_overrides_token_var(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": fake) + monkeypatch.setenv("PROTON_PASS_PERSONAL_ACCESS_TOKEN", "original") + calls = {"n": 0} + + def fake_run(*a, **k): + calls["n"] += 1 + return _ok("malicious") + + monkeypatch.setattr(pp.subprocess, "run", fake_run) + + result = pp.apply_protonpass_secrets( + enabled=True, + env={"PROTON_PASS_PERSONAL_ACCESS_TOKEN": "pass://V/I/F"}, + override_existing=True, cache_ttl_seconds=0, + ) + assert "PROTON_PASS_PERSONAL_ACCESS_TOKEN" in result.skipped + assert os.environ["PROTON_PASS_PERSONAL_ACCESS_TOKEN"] == "original" + assert calls["n"] == 0 + + +def test_apply_never_raises_on_read_failure(monkeypatch, tmp_path): + fake = tmp_path / "pass-cli" + fake.write_text("") + monkeypatch.setattr(pp, "find_pass_cli", lambda binary_path="": fake) + monkeypatch.setattr(pp.subprocess, "run", lambda *a, **k: _err(1, "locked")) + monkeypatch.delenv("MY_PP_KEY", raising=False) + + result = pp.apply_protonpass_secrets( + enabled=True, env={"MY_PP_KEY": "pass://V/I/F"}, cache_ttl_seconds=0, + ) + # Fail-open: warnings, nothing applied, no fatal error, no exception. + assert result.ok + assert result.applied == [] + assert result.warnings + + +def test_apply_no_valid_refs_is_noop(monkeypatch): + # find_pass_cli must never be reached when there's nothing to fetch. + monkeypatch.setattr( + pp, "find_pass_cli", + lambda binary_path="": (_ for _ in ()).throw( + AssertionError("should not resolve pass-cli") + ), + ) + result = pp.apply_protonpass_secrets(enabled=True, env={"BAD NAME": "pass://V/I/F"}) + assert result.ok + assert result.applied == [] + assert result.warnings # the bad mapping warned diff --git a/website/docs/user-guide/secrets/index.md b/website/docs/user-guide/secrets/index.md index bf8d85cfed69..ea2b5e005d1c 100644 --- a/website/docs/user-guide/secrets/index.md +++ b/website/docs/user-guide/secrets/index.md @@ -5,5 +5,6 @@ Hermes can pull API keys from external secret managers at process startup instea Supported: - [Bitwarden Secrets Manager](./bitwarden) — `bws` CLI, lazy-installed, free tier works. +- [Proton Pass](./protonpass) — `pass-cli`, `pass://vault/item/field` references, persistent-session auth via a personal access token. More backends (Vault, AWS Secrets Manager, 1Password CLI) are easy to add behind the same interface — the lift is one module in `agent/secret_sources/` and one CLI handler. File a request if you have a specific one in mind. diff --git a/website/docs/user-guide/secrets/protonpass.md b/website/docs/user-guide/secrets/protonpass.md new file mode 100644 index 000000000000..a68e6e5dd415 --- /dev/null +++ b/website/docs/user-guide/secrets/protonpass.md @@ -0,0 +1,129 @@ +# Proton Pass + +Resolve provider API keys from [Proton Pass](https://proton.me/pass) at process startup instead of storing them in plaintext inside `~/.hermes/.env`. You keep your keys as Proton Pass items and reference them by `pass://vault/item/field`; rotating a credential becomes a single change in Proton Pass — every Hermes process picks up the new value on its next start (or, for the long-running gateway, within `cache_ttl_seconds`), with no container rebuild or `.env` edit. + +## How it works + +1. You install the official [Proton Pass CLI](https://protonpass.github.io/pass-cli/) (`pass-cli`) and authenticate it once with a **personal access token** (works headlessly in containers/CI). `pass-cli` stores a session in a platform keyring that persists across invocations. +2. You map environment-variable names to `pass://` references in `~/.hermes/config.yaml`. +3. Every time `hermes` (or the gateway, or a cron job) starts, after `~/.hermes/.env` has loaded, Hermes resolves each reference and sets the resolved values into `os.environ`. +4. By default Hermes **overrides** values already in your environment, so Proton Pass is the source of truth. Flip `override_existing: false` if you want `.env` to win instead. + +Hermes never downloads `pass-cli`: it shells out to your already-installed CLI. If `pass-cli` is missing, your session can't be established, or a reference is wrong, Hermes prints a one-line warning and continues with whatever credentials `.env` already had — it never blocks startup. + +## How resolution works + +For each `pass://` reference Hermes runs (conceptually): + +```bash +ENV_VAR='pass://vault/item/field' pass-cli run --no-masking -- -c "" +``` + +with the reference placed in that environment variable. The `run` command substitutes the `pass://` URI for the real secret value before executing the command; `--no-masking` is required so the value reaches stdout instead of being replaced with ``. Hermes wraps its own Python interpreter (rather than a shell builtin like `printenv`) so resolution behaves identically on Linux, macOS, and Windows. This uses only documented `pass-cli` behaviour and avoids the decorated output of `pass-cli item view`. + +## Authentication + +Proton Pass uses a **persistent session**, unlike the per-invocation token model of some other CLIs: + +- Create a **personal access token** (PAT) in Proton Pass, scoped to the vault(s) Hermes needs. +- Make the token available to Hermes as `PROTON_PASS_PERSONAL_ACCESS_TOKEN` (see [Bootstrap token](#bootstrap-token)). +- On startup Hermes relies on an existing `pass-cli` session if one is present. Only when a resolve fails for an auth/session reason does Hermes run `pass-cli login` (consuming the token from the configured env var, passed via the child environment — never on the argv) and retry once. + +You can also establish the session out of band — e.g. run `pass-cli login` once in your container entrypoint — and leave the token env var unset; Hermes will simply use the session. + +## Bootstrap token + +The personal access token is the one bootstrap credential Hermes may need *before* it can (re)establish a session. When you rely on Hermes to log in, the token must be present in `os.environ` of every process that resolves secrets — including cron jobs, subprocess invocations, CLI runs, and Docker containers. Put it in `~/.hermes/.env` (recommended), exactly like Bitwarden's `BWS_ACCESS_TOKEN`: + +```bash +echo 'PROTON_PASS_PERSONAL_ACCESS_TOKEN=pst_...' >> ~/.hermes/.env +chmod 600 ~/.hermes/.env +``` + +If the token is reachable only through an interactive shell, it will **not** be inherited by cron jobs or freshly spawned subprocesses — establish a session out of band for those contexts, or place the token in `.env`. + +## Setup + +### 1. Install and log in to `pass-cli` + +Follow the [Proton Pass CLI docs](https://protonpass.github.io/pass-cli/). Authenticate non-interactively with a personal access token: + +```bash +PROTON_PASS_PERSONAL_ACCESS_TOKEN='pst_...' pass-cli login +pass-cli vault list # verify +``` + +### 2. Map your credentials and enable + +Edit `~/.hermes/config.yaml`: + +```yaml +secrets: + protonpass: + enabled: true + env: + OPENAI_API_KEY: "pass://Private/OpenAI/api key" + ANTHROPIC_API_KEY: "pass://Private/Anthropic/credential" +``` + +From now on, every `hermes` invocation resolves the references at startup. You'll see a one-line summary in stderr the first time secrets are applied in a process. + +## Configuration + +Defaults in `~/.hermes/config.yaml`: + +```yaml +secrets: + protonpass: + enabled: false + env: {} + personal_access_token_env: PROTON_PASS_PERSONAL_ACCESS_TOKEN + binary_path: "" + cache_ttl_seconds: 300 + override_existing: true +``` + +| Key | Default | What it does | +|---|---|---| +| `enabled` | `false` | Master switch. When false, `pass-cli` is never invoked. | +| `env` | `{}` | Mapping of env-var name → `pass://vault/item/field` reference. Entries whose name isn't a valid env-var name, or whose value isn't a `pass://` reference, are skipped with a warning. | +| `personal_access_token_env` | `PROTON_PASS_PERSONAL_ACCESS_TOKEN` | Env var Hermes reads the personal access token from when it needs to (re)establish a `pass-cli` session. Leave the var set out of band if you manage the session yourself. | +| `binary_path` | `""` | Absolute path to `pass-cli`. When set, it is used verbatim and `PATH` is **not** consulted — pin this to avoid trusting whatever `pass-cli` appears first on `PATH`. | +| `cache_ttl_seconds` | `300` | How long resolved values are reused (in-process and on disk). Set to `0` to disable **both** cache layers — no values are written to disk at all. | +| `override_existing` | `true` | When true, resolved values overwrite anything already in env (so rotation takes effect). Flip to `false` to let `.env` / shell exports win; those references are then skipped *before* `pass-cli` is invoked. | + +## Failure modes + +Proton Pass never blocks Hermes startup. If anything goes wrong you'll see a one-line warning in stderr and Hermes continues: + +| Symptom | Cause | Fix | +|---|---|---| +| `pass-cli not found` | `pass-cli` not installed / not on PATH | Install the CLI, or set `secrets.protonpass.binary_path` | +| `pass-cli run failed for 'pass://…': …` | Bad reference, no vault access, or a locked session | Fix the reference, grant the token access, or re-`login` | +| `pass-cli login failed: …` | Missing/expired/invalid token | Refresh `PROTON_PASS_PERSONAL_ACCESS_TOKEN` | +| `pass-cli returned an empty value for 'pass://…'` | The referenced field exists but is empty | Fix the item/field in Proton Pass (an empty value is never applied — your existing env var is left intact) | +| `… is not a pass:// secret reference` | A mapping value isn't a `pass://` reference | Re-set it with the correct `pass://vault/item/field` form | + +## Caching + +Successful, complete pulls are cached in-process and on disk under `/cache/protonpass_cache.json` (written atomically, mode `0600`), so back-to-back short-lived `hermes` invocations don't re-shell `pass-cli` for every reference. The cache: + +- stores only resolved secret **values** — never the personal access token or any raw auth material (the token is fingerprinted into the cache key); +- is invalidated when the token or the set of references change; +- is **not** written when a pull had any per-reference error, so a transient auth failure isn't frozen in for the TTL; +- is fully disabled — reads *and* writes — when `cache_ttl_seconds: 0`. + +## Security notes + +- A Proton Pass personal access token can read every secret in the vaults it's scoped to. Store it in `~/.hermes/.env` (not `config.yaml`), scope it as narrowly as possible, and revoke + regenerate from Proton Pass if it leaks. +- Hermes refuses to let a resolved value overwrite the token env var itself, even with `override_existing: true`. +- The `pass-cli` child process gets a minimal allowlisted environment (session/keyring vars + `PATH`/`HOME`), not a copy of the full `os.environ`, so post-dotenv provider credentials aren't all inherited by the child. +- References are validated to start with `pass://`, and resolution goes through `pass-cli run`'s documented substitution rather than string-splicing values into a shell. + +## When NOT to use this + +- **Single-machine personal setups** where `~/.hermes/.env` is fine. +- **Air-gapped environments** that can't reach Proton. +- **CI/CD** where an existing secrets-injection mechanism is already wired up — pick one path, not two. + +The good case for this is multi-machine fleets, shared dev boxes, gateway VPSes, or anywhere you want centralized rotation and revocation across multiple Hermes installations.