From 924e290741c6329025a979ef646d526f3e0664b2 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 10:07:21 -0700 Subject: [PATCH] refactor(agent/creds): rebase bws/op/command sources onto the shared base+cache substrate - bitwarden: drop dead apply_bitwarden_secrets (0 refs); encrypted cache uses atomic_write_json/entry_from_payload; fetch goes through SecretCache.lookup with an encrypted L2 reader; stale-fallback branches merged; _classify_bws_error is a rule table; token/override hooks come from the ABC. - onepassword: same substrate; _missing_binary_error, _fingerprint, _guarded dedupe repeated text/logic; _classify_op_error is a rule table. - command: drop dead parse_secret_output/get_command_secret/list_command_secrets/ apply_command_secrets (0 refs outside own test); _log helper; tests repointed to _run_helper / CommandSource.fetch. --- agent/secret_sources/bitwarden.py | 716 ++++++++-------------------- agent/secret_sources/command.py | 387 +++------------ agent/secret_sources/onepassword.py | 460 +++++------------- tests/test_command_secret_source.py | 64 +-- 4 files changed, 396 insertions(+), 1231 deletions(-) diff --git a/agent/secret_sources/bitwarden.py b/agent/secret_sources/bitwarden.py index 6419522177..b305d973b9 100644 --- a/agent/secret_sources/bitwarden.py +++ b/agent/secret_sources/bitwarden.py @@ -1,30 +1,19 @@ """Bitwarden Secrets Manager (`bws` CLI) integration. -Hermes pulls API keys from Bitwarden Secrets Manager at process startup -so they don't have to live in plaintext in ``~/.hermes/.env``. +Hermes pulls API keys from Bitwarden Secrets Manager at startup so they don't +have to live in plaintext in ``~/.hermes/.env``. -Design summary --------------- +* ``bws`` is auto-installed into ``/bin/bws`` on first use: one + pinned version (``_BWS_VERSION``) downloaded from the official GitHub + release and SHA-256-verified against the published checksum file. +* The one bootstrap secret is the access token in ``.env`` (``BWS_ACCESS_TOKEN`` + or ``secrets.bitwarden.access_token_env``); every other key can live in BSM. +* One ``bws secret list --output json`` call per fetch, cached + in-process and on disk for ``cache_ttl_seconds``. +* Failures NEVER block startup: a one-line warning, then continue with .env. -* The ``bws`` binary is auto-installed into ``/bin/bws`` on - first use. Hermes pins one version (``_BWS_VERSION``) and downloads - the matching asset from the official GitHub Releases page, verifying - the SHA-256 against the release's published checksum file. -* The access token is stored in ``~/.hermes/.env`` as - ``BWS_ACCESS_TOKEN`` (or whatever name the user picked in - ``secrets.bitwarden.access_token_env``). This is the one - bootstrap secret — every other provider key can live in Bitwarden. -* Pulling secrets is a single ``bws secret list - --output json`` call. We cache the result in-process for - ``cache_ttl_seconds`` so back-to-back ``hermes`` invocations don't - hammer the API. -* Failures NEVER block Hermes startup. Missing binary, no network, - expired token, etc. all emit a one-line warning and continue with - whatever credentials ``.env`` already had. - -The module is intentionally subprocess-driven rather than going through -the ``bitwarden-sdk-secrets`` Python package: one cross-platform binary -is easier to lazy-install than a wheels-with-Rust-extension dependency. +Subprocess-driven on purpose: one cross-platform binary is easier to +lazy-install than the ``bitwarden-sdk-secrets`` Rust-extension wheel. """ from __future__ import annotations @@ -37,7 +26,6 @@ import os import platform import re import shutil -import stat import subprocess import tempfile import time @@ -49,50 +37,39 @@ from typing import Dict, List, Optional, Tuple from agent.secret_sources._cache import ( CachedFetch as _CachedFetch, - DiskCache, - FetchResult, - is_valid_env_name as _is_valid_env_name, + SecretCache, + atomic_write_json, + entry_from_payload, + resolve_cache_home, +) +from agent.secret_sources.base import ( + ErrorKind, + FetchResult, + SecretSource, + classify_cli_error, + coerce_float, + is_valid_env_name as _is_valid_env_name, + get_source_environment, + run_cli, + source_child_env, ) -from agent.secret_sources.base import ErrorKind, SecretSource -from agent.secret_sources.base import get_source_environment logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Configuration constants -# --------------------------------------------------------------------------- - -# Pinned upstream version. Bump in a follow-up PR — never auto-resolve -# "latest" because upstream release shape (asset names, CLI flags) is -# allowed to change between majors and we want updates to be deliberate. +# Pinned upstream version — never auto-resolve "latest": release shape (asset +# names, CLI flags) may change between majors and updates must be deliberate. _BWS_VERSION = "2.0.0" - _BWS_RELEASE_BASE = ( f"https://github.com/bitwarden/sdk-sm/releases/download/bws-v{_BWS_VERSION}" ) _BWS_CHECKSUM_NAME = f"bws-sha256-checksums-{_BWS_VERSION}.txt" - -# How long to wait for bws subprocesses and HTTP downloads, in seconds. _BWS_DOWNLOAD_TIMEOUT = 60 _BWS_RUN_TIMEOUT = 30 -# In-process cache so repeated load_hermes_dotenv() calls (CLI startup, -# gateway hot-reload, test suites) don't re-fetch from BSM. +# Cache layout: /cache/bws_cache.json holds only secret VALUES +# (never the access token) — plaintext-equivalent to .env but kept out of it so +# users editing .env don't commit BSM-sourced secrets. _CacheKey = Tuple[str, str, str] # (access_token_fingerprint, project_id, server_url) -_CACHE: Dict[_CacheKey, _CachedFetch] = {} - -# Disk-persisted cache so back-to-back CLI invocations (e.g. `hermes chat -q ...` -# called from scripts, cron, the gateway forking new agents) don't each pay the -# ~380ms `bws secret list` tax. The in-process _CACHE above only saves repeated -# fetches WITHIN one process; this saves repeated fetches ACROSS processes. -# -# Layout: one JSON object per cache key, written atomically with mode 0600 in -# /cache/bws_cache.json. The file holds only the secret VALUES, -# never the access token. It's plaintext-equivalent to ~/.hermes/.env (which -# we already accept) but kept out of the .env file so users editing it won't -# accidentally commit BSM-sourced secrets. The atomic-write/0600/TTL mechanics -# live in agent.secret_sources._cache.DiskCache, shared with the other backends. _DISK_CACHE_BASENAME = "bws_cache.json" _ENCRYPTED_CACHE_BASENAME = "bws_cache.enc.json" _ENCRYPTED_CACHE_VERSION = 1 @@ -100,29 +77,18 @@ _ENCRYPTED_CACHE_INFO = b"hermes-bws-encrypted-cache-v1" def _cache_key_str(cache_key: _CacheKey) -> str: - """Serialize a cache key to a stable string for JSON storage.""" token_fp, project_id, server_url = cache_key return f"{token_fp}|{project_id}|{server_url}" -_DISK_CACHE: DiskCache = DiskCache( - _DISK_CACHE_BASENAME, key_serializer=_cache_key_str -) - - -def _disk_cache_path(home_path: Optional[Path] = None) -> Path: - """Return the disk cache path under hermes_home/cache/. - - Thin wrapper over the shared DiskCache, kept for tests and any direct - callers; falls back to `$HERMES_HOME` / `~/.hermes` when home is None. - """ - return _DISK_CACHE.path(home_path) +_STORE: SecretCache[_CacheKey] = SecretCache(_DISK_CACHE_BASENAME, key_serializer=_cache_key_str) +# Test seams: L1 dict, L2 DiskCache, and its path. +_CACHE = _STORE.memory +_DISK_CACHE = _STORE.disk +_disk_cache_path = _DISK_CACHE.path def _encrypted_disk_cache_path(home_path: Optional[Path] = None) -> Path: - """Return the encrypted disk cache path under hermes_home/cache/.""" - from agent.secret_sources._cache import resolve_cache_home - return resolve_cache_home(home_path) / "cache" / _ENCRYPTED_CACHE_BASENAME @@ -139,15 +105,7 @@ def _hermes_bin_dir() -> Path: def find_bws(*, install_if_missing: bool = False) -> Optional[Path]: - """Return a path to a usable ``bws`` binary, or None. - - Resolution order: - 1. ``/bin/bws`` (our managed copy — preferred) - 2. ``shutil.which("bws")`` (system PATH) - - When ``install_if_missing`` is True and neither resolves, this calls - :func:`install_bws` to download and verify the pinned version. - """ + """Managed ``/bin/bws`` first, then PATH, then optional auto-install.""" managed = _hermes_bin_dir() / _platform_binary_name() if managed.exists() and os.access(managed, os.X_OK): return managed @@ -170,29 +128,19 @@ def _platform_binary_name() -> str: def _platform_asset_name() -> str: - """Map (uname, arch, libc) → the upstream asset filename. - - Asset names follow Rust's target triple convention. Linux defaults - to gnu (glibc); we switch to musl only if ldd --version says so. - """ + """Map (uname, arch, libc) → upstream asset filename (Rust target-triple style).""" system = platform.system() machine = platform.machine().lower() + arch = "aarch64" if machine in ("arm64", "aarch64") else "x86_64" - if system == "Darwin": - # Universal binary works on both Intel and Apple Silicon — no - # need to pick a per-arch asset. + if system == "Darwin": # universal binary covers Intel + Apple Silicon return f"bws-macos-universal-{_BWS_VERSION}.zip" - if system == "Windows": - arch = "aarch64" if machine in ("arm64", "aarch64") else "x86_64" return f"bws-{arch}-pc-windows-msvc-{_BWS_VERSION}.zip" - if system == "Linux": - arch = "aarch64" if machine in ("arm64", "aarch64") else "x86_64" + # glibc default; musl only if ldd says so (glibc prints to stderr, musl + # to stdout). A wrong guess surfaces as a loader error we catch. libc = "gnu" - # ldd --version writes to stderr on glibc, stdout on musl. We - # don't need bullet-proof detection — getting it wrong falls - # back to a clear error from the binary loader, which we catch. try: res = subprocess.run( ["ldd", "--version"], @@ -213,12 +161,10 @@ def _platform_asset_name() -> str: def install_bws(*, force: bool = False) -> Path: - """Download, verify, and install the pinned ``bws`` binary. + """Download, verify, and install the pinned ``bws`` binary; raises on any failure. - Returns the path to the installed executable. Raises on any - failure (network, checksum, extraction) — callers in the auto-install - path catch these; the user-facing ``hermes secrets bitwarden setup`` - surface lets them propagate so the wizard can show a clear error. + The auto-install path catches; ``hermes secrets bitwarden setup`` lets the + error propagate so the wizard can show it. """ bin_dir = _hermes_bin_dir() bin_dir.mkdir(parents=True, exist_ok=True) @@ -250,23 +196,13 @@ def install_bws(*, force: bool = False) -> Path: with zipfile.ZipFile(zip_path) as zf: member = _pick_zip_member(zf, _platform_binary_name()) - # Zip-slip guard: a malicious archive can carry member names like - # ``../../etc/cron.d/x`` or absolute paths. ``ZipFile.extract`` - # joins the member onto ``tmp`` without verifying the result stays - # inside it, so validate containment before touching the disk. extracted = _safe_extract_member(zf, member, tmp) - # Move into place atomically. We write to a sibling tempfile in - # the final directory so the rename can't cross filesystems. + # Stage in the final directory so the rename can't cross filesystems. fd, staged = tempfile.mkstemp(dir=str(bin_dir), prefix=".bws_") os.close(fd) shutil.copy2(extracted, staged) - os.chmod( - staged, - stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR - | stat.S_IRGRP | stat.S_IXGRP - | stat.S_IROTH | stat.S_IXOTH, - ) + os.chmod(staged, 0o755) os.replace(staged, target) logger.info("Installed bws %s at %s", _BWS_VERSION, target) @@ -284,11 +220,7 @@ def _http_download(url: str, dest: Path) -> None: def _expected_sha256(checksum_file: Path, asset_name: str) -> str: - """Parse the upstream ``bws-sha256-checksums-X.Y.Z.txt`` file. - - Format is the standard ``sha256sum`` output: `` ``, - one per line. - """ + """Parse standard ``sha256sum`` output (`` `` per line).""" text = checksum_file.read_text(encoding="utf-8", errors="replace") for line in text.splitlines(): parts = line.strip().split() @@ -308,18 +240,13 @@ def _sha256_file(path: Path) -> str: def _pick_zip_member(zf: zipfile.ZipFile, binary_name: str) -> str: - """Find the binary inside the upstream zip. - - Historically the archive has been flat (``bws`` at the root) but we - tolerate a top-level directory just in case upstream changes. - """ + """Find the binary in the zip; tolerate a top-level dir, prefer the shortest path.""" candidates = [n for n in zf.namelist() if n.split("/")[-1] == binary_name] if not candidates: raise RuntimeError( f"Could not find {binary_name} inside downloaded archive " f"(members: {zf.namelist()[:5]}...)" ) - # Prefer the shortest path (i.e. root over nested) for determinism. candidates.sort(key=len) return candidates[0] @@ -327,18 +254,14 @@ def _pick_zip_member(zf: zipfile.ZipFile, binary_name: str) -> str: def _safe_extract_member( zf: zipfile.ZipFile, member: str, dest_dir: Path ) -> Path: - """Extract a single archive member, refusing path traversal. + """Extract one member, refusing zip-slip (``../`` or absolute member names). - ``ZipFile.extract`` will happily honour member names containing - ``../`` or absolute paths, letting a malicious archive write outside - ``dest_dir`` (a "zip-slip"). We resolve the would-be target and - confirm it stays within ``dest_dir`` before extracting. + ``ZipFile.extract`` joins the member onto ``dest_dir`` without verifying the + result stays inside it, so containment is checked here first. """ dest_root = os.path.realpath(dest_dir) target = os.path.realpath(os.path.join(dest_root, member)) - # ``commonpath`` raises ValueError for e.g. different drives on - # Windows; treat that as an escape too. - try: + try: # commonpath raises for e.g. different Windows drives — treat as escape contained = os.path.commonpath([dest_root, target]) == dest_root except ValueError: contained = False @@ -352,7 +275,7 @@ def _safe_extract_member( # --------------------------------------------------------------------------- -# Secret fetch + apply +# Encrypted last-good cache (opt-in) # --------------------------------------------------------------------------- @@ -370,11 +293,12 @@ def _b64d(text: str) -> bytes: def _derive_encrypted_cache_key(access_token: str, salt: bytes) -> bytes: - """Derive the local cache encryption key from the bootstrap BWS token.""" - # Keep the native cryptography extension lazy. Most CLI commands import - # this module while building argparse, even though only encrypted-cache - # reads/writes need it. Eagerly importing it maps ``_rust.pyd`` into a - # Windows updater and prevents uv from replacing that file (#73381). + """HKDF the local cache key from the bootstrap BWS token. + + cryptography is imported lazily: most CLI commands import this module while + building argparse, and eagerly mapping ``_rust.pyd`` on Windows blocks the + updater from replacing that file. + """ from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.hkdf import HKDF @@ -393,21 +317,14 @@ def _write_encrypted_disk_cache( entry: _CachedFetch, home_path: Optional[Path] = None, ) -> None: - """Persist an encrypted last-good cache entry atomically. + """Persist an AES-GCM encrypted last-good entry atomically (best-effort). - Best-effort by design: cache write failure must never block a fresh BWS - fetch. The raw BWS access token is not stored; it only derives the AES key. + The raw token is never stored; it only derives the key. A successful write + completes migration, so the legacy plaintext cache is removed. """ - path = _encrypted_disk_cache_path(home_path) try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM - cache_dir = path.parent - cache_dir.mkdir(parents=True, exist_ok=True) - try: - os.chmod(cache_dir, 0o700) - except OSError: - pass salt = os.urandom(16) nonce = os.urandom(12) serialized_key = _cache_key_str(cache_key) @@ -426,28 +343,9 @@ def _write_encrypted_disk_cache( "nonce": _b64e(nonce), "ciphertext": _b64e(ciphertext), } - fd, tmp = tempfile.mkstemp( - prefix=".bws_cache_enc_", 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) - # A successful encrypted write completes migration; remove the - # legacy plaintext cache so stale secrets cannot remain on disk. - try: - _disk_cache_path(home_path).unlink() - except FileNotFoundError: - pass - except OSError: - pass - except BaseException: - try: - os.unlink(tmp) - except OSError: - pass - raise + atomic_write_json(_encrypted_disk_cache_path(home_path), payload, + tmp_prefix=".bws_cache_enc_") + _STORE.disk.clear(home_path) except Exception: # noqa: BLE001 — best-effort cache only return @@ -459,47 +357,40 @@ def _read_encrypted_disk_cache( max_age_seconds: float, home_path: Optional[Path] = None, ) -> Optional[_CachedFetch]: - """Return a decrypted encrypted-cache entry if it matches and is in-window.""" + """Decrypted encrypted-cache entry if it matches ``cache_key`` and is in-window.""" if max_age_seconds <= 0: return None - path = _encrypted_disk_cache_path(home_path) try: from cryptography.hazmat.primitives.ciphers.aead import AESGCM - payload = json.loads(path.read_text(encoding="utf-8")) - if not isinstance(payload, dict): - return None + payload = json.loads(_encrypted_disk_cache_path(home_path).read_text(encoding="utf-8")) serialized_key = _cache_key_str(cache_key) - if payload.get("version") != _ENCRYPTED_CACHE_VERSION: - return None - if payload.get("key") != serialized_key: + if (not isinstance(payload, dict) + or payload.get("version") != _ENCRYPTED_CACHE_VERSION + or payload.get("key") != serialized_key): return None salt = _b64d(str(payload.get("salt", ""))) nonce = _b64d(str(payload.get("nonce", ""))) ciphertext = _b64d(str(payload.get("ciphertext", ""))) key = _derive_encrypted_cache_key(access_token, salt) - raw = AESGCM(key).decrypt( + entry = entry_from_payload(json.loads(AESGCM(key).decrypt( nonce, ciphertext, serialized_key.encode("utf-8") - ) - inner = json.loads(raw.decode("utf-8")) - if not isinstance(inner, dict): + ).decode("utf-8"))) + if entry is None: return None - secrets = inner.get("secrets") - inner_fetched_at = inner.get("fetched_at") - if not isinstance(secrets, dict) or not isinstance(inner_fetched_at, (int, float)): - return None - entry_age = time.time() - float(inner_fetched_at) + entry_age = time.time() - entry.fetched_at if entry_age < 0 or entry_age > max_age_seconds: return None - typed = { - k: v for k, v in secrets.items() - if isinstance(k, str) and isinstance(v, str) - } - return _CachedFetch(secrets=typed, fetched_at=float(inner_fetched_at)) + return entry except Exception: # noqa: BLE001 — cache miss on parse/decrypt/I/O errors return None +# --------------------------------------------------------------------------- +# Secret fetch +# --------------------------------------------------------------------------- + + def fetch_bitwarden_secrets( *, access_token: str, @@ -512,28 +403,19 @@ def fetch_bitwarden_secrets( encrypted_cache_enabled: bool = False, encrypted_cache_max_stale_seconds: float = 0, ) -> Tuple[Dict[str, str], List[str]]: - """Pull the secrets for ``project_id`` from Bitwarden Secrets Manager. + """Pull the secrets for ``project_id`` from BSM → ``(secrets, warnings)``. - Returns ``(secrets_dict, warnings_list)``. + ``server_url`` selects a region / self-hosted instance (``BWS_SERVER_URL``; + empty = bws default, US Cloud). ``cache_ttl_seconds`` governs the fresh + cache. With ``encrypted_cache_enabled`` fresh entries are written AES-GCM + encrypted instead of plaintext, and a last-good entry may be served after + NETWORK/TIMEOUT failures for up to ``encrypted_cache_max_stale_seconds`` — + independent of the fresh TTL, so ``cache_ttl_seconds: 0`` can coexist with + a break-glass offline cache. - Set ``server_url`` to point at a non-default Bitwarden region or a - self-hosted instance — e.g. ``https://vault.bitwarden.eu`` for EU - Cloud accounts. When empty, ``bws`` uses its built-in default - (``https://vault.bitwarden.com``, US Cloud). This is plumbed into - the subprocess as ``BWS_SERVER_URL``. - - ``cache_ttl_seconds`` controls the normal fresh cache. When - ``encrypted_cache_enabled`` is true, fresh cache entries are written as - AES-GCM encrypted JSON instead of plaintext, and a last-good encrypted - entry may be used after NETWORK/TIMEOUT failures for up to - ``encrypted_cache_max_stale_seconds``. This stale fallback is separate - from the fresh-cache TTL so operators can set ``cache_ttl_seconds: 0`` - while still keeping an encrypted break-glass cache for offline startup. - - Raises :class:`RuntimeError` for fatal conditions (missing binary, - auth failure, unparseable output). Callers in the env_loader path - catch this and emit a single warning; callers in the user-facing - setup wizard let it propagate. + Raises ``RuntimeError`` for fatal conditions (missing binary, auth failure, + unparseable output); the env_loader path catches, the setup wizard lets it + propagate. """ if not access_token: raise RuntimeError("Bitwarden access token is empty") @@ -541,25 +423,21 @@ def fetch_bitwarden_secrets( raise RuntimeError("Bitwarden project_id is empty") cache_key = (_token_fingerprint(access_token), project_id, server_url or "") + + def _read_encrypted(max_age: float) -> Optional[_CachedFetch]: + return _read_encrypted_disk_cache( + cache_key=cache_key, access_token=access_token, + max_age_seconds=max_age, home_path=home_path, + ) + if use_cache and cache_ttl_seconds > 0: - cached = _CACHE.get(cache_key) - if cached and cached.is_fresh(cache_ttl_seconds): + # L2 (~5ms) vs ~380ms for `bws secret list`. + cached = _STORE.lookup( + cache_key, cache_ttl_seconds, home_path, + read_disk=(lambda: _read_encrypted(cache_ttl_seconds)) if encrypted_cache_enabled else None, + ) + if cached is not None: return cached.secrets, [] - # L2: disk cache. ~5ms on cache hit vs ~380ms for `bws secret list`. - if encrypted_cache_enabled: - disk_cached = _read_encrypted_disk_cache( - cache_key=cache_key, - access_token=access_token, - max_age_seconds=cache_ttl_seconds, - home_path=home_path, - ) - else: - disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path) - if disk_cached is not None: - # Promote into in-process cache so subsequent fetches in the - # same process skip the disk read too. - _CACHE[cache_key] = disk_cached - return disk_cached.secrets, [] bws = binary or find_bws(install_if_missing=True) if bws is None: @@ -573,86 +451,52 @@ def fetch_bitwarden_secrets( try: secrets, warnings = _run_bws_list(bws, access_token, project_id, server_url) except RuntimeError as exc: - # Live fetch failed. Fall back to a stale disk cache ONLY for - # transport-level failures (network down, DNS error, transient BWS - # outage / timeout) — never for AUTH_FAILED or a malformed-output - # INTERNAL error, where serving old secrets would mask a real - # config/credential problem the caller needs to see. Without this - # fallback a fleet of bots sharing one BWS project all stop working - # on a single network blip. - # - # Two fallback tiers share the transport-only gate: - # * encrypted cache (opt-in) — AES-GCM payload keyed off the - # bootstrap token, with its own max_stale_seconds window. When - # enabled it is the ONLY fallback consulted: the whole point is - # that the at-rest payload is never plaintext, so we don't - # quietly serve the plaintext file alongside it. - # * plaintext disk cache (default) — the ordinary DiskCache file. - # `cache_ttl_seconds <= 0` means the caller opted out of caching - # entirely (DiskCache.read/write both short-circuit on it) — - # honor that on the fallback path too. `ttl_seconds=inf` on the - # read bypasses freshness (we explicitly want a stale hit); the - # caller's real TTL gates whether we even attempt the read. + # Stale fallback ONLY for transport failures — never AUTH_FAILED or a + # malformed-output INTERNAL error, where serving old secrets would mask + # a real config/credential problem. Without it a fleet sharing one BSM + # project all stops on a single network blip. + # When the encrypted cache is enabled it is the ONLY fallback consulted + # (the at-rest payload must never be plaintext). Otherwise the plain + # DiskCache is read with ttl=inf (a stale hit is the point) but only if + # the caller's real TTL > 0 — ttl<=0 means caching is opted out. kind = _classify_bws_error(str(exc)) if use_cache and kind in (ErrorKind.NETWORK, ErrorKind.TIMEOUT): + stale = label = None if encrypted_cache_enabled: - stale = _read_encrypted_disk_cache( - cache_key=cache_key, - access_token=access_token, - max_age_seconds=encrypted_cache_max_stale_seconds, - home_path=home_path, - ) - if stale is not None: - age = max(0.0, time.time() - stale.fetched_at) - _CACHE[cache_key] = stale - return stale.secrets, [ - f"bws live fetch failed ({exc}); falling back to " - f"stale ENCRYPTED disk cache ({int(age)}s old)" - ] + stale = _read_encrypted(encrypted_cache_max_stale_seconds) + label = "stale ENCRYPTED disk cache" elif cache_ttl_seconds > 0: - stale = _DISK_CACHE.read(cache_key, float("inf"), home_path) - if stale is not None: - age = max(0.0, time.time() - stale.fetched_at) - _CACHE[cache_key] = stale - return stale.secrets, [ - f"bws live fetch failed ({exc}); " - f"falling back to stale disk cache ({int(age)}s old)" - ] + stale = _STORE.disk.read(cache_key, float("inf"), home_path) + label = "stale disk cache" + if stale is not None: + age = max(0.0, time.time() - stale.fetched_at) + _STORE.memory[cache_key] = stale + return stale.secrets, [ + f"bws live fetch failed ({exc}); falling back to {label} ({int(age)}s old)" + ] raise entry = _CachedFetch(secrets=secrets, fetched_at=time.time()) if use_cache: if cache_ttl_seconds > 0: - _CACHE[cache_key] = entry + _STORE.memory[cache_key] = entry if encrypted_cache_enabled: - # Encryption is the storage policy; max_stale_seconds only controls - # whether an outage may consume the last-good entry. Never fall - # back to the plaintext cache just because stale fallback is off. + # Encryption is the storage policy; max_stale only gates outage + # reads. Never fall back to plaintext because stale fallback is off. _write_encrypted_disk_cache( - cache_key=cache_key, - access_token=access_token, - entry=entry, - home_path=home_path, + cache_key=cache_key, access_token=access_token, + entry=entry, home_path=home_path, ) - elif cache_ttl_seconds > 0: - _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path) + else: + _STORE.disk.write(cache_key, entry, cache_ttl_seconds, home_path) return secrets, warnings def _summarize_bws_stderr(raw: str) -> str: """Reduce a bws (Rust color-eyre) error dump to its cause line(s). - bws failures look like:: - - Error: - 0: Received error message from server: [400 Bad Request] {"error":"invalid_client"} - - Location: - crates/bws/src/main.rs:108 - ... - - Everything from ``Location:`` on is diagnostic noise for a Hermes - user. Keep the numbered cause lines (joined), drop the rest, and - fall back to the stripped raw text when the shape is unrecognized. + Keeps the numbered ``0: ...`` cause lines (joined with ``; ``), drops + everything from ``Location:``/``Backtrace omitted`` on, and falls back to + the stripped raw text when the shape is unrecognized. """ text = raw.replace("\x1b", "").strip() if not text: @@ -664,7 +508,6 @@ def _summarize_bws_stderr(raw: str) -> str: break if stripped in ("", "Error:"): continue - # Cause lines are numbered "0: ...", "1: ..." — strip the index. stripped = re.sub(r"^\d+:\s*", "", stripped) if stripped: causes.append(stripped) @@ -675,47 +518,19 @@ def _run_bws_list( bws: Path, access_token: str, project_id: str, server_url: str = "" ) -> Tuple[Dict[str, str], List[str]]: cmd = [str(bws), "secret", "list", project_id, "--output", "json"] - # bws child intentionally receives the access token. Under a profile-local - # fetch it must not inherit sibling credentials from process-global env. - source_env = get_source_environment() - if source_env is os.environ: - from tools.environments.local import build_subprocess_env - - env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False) - else: - env = dict(source_env) + # The bws child intentionally receives the access token; a profile-local + # fetch must not inherit sibling credentials (source_child_env). + env = source_child_env() env["BWS_ACCESS_TOKEN"] = access_token - # Make sure we're not echoing telemetry / colour codes into json. env.setdefault("NO_COLOR", "1") - # Region / self-hosted support. bws defaults to https://vault.bitwarden.com - # (US Cloud); EU Cloud users need https://vault.bitwarden.eu, and - # self-hosted users need their own URL. When unset, fall back to whatever - # BWS_SERVER_URL the caller already had in their shell env (preserved by - # the copy above) so manual overrides keep working too. + # Empty server_url keeps whatever BWS_SERVER_URL the shell already had. if server_url: env["BWS_SERVER_URL"] = server_url - try: - proc = subprocess.run( # noqa: S603 — bws path is trusted - cmd, - env=env, - capture_output=True, - text=True, encoding='utf-8', errors='replace', - timeout=_BWS_RUN_TIMEOUT, - stdin=subprocess.DEVNULL, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError( - f"bws timed out after {_BWS_RUN_TIMEOUT}s fetching secrets" - ) from exc - except OSError as exc: - raise RuntimeError(f"failed to invoke bws: {exc}") from exc + proc = run_cli(cmd, env=env, timeout=_BWS_RUN_TIMEOUT, label="bws", + timeout_message=f"bws timed out after {_BWS_RUN_TIMEOUT}s fetching secrets") if proc.returncode != 0: - # bws writes auth/network errors to stderr as a Rust error-report - # dump (color-eyre): an "Error:" header, indented cause lines, then - # "Location:" / "Backtrace omitted" noise. Strip ANSI and boil it - # down to the meaningful cause line(s) before surfacing. err = _summarize_bws_stderr(proc.stderr or proc.stdout or "") raise RuntimeError( f"bws exited {proc.returncode}: {err[:200]}" @@ -753,135 +568,28 @@ def _run_bws_list( return secrets, warnings -# --------------------------------------------------------------------------- -# Public entry point — called from hermes_cli.env_loader -# --------------------------------------------------------------------------- - - -def apply_bitwarden_secrets( - *, - enabled: bool, - access_token_env: str = "BWS_ACCESS_TOKEN", - project_id: str = "", - override_existing: bool = False, - cache_ttl_seconds: float = 300, - auto_install: bool = True, - server_url: str = "", - home_path: Optional[Path] = None, - encrypted_cache_enabled: bool = False, - encrypted_cache_max_stale_seconds: float = 0, -) -> FetchResult: - """Pull secrets from BSM and set them on ``os.environ``. - - This is the function ``load_hermes_dotenv()`` calls after the .env - files have loaded. It is intentionally defensive — any failure - returns a :class:`FetchResult` with ``error`` set; it never raises. - - ``server_url`` selects the Bitwarden region or self-hosted endpoint - (e.g. ``https://vault.bitwarden.eu`` for EU Cloud). Empty string - means use ``bws``'s default (US Cloud). - - Parameters mirror the ``secrets.bitwarden.*`` config keys so the - caller can just splat the dict in. - """ - result = FetchResult() - - if not enabled: - return result - - access_token = os.environ.get(access_token_env, "").strip() - if not access_token: - result.error = ( - f"secrets.bitwarden.enabled is true but {access_token_env} is " - "not set. Run `hermes secrets bitwarden setup`." - ) - return result - - if not project_id: - result.error = ( - "secrets.bitwarden.project_id is empty. " - "Run `hermes secrets bitwarden setup`." - ) - return result - - binary = find_bws(install_if_missing=auto_install) - result.binary_path = binary - if binary is None: - result.error = ( - "bws binary not available and auto-install is disabled. " - "Run `hermes secrets bitwarden setup` to install." - ) - return result - - try: - secrets, warnings = fetch_bitwarden_secrets( - access_token=access_token, - project_id=project_id, - binary=binary, - cache_ttl_seconds=cache_ttl_seconds, - server_url=server_url, - home_path=home_path, - encrypted_cache_enabled=encrypted_cache_enabled, - encrypted_cache_max_stale_seconds=encrypted_cache_max_stale_seconds, - ) - except RuntimeError as exc: - result.error = str(exc) - return result - - result.secrets = secrets - result.warnings.extend(warnings) - - for key, value in secrets.items(): - if key == access_token_env: - # Don't let BSM clobber the very token we used to fetch - # itself — that would be a footgun if someone stored the - # token as a BSM secret too. - result.skipped.append(key) - continue - if not override_existing and os.environ.get(key): - result.skipped.append(key) - continue - os.environ[key] = value - result.applied.append(key) - - return result - - # --------------------------------------------------------------------------- # SecretSource adapter — the registry-facing wrapper around this module. # --------------------------------------------------------------------------- class BitwardenSource(SecretSource): - """Bitwarden Secrets Manager as a registered secret source. + """Bitwarden Secrets Manager as a registered **bulk** source. - Thin adapter over the module's fetch machinery. ``fetch()`` only - *fetches* — precedence, override semantics, conflict warnings, and - the ``os.environ`` writes are the orchestrator's job - (see ``agent.secret_sources.registry.apply_all``). - - Bitwarden is a **bulk** source: it injects every secret in the - configured BSM project, so explicit per-var bindings from mapped - sources (e.g. the 1Password ``env:`` map) outrank it. + ``fetch()`` only fetches — precedence, overrides and the ``os.environ`` + writes are the orchestrator's. Bulk: it injects every secret in the BSM + project, so explicit per-var bindings from mapped sources outrank it. """ name = "bitwarden" label = "Bitwarden Secrets Manager" shape = "bulk" scheme = "bws" - - def override_existing(self, cfg: dict) -> bool: - # Default True (matches DEFAULT_CONFIG): the point of BSM is - # centralized rotation — if .env had the final say, rotating a - # key in Bitwarden wouldn't take effect until the stale .env - # line was also deleted. - return bool(isinstance(cfg, dict) and cfg.get("override_existing", True)) - - def protected_env_vars(self, cfg: dict): - token_env = "BWS_ACCESS_TOKEN" - if isinstance(cfg, dict): - token_env = str(cfg.get("access_token_env") or token_env) - return frozenset({token_env}) + token_env_key = "access_token_env" + default_token_env = "BWS_ACCESS_TOKEN" + # override_existing defaults True: the point of BSM is centralized rotation + # — a stale .env line must not have the final say. + override_existing_default = True def config_schema(self) -> dict: return { @@ -897,89 +605,62 @@ class BitwardenSource(SecretSource): }, "encrypted_cache": { "description": "Encrypted last-good cache for network/timeout fallback", - "default": { - "enabled": False, - "max_stale_seconds": 0, - }, - }, - "override_existing": { - "description": "BSM values overwrite .env/shell values", - "default": True, - }, - "auto_install": { - "description": "Auto-download the pinned bws binary", - "default": True, - }, - "server_url": { - "description": "Region / self-hosted endpoint (empty = US Cloud)", - "default": "", + "default": {"enabled": False, "max_stale_seconds": 0}, }, + "override_existing": {"description": "BSM values overwrite .env/shell values", "default": True}, + "auto_install": {"description": "Auto-download the pinned bws binary", "default": True}, + "server_url": {"description": "Region / self-hosted endpoint (empty = US Cloud)", "default": ""}, } def fetch(self, cfg: dict, home_path: Path) -> FetchResult: cfg = cfg if isinstance(cfg, dict) else {} result = FetchResult() - access_token_env = str(cfg.get("access_token_env") or "BWS_ACCESS_TOKEN") + access_token_env = self.token_env(cfg) access_token = get_source_environment().get(access_token_env, "").strip() if not access_token: - result.error = ( + return result.fail( f"secrets.bitwarden.enabled is true but {access_token_env} is " - "not set. Run `hermes secrets bitwarden setup`." + "not set. Run `hermes secrets bitwarden setup`.", + ErrorKind.NOT_CONFIGURED, ) - result.error_kind = ErrorKind.NOT_CONFIGURED - return result project_id = str(cfg.get("project_id") or "") if not project_id: - result.error = ( + return result.fail( "secrets.bitwarden.project_id is empty. " - "Run `hermes secrets bitwarden setup`." + "Run `hermes secrets bitwarden setup`.", + ErrorKind.NOT_CONFIGURED, ) - result.error_kind = ErrorKind.NOT_CONFIGURED - return result - auto_install = bool(cfg.get("auto_install", True)) - binary = find_bws(install_if_missing=auto_install) + binary = find_bws(install_if_missing=bool(cfg.get("auto_install", True))) result.binary_path = binary if binary is None: - result.error = ( + return result.fail( "bws binary not available and auto-install is disabled. " - "Run `hermes secrets bitwarden setup` to install." + "Run `hermes secrets bitwarden setup` to install.", + ErrorKind.BINARY_MISSING, ) - result.error_kind = ErrorKind.BINARY_MISSING - return result - - try: - ttl = float(cfg.get("cache_ttl_seconds", 300)) - except (TypeError, ValueError): - ttl = 300.0 encrypted_cfg = cfg.get("encrypted_cache") encrypted_cfg = encrypted_cfg if isinstance(encrypted_cfg, dict) else {} - encrypted_enabled = bool(encrypted_cfg.get("enabled", False)) - try: - encrypted_max_stale = float(encrypted_cfg.get("max_stale_seconds", 0)) - except (TypeError, ValueError): - encrypted_max_stale = 0.0 try: secrets, warnings = fetch_bitwarden_secrets( access_token=access_token, project_id=project_id, binary=binary, - cache_ttl_seconds=ttl, + cache_ttl_seconds=coerce_float(cfg.get("cache_ttl_seconds", 300), 300.0), server_url=str(cfg.get("server_url", "") or "").strip(), home_path=home_path, - encrypted_cache_enabled=encrypted_enabled, - encrypted_cache_max_stale_seconds=encrypted_max_stale, + encrypted_cache_enabled=bool(encrypted_cfg.get("enabled", False)), + encrypted_cache_max_stale_seconds=coerce_float( + encrypted_cfg.get("max_stale_seconds", 0), 0.0), ) except RuntimeError as exc: - result.error = str(exc) - result.error_kind = _classify_bws_error(str(exc)) + result.fail(str(exc), _classify_bws_error(str(exc))) if result.error_kind == ErrorKind.AUTH_FAILED: - # Translate the raw OAuth reject into what it actually means - # for the user before the mechanics. + # Say what the raw OAuth reject means for the user first. result.error = ( "Bitwarden rejected the machine-account access token " f"({access_token_env}) — it was likely revoked, expired, " @@ -1002,54 +683,33 @@ class BitwardenSource(SecretSource): return super().remediation(kind, cfg) +# First matching rule wins. The BSM identity endpoint rejects a revoked / +# expired machine-account token with an OAuth-style +# `[400 Bad Request] {"error":"invalid_client"}`, hence those AUTH tokens. +_BWS_ERROR_RULES = ( + (ErrorKind.TIMEOUT, ("timed out",)), + (ErrorKind.BINARY_MISSING, ("binary not available", "failed to invoke")), + (ErrorKind.AUTH_FAILED, ("unauthorized", "invalid token", "access token", "401", "403", + "invalid_client", "invalid_grant", "400 bad request")), + (ErrorKind.NETWORK, ("network", "connection", "resolve", "download", "dns")), +) + + def _classify_bws_error(message: str) -> ErrorKind: - """Best-effort mapping of bws failure text onto the shared taxonomy.""" - lowered = message.lower() - if "timed out" in lowered: - return ErrorKind.TIMEOUT - if "binary not available" in lowered or "failed to invoke" in lowered: - return ErrorKind.BINARY_MISSING - if any(tok in lowered for tok in ("unauthorized", "invalid token", - "access token", "401", "403", - # The BSM identity endpoint rejects a - # revoked/expired/deleted machine-account - # token with an OAuth-style - # `[400 Bad Request] {"error":"invalid_client"}`. - "invalid_client", "invalid_grant", - "400 bad request")): - return ErrorKind.AUTH_FAILED - if any(tok in lowered for tok in ("network", "connection", "resolve", - "download", "dns")): - return ErrorKind.NETWORK - return ErrorKind.INTERNAL - - -# --------------------------------------------------------------------------- -# Test hook — used by hermetic tests to flush the cache between cases. -# --------------------------------------------------------------------------- + return classify_cli_error(message, _BWS_ERROR_RULES) def clear_caches(home_path: Optional[Path] = None) -> None: """Drop in-process AND disk caches (plaintext and encrypted). - Used after a token rotation (`hermes secrets bitwarden token`) so the - next startup fetches fresh with the new credential instead of serving - a pull cached under the old token's fingerprint. The encrypted cache - is keyed off the old token too, so it must go as well. + Used after a token rotation so the next startup fetches fresh instead of + serving a pull cached under the old token's fingerprint. """ - _CACHE.clear() - _DISK_CACHE.clear(home_path) + _STORE.clear(home_path) try: _encrypted_disk_cache_path(home_path).unlink() except (FileNotFoundError, OSError): pass -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 cache - writer itself. - """ - clear_caches(home_path) +_reset_cache_for_tests = clear_caches diff --git a/agent/secret_sources/command.py b/agent/secret_sources/command.py index 0831648418..c8d3ee7ce7 100644 --- a/agent/secret_sources/command.py +++ b/agent/secret_sources/command.py @@ -1,33 +1,25 @@ """``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. +``CommandSecretsProvider`` (``src/main/secrets/commandProvider.ts``). The +helper command (``keepassxc-cli``, ``secret-tool``, 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): +Security model: -* 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. +* The command string is the USER'S OWN configuration (same trust level as the + ``.env`` they control), so it runs via ``/bin/sh -c ``. +* The requested key reaches the child ONLY via ``HERMES_SECRET_KEY`` — never + interpolated into the shell string, so a hostile key name is inert data. +* Hard timeout (default 3s) + output cap (1 MiB); every failure (non-zero exit, + timeout, spawn failure, oversized output) degrades to "no value", never raises. +* Failure logs carry ONLY structured fields (exit code / signal / errno) — never + the command string, the helper's stderr (captured and DISCARDED, it can carry + secret material) or any value. +* Startup runs the helper exactly ONCE with an empty ``HERMES_SECRET_KEY`` — + never per key — so a helper blocked on a vault prompt isn't spawned N times. +* POSIX-only (needs ``/bin/sh``); on Windows it degrades to an empty result. """ from __future__ import annotations @@ -41,33 +33,25 @@ 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.base import ErrorKind, SecretSource -from agent.secret_sources.base import get_source_environment -from agent.secret_sources.bitwarden import FetchResult +from agent.secret_sources.base import ( + ErrorKind, + FetchResult, + SecretSource, + coerce_float, + source_child_env, +) __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. +# TIGHT on purpose: a helper MUST be fast and NON-INTERACTIVE (an already +# unlocked DB, `secret-tool lookup`, `cat` of a tmpfs file) — not a PIN prompt. _COMMAND_TIMEOUT_SECONDS = 3.0 -# Defensive cap on helper output (1 MiB) — a misbehaving command can't OOM us. -_MAX_OUTPUT_BYTES = 1024 * 1024 +_MAX_OUTPUT_BYTES = 1024 * 1024 # a misbehaving helper can't OOM us -# 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. +# Anchored; `.` does not cross newlines, so a multi-line blob never matches. _ENV_LINE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)=(.*)$") @@ -75,13 +59,15 @@ 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. +def _log(message: str) -> None: + print(f"[secrets:command] {message}", file=sys.stderr) - 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. + +def unquote_dotenv_value(raw: str) -> str: + """Strip one layer of matching surrounding quotes from a dotenv value. + + Requires length >= 2 so a lone ``"`` stays intact while ``""``/``''`` + correctly yield an empty string. """ t = raw.strip() if len(t) >= 2 and ( @@ -92,73 +78,6 @@ def unquote_dotenv_value(raw: str) -> str: 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, @@ -167,32 +86,17 @@ def _run_helper( ) -> 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. + The key travels as DATA in ``HERMES_SECRET_KEY``. stdout/stderr are piped + (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, - ) + _log("the 'command' provider is POSIX-only (needs /bin/sh); resolving no value on Windows") return None - # User-configured secret-helper command: runs with the user's full shell - # env by design (it may need any credential to resolve the secret). - source_env = get_source_environment() - if source_env is os.environ: - # Legacy single-profile startup intentionally preserves the existing - # helper contract, which may rely on the user's full environment. - from tools.environments.local import build_subprocess_env - env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False) - else: - # A multiplex profile must never inherit sibling secrets from the - # process-global environment. hydrate_profile_secret_sources seeds - # only global-safe values plus this profile's own .env. - env = dict(source_env) + # The helper legitimately gets the user's shell env (it may need any + # credential to resolve the secret) — but a multiplex profile only its own. + env = source_child_env() env["HERMES_SECRET_KEY"] = secret_key try: @@ -205,20 +109,14 @@ def _run_helper( 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, - ) + _log(f"helper failed to spawn; resolving no value: errno={exc.errno}") 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). - # POSIX-only by construction: _run_helper early-returns on Windows - # before ever spawning, so this line can't execute there. + # Kill the whole group: a helper may have forked children that would + # otherwise keep the pipe open. POSIX-only by the early return above. try: os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) # windows-footgun: ok except (ProcessLookupError, PermissionError, OSError): @@ -227,16 +125,10 @@ def _run_helper( 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, - ) + _log(f"helper timed out after {timeout_seconds:g}s; resolving no value") 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 @@ -245,176 +137,40 @@ def _run_helper( 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, - ) + _log(f"helper failed; resolving no value: code={code} signal={signame}") 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, - ) + _log(f"helper output exceeded the {max_output_bytes}-byte cap; resolving no value") 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. - """ + """Parse a KEY=VALUE blob; comments and non-env-shaped lines are skipped.""" 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)) + if m: + 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``. - - LEGACY shim retained for API symmetry with ``apply_bitwarden_secrets``; - the startup path goes through :class:`CommandSource` + the registry - orchestrator instead (which owns precedence and the environ writes). - """ - result = FetchResult() - - command = (command or "").strip() - if not command: - result.error = ( - "secrets.command.enabled is true but secrets.command.command is " - "empty. Set the helper command in config.yaml." - ) - return result - - if _is_windows(): - result.warnings.append( - "the 'command' secret source is POSIX-only (needs /bin/sh); " - "skipping on Windows" - ) - 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 - - -# --------------------------------------------------------------------------- -# SecretSource adapter — the registry-facing wrapper around this module. -# --------------------------------------------------------------------------- - - class CommandSource(SecretSource): - """User-configured helper command as a registered secret source. + """User-configured helper command as a registered **bulk** source. - Composes with the other sources (Bitwarden, 1Password, plugins) through - the ``apply_all()`` orchestrator — enable any combination simultaneously; - there is deliberately NO single-provider selector. ``fetch()`` only - fetches: precedence, ``override_existing`` semantics, conflict warnings, - and the ``os.environ`` writes are the orchestrator's job. - - Bulk shape: the helper enumerates a KEY=VALUE blob in one run. Config:: + Composes with the other sources through ``apply_all()``; there is + deliberately NO single-provider selector. The helper enumerates a + KEY=VALUE blob in one run. Config:: secrets: command: enabled: true command: "cat /run/user/1000/hermes-secrets.env" - # or per-vault CLIs: keepassxc-cli / secret-tool / pass / gpg — - # anything fast and NON-interactive. """ name = "command" @@ -445,36 +201,29 @@ class CommandSource(SecretSource): command = str(cfg.get("command") or "").strip() if not command: - result.error = ( + return result.fail( "secrets.command.enabled is true but secrets.command.command " - "is empty. Set the helper command in config.yaml." + "is empty. Set the helper command in config.yaml.", + ErrorKind.NOT_CONFIGURED, ) - result.error_kind = ErrorKind.NOT_CONFIGURED - return result if _is_windows(): - result.error = ( + return result.fail( "the 'command' secret source is POSIX-only (needs /bin/sh); " - "skipping on Windows" + "skipping on Windows", + ErrorKind.NOT_CONFIGURED, ) - result.error_kind = ErrorKind.NOT_CONFIGURED - return result - try: - timeout = float(cfg.get("helper_timeout_seconds", - _COMMAND_TIMEOUT_SECONDS)) - except (TypeError, ValueError): - timeout = _COMMAND_TIMEOUT_SECONDS + timeout = coerce_float(cfg.get("helper_timeout_seconds", _COMMAND_TIMEOUT_SECONDS), + _COMMAND_TIMEOUT_SECONDS) stdout = _run_helper(command, "", timeout, _MAX_OUTPUT_BYTES) - if stdout is None: - # _run_helper already logged structured fields to stderr. - result.error = ( + if stdout is None: # _run_helper already logged structured fields + return result.fail( "helper command failed (see structured fields above); " - "no secrets applied" + "no secrets applied", + ErrorKind.INTERNAL, ) - result.error_kind = ErrorKind.INTERNAL - return result secrets = _parse_dotenv_map(stdout) if not secrets: diff --git a/agent/secret_sources/onepassword.py b/agent/secret_sources/onepassword.py index 3aef02df5b..864c6aa9aa 100644 --- a/agent/secret_sources/onepassword.py +++ b/agent/secret_sources/onepassword.py @@ -1,40 +1,16 @@ """1Password (`op` CLI) secret source. -Resolve provider credentials from 1Password ``op://vault/item/field`` -references at process startup so they don't have to live in plaintext in -``~/.hermes/.env``. +Users map env-var names to official ``op://vault/item/field`` references in +``secrets.onepassword.env``; after ``.env`` loads each reference is resolved +with one ``op read -- `` call. Authentication is whatever the +user's ``op`` CLI already uses (``OP_SERVICE_ACCOUNT_TOKEN`` for headless +boxes, ``OP_SESSION_*`` for interactive sessions) — Hermes never authenticates +on the user's behalf. Failures NEVER block startup. -Design summary --------------- - -* Users map environment-variable names to official 1Password secret - references in ``secrets.onepassword.env``:: - - secrets: - onepassword: - enabled: true - env: - OPENAI_API_KEY: "op://Private/OpenAI/api key" - ANTHROPIC_API_KEY: "op://Private/Anthropic/credential" - -* After ``.env`` loads, each reference is resolved with a single - ``op read -- `` call and injected into ``os.environ`` (the - same point in startup as the Bitwarden source). -* Authentication is whatever the user's ``op`` CLI already uses — a - service-account token (``OP_SERVICE_ACCOUNT_TOKEN``) for headless boxes, - or a desktop/interactive session (``OP_SESSION_*``). Hermes never - authenticates on the user's behalf; it shells out to an already-trusted, - already-authenticated CLI. -* Failures NEVER block startup. A missing ``op`` binary, expired auth, a - bad reference, or a permission error each surface a one-line warning and - Hermes continues with whatever credentials ``.env`` already had. - -The atomic-write / ``0600`` / TTL cache mechanics are shared with the other -backends via :mod:`agent.secret_sources._cache` — successful, complete pulls -are cached in-process and on disk under ``/cache/op_cache.json`` -so back-to-back short-lived ``hermes`` invocations don't re-shell ``op`` for -every reference. The disk file holds only resolved secret *values*; auth -material is fingerprinted, never stored. +Successful, complete pulls are cached in-process and under +``/cache/op_cache.json`` (values only; auth material is +fingerprinted, never stored) so back-to-back ``hermes`` invocations don't +re-shell ``op`` for every reference. """ from __future__ import annotations @@ -43,61 +19,38 @@ import hashlib import logging import os import shutil -import subprocess +import subprocess # noqa: F401 — tests monkeypatch ``op.subprocess.run`` import time from pathlib import Path from typing import Dict, List, Optional, Tuple -from agent.secret_sources._cache import ( - CachedFetch, - DiskCache, +from agent.secret_sources._cache import CachedFetch, SecretCache +from agent.secret_sources.base import ( + ErrorKind, FetchResult, + SecretSource, + classify_cli_error, + coerce_float, + get_source_environment, is_valid_env_name, + run_cli, ) -from agent.secret_sources.base import ErrorKind, SecretSource -from agent.secret_sources.base import get_source_environment logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Configuration constants -# --------------------------------------------------------------------------- - -# How long to wait for a single `op read`, in seconds. _OP_RUN_TIMEOUT = 30 -# Default env var the official `op` CLI reads for service-account auth. Users -# can point `service_account_token_env` at a different name; we always export -# the value to the child as OP_SERVICE_ACCOUNT_TOKEN, which is what `op` itself -# looks for. +# `op` itself reads OP_SERVICE_ACCOUNT_TOKEN; `service_account_token_env` lets +# the user source it from another name, and _op_child_env normalizes it back. _DEFAULT_TOKEN_ENV = "OP_SERVICE_ACCOUNT_TOKEN" -# ANSI stripping for `op` diagnostics we surface uses the shared -# tools.ansi_strip.strip_ansi (full ECMA-48: CSI, OSC, DCS/SOS/PM/APC, -# C1) so a control sequence can't reposition the cursor or hide text -# after a redaction marker. - -# Env vars the `op` 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 `op` or -# anything it execs ever misbehaves. OP_SESSION_* and the token are added +# Minimal allowlisted child env (never the full post-dotenv os.environ, which +# holds every provider credential). OP_SESSION_* and the token are added # dynamically in _op_child_env(). _OP_ENV_ALLOWLIST = ( - "PATH", - "HOME", - "USERPROFILE", - "APPDATA", - "LOCALAPPDATA", - "SystemRoot", - "TMPDIR", - "TMP", - "TEMP", - "XDG_CONFIG_HOME", - "XDG_RUNTIME_DIR", - "OP_ACCOUNT", - "OP_CONNECT_HOST", - "OP_CONNECT_TOKEN", + "PATH", "HOME", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "SystemRoot", + "TMPDIR", "TMP", "TEMP", "XDG_CONFIG_HOME", "XDG_RUNTIME_DIR", + "OP_ACCOUNT", "OP_CONNECT_HOST", "OP_CONNECT_TOKEN", # Lets a user skip op's desktop-app integration probe (which can hang with # no timeout on a wedged desktop container) and go straight to token auth. "OP_LOAD_DESKTOP_APP_SETTINGS", @@ -108,35 +61,20 @@ _OP_ENV_ALLOWLIST = ( # 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. The disk layer omits home from its serialized -# key because the file already lives under the home dir (see _disk_key_str). +# L1 key folds in str(home_path) so a HERMES_HOME switch inside one long-lived +# process (the gateway) can't return another profile's secrets. The disk key +# omits home because the file already lives under /cache/. _CacheKey = Tuple[str, str, str, str] # (auth_fp, account, home, refs_fp) -_CACHE: Dict[_CacheKey, CachedFetch] = {} - _DISK_CACHE_BASENAME = "op_cache.json" 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; folding it - into the key string too would be redundant. - """ auth_fp, account, _home, refs_fp = cache_key return f"{auth_fp}|{account}|{refs_fp}" -_DISK_CACHE: DiskCache = DiskCache( - _DISK_CACHE_BASENAME, key_serializer=_disk_key_str -) - - -def _disk_cache_path(home_path: Optional[Path] = None) -> Path: - """Path to the on-disk cache (exposed for tests and direct callers).""" - return _DISK_CACHE.path(home_path) +_STORE: SecretCache[_CacheKey] = SecretCache(_DISK_CACHE_BASENAME, key_serializer=_disk_key_str) +_CACHE = _STORE.memory # tests flush L1 directly # --------------------------------------------------------------------------- @@ -147,12 +85,7 @@ def _disk_cache_path(home_path: Optional[Path] = None) -> Path: 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 ``op://…`` reference string. Everything - else produces a warning and is dropped (never fatal). - """ + """``(valid_refs, warnings)``: keep valid env names bound to stripped ``op://`` strings.""" valid: Dict[str, str] = {} warnings: List[str] = [] for name, ref in (references or {}).items(): @@ -172,16 +105,16 @@ def _validate_references( return valid, warnings -def _auth_fingerprint(token_env: str) -> str: - """SHA-256 prefix over the auth material `op` would use. +def _fingerprint(material: str) -> str: + return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16] - Folds in the service-account token, ``OP_ACCOUNT``, the 1Password Connect - ``OP_CONNECT_HOST``/``OP_CONNECT_TOKEN``, and *all* ``OP_SESSION_*`` vars - (the names `op` actually exports for interactive sessions — - ``OP_SESSION_``). Signing out and into a different - identity therefore changes the cache key, so a value cached under a - previous identity is never served under a new one. Never logged or - displayed; the raw token never leaves this hash. + +def _auth_fingerprint(token_env: str) -> str: + """SHA-256 prefix over everything `op` would authenticate with. + + Folds in the service-account token, OP_ACCOUNT, Connect host/token and all + ``OP_SESSION_*`` vars, so signing into a different identity changes the + cache key and a value cached under the old identity is never served. """ source_env = get_source_environment() parts: List[str] = [ @@ -193,28 +126,23 @@ def _auth_fingerprint(token_env: str) -> str: for key in sorted(source_env): if key.startswith("OP_SESSION_"): parts.append(f"{key}={source_env[key]}") - material = "\n".join(parts) - return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16] + return _fingerprint("\n".join(parts)) 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] + return _fingerprint("\n".join(f"{name}={references[name]}" for name in sorted(references))) # --------------------------------------------------------------------------- -# Binary discovery +# Binary discovery + `op read` # --------------------------------------------------------------------------- def find_op(binary_path: str = "") -> Optional[Path]: """Resolve a usable ``op`` 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 ``op`` shows - up first on ``PATH``. A pinned-but-missing path returns None (the caller - surfaces a clear error) rather than silently falling back. + A pinned ``binary_path`` is used verbatim (PATH is NOT consulted) and a + pinned-but-missing path returns None rather than silently falling back. """ if binary_path: pinned = Path(binary_path) @@ -225,33 +153,17 @@ def find_op(binary_path: str = "") -> Optional[Path]: return Path(found) if found else None -# --------------------------------------------------------------------------- -# `op read` invocation -# --------------------------------------------------------------------------- - - def _scrub(text: str) -> str: - """Remove ANSI control sequences and trim, for safe message surfacing.""" + """Full ECMA-48 ANSI strip (so a control sequence can't hide text after a redaction marker) + trim.""" from tools.ansi_strip import strip_ansi - # strip_ansi removes well-formed sequences; drop any stray lone ESC too. return strip_ansi(text).replace("\x1b", "").strip() def _op_child_env(token_value: str) -> Dict[str, str]: - """Build a minimal allowlisted environment for the ``op`` child process.""" source_env = get_source_environment() - env: Dict[str, str] = {} - for key in _OP_ENV_ALLOWLIST: - val = source_env.get(key) - if val is not None: - env[key] = val - # Desktop / interactive session credentials. - for key, val in source_env.items(): - if key.startswith("OP_SESSION_"): - env[key] = val - # `op` reads OP_SERVICE_ACCOUNT_TOKEN regardless of which env var the user - # configured Hermes to source it from, so normalize to that name here. + env = {k: source_env[k] for k in _OP_ENV_ALLOWLIST if k in source_env} + env.update((k, v) for k, v in source_env.items() if k.startswith("OP_SESSION_")) if token_value: env["OP_SERVICE_ACCOUNT_TOKEN"] = token_value env["NO_COLOR"] = "1" @@ -265,35 +177,21 @@ def _run_op_read( account: str = "", token_value: str = "", ) -> str: - """Resolve a single ``op://`` reference to its value. + """Resolve one ``op://`` reference; raises ``RuntimeError`` on any failure. - Raises :class:`RuntimeError` on any failure — including a ``returncode 0`` - with empty output, which would otherwise silently clobber a good - ``.env``/shell credential with ``""``. + An exit-0 empty/whitespace-only value is a failure too — applying it would + silently clobber a good .env/shell credential with ``""``. """ cmd: List[str] = [str(op), "read"] if account: cmd += ["--account", account] - # `--` terminates option parsing so a reference can never be mis-parsed as - # an `op` flag even if validation is ever loosened. - cmd += ["--", reference] + cmd += ["--", reference] # `--` so a reference can never parse as an op flag - try: - proc = subprocess.run( # noqa: S603 — op path is user-trusted, argv list - cmd, - env=_op_child_env(token_value), - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=_OP_RUN_TIMEOUT, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError( - f"op read timed out after {_OP_RUN_TIMEOUT}s for {reference!r}" - ) from exc - except OSError as exc: - raise RuntimeError(f"failed to invoke op: {exc}") from exc + proc = run_cli( + cmd, env=_op_child_env(token_value), timeout=_OP_RUN_TIMEOUT, label="op", + timeout_message=f"op read timed out after {_OP_RUN_TIMEOUT}s for {reference!r}", + stdin=None, + ) if proc.returncode != 0: err = _scrub(proc.stderr or "")[:200] @@ -303,10 +201,7 @@ def _run_op_read( f"op read exited {proc.returncode} for {reference!r}" ) - # `op` appends a trailing newline; strip only that so a value with - # intentional internal/edge spaces survives. But a value that is empty or - # whitespace-only is treated as empty: applying it would silently clobber a - # good .env/shell credential with effectively nothing. + # Strip only op's trailing newline so intentional edge spaces survive. value = (proc.stdout or "").rstrip("\r\n") if not value.strip(): raise RuntimeError(f"op read returned an empty value for {reference!r}") @@ -331,13 +226,10 @@ def fetch_onepassword_secrets( ) -> Tuple[Dict[str, str], List[str]]: """Resolve ``references`` (name → ``op://…``) to ``(secrets, warnings)``. - Raises :class:`RuntimeError` only when no ``op`` binary is available — a - fatal "can't fetch anything" condition. Per-reference failures (expired - auth, bad reference, empty value) are collected as warnings and the - reference is dropped, so one bad entry never sinks the rest. - - Only a complete, error-free pull is cached, so a transient auth failure - isn't frozen in for the whole TTL window. + Raises ``RuntimeError`` only when no ``op`` binary is available. Per-ref + failures become warnings and the ref is dropped, so one bad entry never + sinks the rest. 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: @@ -352,14 +244,9 @@ def fetch_onepassword_secrets( ) if use_cache: - cached = _CACHE.get(cache_key) - if cached and cached.is_fresh(cache_ttl_seconds): + cached = _STORE.lookup(cache_key, cache_ttl_seconds, home_path) + if cached is not None: return dict(cached.secrets), warnings - disk_cached = _DISK_CACHE.read(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 op = binary or find_op(binary_path) if op is None: @@ -382,14 +269,27 @@ def fetch_onepassword_secrets( if use_cache and not read_errors and secrets: entry = CachedFetch(secrets=dict(secrets), fetched_at=time.time()) - _CACHE[cache_key] = entry - _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path) + _STORE.store(cache_key, entry, cache_ttl_seconds, home_path) return secrets, warnings +def _missing_binary_error(binary_path: str) -> str: + if binary_path: + return ( + f"secrets.onepassword.binary_path ({binary_path!r}) is not an " + "executable op binary." + ) + return ( + "secrets.onepassword.enabled is true but the op CLI was not " + "found on PATH. Install it " + "(https://developer.1password.com/docs/cli/get-started/) or set " + "secrets.onepassword.binary_path." + ) + + # --------------------------------------------------------------------------- -# Public entry point — called from hermes_cli.env_loader +# Public entry point — used by `hermes secrets onepassword sync --apply` # --------------------------------------------------------------------------- @@ -406,13 +306,8 @@ def apply_onepassword_secrets( ) -> FetchResult: """Resolve configured ``op://`` 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.onepassword.*`` config keys so the caller - can splat the dict in. References that are already satisfied by the - current environment (when ``override_existing`` is false) are skipped + Never raises. References already satisfied by the environment (when + ``override_existing`` is false) and the token var itself are skipped *before* fetching, so ``op`` is never invoked for a value that would be discarded. """ @@ -424,17 +319,18 @@ def apply_onepassword_secrets( valid, warnings = _validate_references(env) result.warnings.extend(warnings) - # Skip-before-fetch: never resolve a reference we'd only throw away. + def _guarded(name: str) -> bool: + """True when ``name`` must not be applied (token var or env already set).""" + return name == service_account_token_env or ( + not override_existing and bool(os.environ.get(name)) + ) + refs_to_fetch: Dict[str, str] = {} for name, ref in valid.items(): - if name == service_account_token_env: - # Never let a resolved secret clobber the very token used to auth. + if _guarded(name): result.skipped.append(name) - continue - if not override_existing and os.environ.get(name): - result.skipped.append(name) - continue - refs_to_fetch[name] = ref + else: + refs_to_fetch[name] = ref if not refs_to_fetch: return result @@ -442,18 +338,7 @@ def apply_onepassword_secrets( binary = find_op(binary_path) result.binary_path = binary if binary is None: - if binary_path: - result.error = ( - f"secrets.onepassword.binary_path ({binary_path!r}) is not an " - "executable op binary." - ) - else: - result.error = ( - "secrets.onepassword.enabled is true but the op CLI was not " - "found on PATH. Install it " - "(https://developer.1password.com/docs/cli/get-started/) or set " - "secrets.onepassword.binary_path." - ) + result.error = _missing_binary_error(binary_path) return result try: @@ -473,13 +358,8 @@ def apply_onepassword_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 == service_account_token_env: - if name not in result.skipped: - result.skipped.append(name) - continue - if not override_existing and os.environ.get(name): + # Defensive re-check: keys should already be ⊆ refs_to_fetch. + if _guarded(name): if name not in result.skipped: result.skipped.append(name) continue @@ -495,65 +375,36 @@ def apply_onepassword_secrets( class OnePasswordSource(SecretSource): - """1Password as a registered secret source. + """1Password as a registered **mapped** source. - Thin adapter over the module's fetch machinery. ``fetch()`` only - *fetches* — precedence, override semantics, conflict warnings, and - the ``os.environ`` writes are the orchestrator's job - (see ``agent.secret_sources.registry.apply_all``). - - 1Password is a **mapped** source: the user explicitly binds each env - var to an ``op://`` reference under ``secrets.onepassword.env``, so - its claims outrank bulk sources (e.g. a Bitwarden project dump) on - contested vars. + ``fetch()`` only fetches — precedence, overrides and the ``os.environ`` + writes are the orchestrator's. Mapped: the user explicitly binds each env + var to an ``op://`` ref, so its claims outrank bulk sources on contested vars. """ name = "onepassword" label = "1Password" shape = "mapped" scheme = "op" - - def override_existing(self, cfg: dict) -> bool: - # Default True: an explicit VAR→op:// binding is the strongest - # user intent there is — leaving a stale .env line in place - # should not silently defeat it (same rotation rationale as - # Bitwarden). - return bool(isinstance(cfg, dict) and cfg.get("override_existing", True)) - - def protected_env_vars(self, cfg: dict): - token_env = _DEFAULT_TOKEN_ENV - if isinstance(cfg, dict): - token_env = str(cfg.get("service_account_token_env") or token_env) - return frozenset({token_env}) + token_env_key = "service_account_token_env" + default_token_env = _DEFAULT_TOKEN_ENV + # override_existing defaults True: an explicit VAR→op:// binding is the + # strongest user intent; a stale .env line must not silently defeat it. + override_existing_default = True def config_schema(self) -> dict: return { "enabled": {"description": "Master switch", "default": False}, - "env": { - "description": "Map of ENV_VAR -> op://vault/item/field reference", - "default": {}, - }, - "account": { - "description": "op --account shorthand (empty = default account)", - "default": "", - }, + "env": {"description": "Map of ENV_VAR -> op://vault/item/field reference", "default": {}}, + "account": {"description": "op --account shorthand (empty = default account)", "default": ""}, "service_account_token_env": { "description": "Env var holding the service-account token " "(unset = desktop/interactive session)", "default": _DEFAULT_TOKEN_ENV, }, - "binary_path": { - "description": "Pin the op binary (empty = resolve via PATH)", - "default": "", - }, - "cache_ttl_seconds": { - "description": "Disk+memory cache TTL; 0 disables", - "default": 300, - }, - "override_existing": { - "description": "Resolved values overwrite .env/shell values", - "default": True, - }, + "binary_path": {"description": "Pin the op binary (empty = resolve via PATH)", "default": ""}, + "cache_ttl_seconds": {"description": "Disk+memory cache TTL; 0 disables", "default": 300}, + "override_existing": {"description": "Resolved values overwrite .env/shell values", "default": True}, } def fetch(self, cfg: dict, home_path: Path) -> FetchResult: @@ -567,52 +418,30 @@ class OnePasswordSource(SecretSource): result.warnings.extend(warnings) if not valid: if not warnings: - result.error = ( + result.fail( "secrets.onepassword.enabled is true but the env: map is " - "empty. Add ENV_VAR: op://vault/item/field entries." + "empty. Add ENV_VAR: op://vault/item/field entries.", + ErrorKind.NOT_CONFIGURED, ) - result.error_kind = ErrorKind.NOT_CONFIGURED return result binary_path = str(cfg.get("binary_path") or "") binary = find_op(binary_path) result.binary_path = binary if binary is None: - if binary_path: - result.error = ( - f"secrets.onepassword.binary_path ({binary_path!r}) is " - "not an executable op binary." - ) - else: - result.error = ( - "secrets.onepassword.enabled is true but the op CLI was " - "not found on PATH. Install it " - "(https://developer.1password.com/docs/cli/get-started/) " - "or set secrets.onepassword.binary_path." - ) - result.error_kind = ErrorKind.BINARY_MISSING - return result - - try: - ttl = float(cfg.get("cache_ttl_seconds", 300)) - except (TypeError, ValueError): - ttl = 300.0 + return result.fail(_missing_binary_error(binary_path), ErrorKind.BINARY_MISSING) try: secrets, fetch_warnings = fetch_onepassword_secrets( references=valid, account=str(cfg.get("account") or ""), - token_env=str( - cfg.get("service_account_token_env") or _DEFAULT_TOKEN_ENV - ), + token_env=self.token_env(cfg), binary=binary, - cache_ttl_seconds=ttl, + cache_ttl_seconds=coerce_float(cfg.get("cache_ttl_seconds", 300), 300.0), home_path=home_path, ) except RuntimeError as exc: - result.error = str(exc) - result.error_kind = _classify_op_error(str(exc)) - return result + return result.fail(str(exc), _classify_op_error(str(exc))) result.secrets = secrets result.warnings.extend(fetch_warnings) @@ -620,12 +449,9 @@ class OnePasswordSource(SecretSource): def remediation(self, kind, cfg: dict) -> str: if kind in (ErrorKind.AUTH_FAILED, ErrorKind.AUTH_EXPIRED): - token_env = _DEFAULT_TOKEN_ENV - if isinstance(cfg, dict): - token_env = str(cfg.get("service_account_token_env") or token_env) return ( "Run `hermes secrets onepassword token` to paste a fresh " - f"service-account token ({token_env}), or `op signin` for an " + f"service-account token ({self.token_env(cfg)}), or `op signin` for an " "interactive session." ) if kind == ErrorKind.BINARY_MISSING: @@ -637,46 +463,24 @@ class OnePasswordSource(SecretSource): return super().remediation(kind, cfg) +_OP_ERROR_RULES = ( + (ErrorKind.TIMEOUT, ("timed out",)), + (ErrorKind.BINARY_MISSING, ("not found on path", "not an executable", "failed to invoke")), + (ErrorKind.AUTH_FAILED, ("unauthorized", "not signed in", "session expired", + "authentication", "401", "403")), + (ErrorKind.EMPTY_VALUE, ("empty value",)), + (ErrorKind.NETWORK, ("network", "connection", "resolve host", "dns")), +) + + def _classify_op_error(message: str) -> ErrorKind: - """Best-effort mapping of op failure text onto the shared taxonomy.""" - lowered = message.lower() - if "timed out" in lowered: - return ErrorKind.TIMEOUT - if "not found on path" in lowered or "not an executable" in lowered \ - or "failed to invoke" in lowered: - return ErrorKind.BINARY_MISSING - if any(tok in lowered for tok in ("unauthorized", "not signed in", - "session expired", "authentication", - "401", "403")): - return ErrorKind.AUTH_FAILED - if "empty value" in lowered: - return ErrorKind.EMPTY_VALUE - if any(tok in lowered for tok in ("network", "connection", "resolve host", - "dns")): - return ErrorKind.NETWORK - return ErrorKind.INTERNAL - - -# --------------------------------------------------------------------------- -# Test hook — used by hermetic tests to flush the cache between cases. -# --------------------------------------------------------------------------- + return classify_cli_error(message, _OP_ERROR_RULES) def clear_caches(home_path: Optional[Path] = None) -> None: - """Drop in-process AND disk caches. - - Used after a token rotation (`hermes secrets onepassword token`) so - the next startup resolves fresh with the new credential instead of - serving values cached under the old token's fingerprint. - """ - _CACHE.clear() - _DISK_CACHE.clear(home_path) + """Drop in-process AND disk caches (after a token rotation, so the next + startup resolves fresh instead of serving values cached under the old token).""" + _STORE.clear(home_path) -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. - """ - clear_caches(home_path) +_reset_cache_for_tests = clear_caches diff --git a/tests/test_command_secret_source.py b/tests/test_command_secret_source.py index e3be911bbc..2cfd9e5c47 100644 --- a/tests/test_command_secret_source.py +++ b/tests/test_command_secret_source.py @@ -9,9 +9,6 @@ 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. @@ -35,11 +32,8 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from agent.secret_sources.command import ( # noqa: E402 + CommandSource, _run_helper, - apply_command_secrets, - get_command_secret, - list_command_secrets, - parse_secret_output, unquote_dotenv_value, ) from agent.secret_sources.base import ( # noqa: E402 @@ -106,54 +100,26 @@ def test_unquote_strips_one_layer_of_matching_quotes(): 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==" - - - - - - - - # --------------------------------------------------------------------------- # Real-subprocess resolution # --------------------------------------------------------------------------- -def test_bare_value_helper_resolves_single_value(tmp_path): +def test_helper_stdout_is_returned_verbatim(tmp_path): helper = _write_helper(tmp_path, "printf 'sk-test-bare-12345'") - value = get_command_secret(command=str(helper), key="CMDTEST_API_KEY") + value = _run_helper(str(helper), "CMDTEST_API_KEY", 3.0, 1024) assert value == "sk-test-bare-12345" - - - - - - - - - - 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 - ) + value = _run_helper(str(helper), "CMDTEST_API_KEY", 2.0, 1024) 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_failure_logging_never_leaks_command_or_secret(tmp_path, capfd): secret_value = "sk-super-secret-value-do-not-log" helper = _write_helper( @@ -161,7 +127,7 @@ def test_failure_logging_never_leaks_command_or_secret(tmp_path, capfd): f"echo '{secret_value}' >&2\nexit 7", name="my-distinctive-helper-name.sh", ) - value = get_command_secret(command=str(helper), key="CMDTEST_API_KEY") + value = _run_helper(str(helper), "CMDTEST_API_KEY", 3.0, 1024) assert value is None captured = capfd.readouterr() combined = captured.out + captured.err @@ -172,24 +138,14 @@ def test_failure_logging_never_leaks_command_or_secret(tmp_path, capfd): assert "code=7" in combined # the structured field IS logged - - - - -def test_apply_dotenv_blob_sets_environ(tmp_path): +def test_fetch_parses_dotenv_blob(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"] + result = CommandSource().fetch({"enabled": True, "command": str(helper)}, tmp_path) assert result.error is None - assert os.environ["CMDTEST_API_KEY"] == "sk-applied" - assert os.environ["CMDTEST_TOKEN"] == "tok-applied" - - - - + assert result.secrets == {"CMDTEST_API_KEY": "sk-applied", "CMDTEST_TOKEN": "tok-applied"} # --------------------------------------------------------------------------- @@ -245,8 +201,6 @@ def test_registry_status_line_printed_once_per_home(tmp_path, monkeypatch, capsy assert err.count("Command helper: applied 1 secret") == 1 - - def test_registry_failing_helper_does_not_block_startup(tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) (tmp_path / "config.yaml").write_text( @@ -258,8 +212,6 @@ def test_registry_failing_helper_does_not_block_startup(tmp_path, monkeypatch): assert env_loader.get_secret_source("CMDTEST_API_KEY") is None - - def test_registry_helper_error_prints_remediation(tmp_path, monkeypatch, capsys): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) (tmp_path / "config.yaml").write_text(