From d4f2933262c823afcd0262b6d75ff4d2a8627b3f 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): unify secret-source CLI/cache/error plumbing in base and _cache - base: run_cli (shared subprocess wrapper), classify_cli_error (rule tables), coerce_float, source_child_env, FetchResult.fail, SecretSource.token_env / token_env_key / default_token_env / override_existing_default so per-backend override_existing/protected_env_vars overrides collapse into the ABC; generic remediation is a kind->template table. - _cache: atomic_write_json (mkstemp->0600->replace) and entry_from_payload shared by DiskCache and the bws encrypted cache; SecretCache = L1 dict + L2 DiskCache with lookup/store/clear. --- agent/secret_sources/__init__.py | 37 +--- agent/secret_sources/_cache.py | 242 +++++++++++----------- agent/secret_sources/base.py | 341 ++++++++++++++----------------- 3 files changed, 290 insertions(+), 330 deletions(-) diff --git a/agent/secret_sources/__init__.py b/agent/secret_sources/__init__.py index 70343714ab..cfbca8154e 100644 --- a/agent/secret_sources/__init__.py +++ b/agent/secret_sources/__init__.py @@ -1,33 +1,16 @@ """External secret source integrations. -A secret source is anything that can supply environment-variable-shaped -credentials at process startup, _after_ ~/.hermes/.env has loaded. +A secret source supplies environment-variable-shaped credentials at process +startup, _after_ ~/.hermes/.env has loaded. The contract is +:class:`agent.secret_sources.base.SecretSource`; the orchestrator (ordering, +mapped-beats-bulk precedence, first-claim-wins, ``override_existing``, +provenance) is :func:`agent.secret_sources.registry.apply_all`. The +atomic-write / 0600 / TTL disk cache is shared in ``_cache``. -The contract every source implements is -:class:`agent.secret_sources.base.SecretSource`; the orchestrator that -runs the enabled sources (ordering, mapped-beats-bulk precedence, -first-claim-wins conflicts, ``override_existing`` semantics, provenance) -is :func:`agent.secret_sources.registry.apply_all`. Multiple sources -can be enabled at once — see the registry module docstring for the -precedence ladder. The atomic-write / 0600 / TTL disk-cache substrate -is shared across backends in ``agent.secret_sources._cache`` so the -security-sensitive bits live in exactly one place. - -Currently bundled: - - - ``bitwarden`` — Bitwarden Secrets Manager (`bws` CLI). See - ``agent.secret_sources.bitwarden`` for the integration and - ``hermes_cli.secrets_cli`` for the user-facing setup wizard. - - ``onepassword`` — 1Password ``op://`` secret references (`op` CLI). - See ``agent.secret_sources.onepassword`` for the integration and - ``hermes_cli.onepassword_secrets_cli`` for the user-facing commands. - -The bundled set is deliberately closed (policy mirrors memory -providers): new third-party secret managers ship as standalone plugin -repos that subclass ``SecretSource`` and register through -``PluginContext.register_secret_source()`` — they are NOT added to this -package. A generic ``command`` source is a possible future exception; -OS keystores (Keychain/DPAPI/libsecret) are under discussion. +Bundled: ``bitwarden`` (bws CLI), ``onepassword`` (op CLI), ``command`` (user +helper). The set is deliberately closed — new third-party managers ship as +standalone plugin repos that subclass ``SecretSource`` and register through +``PluginContext.register_secret_source()``. """ from agent.secret_sources.base import ( # noqa: F401 diff --git a/agent/secret_sources/_cache.py b/agent/secret_sources/_cache.py index 64e1e5e13c..c789cb7ed1 100644 --- a/agent/secret_sources/_cache.py +++ b/agent/secret_sources/_cache.py @@ -1,21 +1,14 @@ -"""Shared substrate for external secret-source backends. +"""Shared cache substrate for external secret-source backends. -Every backend (Bitwarden, 1Password, …) needs the same handful of -security-sensitive primitives: +Every backend needs the same security-sensitive primitives: a two-layer fetch +cache (in-process + on-disk) whose disk half writes atomically with ``0600`` +permissions and honours a TTL. Keeping them here means the atomic-write / +``0600`` / TTL logic is audited in exactly one place; each backend supplies +only its cache-key shape and a serializer for it. - * a uniform result object (:class:`FetchResult`), - * environment-variable name validation (:func:`is_valid_env_name`), - * a two-layer fetch cache whose disk half writes atomically with ``0600`` - permissions and honours a TTL (:class:`DiskCache`, :class:`CachedFetch`). - -These used to live inline inside ``bitwarden.py``. Pulling them here means -the atomic-write / ``0600`` / TTL logic is audited and fixed in exactly one -place instead of drifting across copy-pasted per-backend modules — each -backend supplies only its own cache-key shape and a serializer for it. - -Nothing in this module ever raises out to the caller's hot path: the disk -layer is strictly best-effort (a miss just triggers a refetch), because a -cache problem must never block Hermes startup. +Nothing here raises into the caller's hot path: the disk layer is strictly +best-effort (a miss just triggers a refetch) because a cache problem must +never block Hermes startup. """ from __future__ import annotations @@ -28,32 +21,20 @@ from dataclasses import dataclass from pathlib import Path from typing import Callable, Dict, Generic, Optional, TypeVar +from agent.secret_sources.base import FetchResult, is_valid_env_name # noqa: F401 — re-export + __all__ = [ "FetchResult", "CachedFetch", "DiskCache", + "SecretCache", + "atomic_write_json", + "entry_from_payload", "is_valid_env_name", "resolve_cache_home", ] -# --------------------------------------------------------------------------- -# Result object + env-name validation — canonical definitions live in -# ``agent.secret_sources.base`` (the SecretSource contract module); re-exported -# here so backends that import from ``_cache`` keep working. -# --------------------------------------------------------------------------- - -from agent.secret_sources.base import ( # noqa: E402 - FetchResult, - is_valid_env_name, -) - - -# --------------------------------------------------------------------------- -# Cache entry -# --------------------------------------------------------------------------- - - @dataclass class CachedFetch: """A set of fetched secret values plus when they were fetched.""" @@ -67,20 +48,8 @@ class CachedFetch: return (time.time() - self.fetched_at) < ttl_seconds - - -# --------------------------------------------------------------------------- -# Disk cache -# --------------------------------------------------------------------------- - - def resolve_cache_home(home_path: Optional[Path] = None) -> Path: - """Resolve the Hermes home used for cache paths. - - ``home_path`` is whatever ``load_hermes_dotenv()`` already resolved; - falling back to ``$HERMES_HOME`` / ``~/.hermes`` keeps direct callers - (and tests that don't thread a home through) working. - """ + """``home_path`` as resolved by ``load_hermes_dotenv()``, else ``$HERMES_HOME``/``~/.hermes``.""" if home_path is None: from hermes_constants import get_hermes_home @@ -88,36 +57,70 @@ def resolve_cache_home(home_path: Optional[Path] = None) -> Path: return home_path +def entry_from_payload(payload: object) -> Optional[CachedFetch]: + """``{"secrets": {...}, "fetched_at": n}`` → :class:`CachedFetch`, or None if malformed. + + Only str→str pairs survive (JSON permits other types; env vars need strings). + """ + if not isinstance(payload, dict): + return None + secrets, fetched_at = payload.get("secrets"), payload.get("fetched_at") + if not isinstance(secrets, dict) or not isinstance(fetched_at, (int, float)): + 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(fetched_at)) + + +def atomic_write_json(path: Path, payload: dict, *, tmp_prefix: str) -> None: + """Write ``payload`` to ``path`` via mkstemp → chmod 0600 → os.replace. + + The containing dir is forced to ``0700`` (``mkdir``'s mode is umask-subject, + so the chmod is the reliable form). Raises ``OSError`` on failure; callers + decide whether that is best-effort. + """ + cache_dir = path.parent + cache_dir.mkdir(parents=True, exist_ok=True) + try: + os.chmod(cache_dir, 0o700) + except OSError: + pass + # tempfile honours os.umask, so chmod 0600 explicitly before the rename. + fd, tmp = tempfile.mkstemp(prefix=tmp_prefix, suffix=".tmp", dir=str(cache_dir)) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(payload, f) + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + K = TypeVar("K") class DiskCache(Generic[K]): """Best-effort, profile-aware on-disk cache for fetched secret values. - One JSON object per backend lives at ``/cache/``:: + One JSON object per backend at ``/cache/``:: {"key": "", "secrets": {...}, "fetched_at": 1.0} - The file holds only secret *values* keyed by the serialized cache key — - never raw auth material. Backends are responsible for fingerprinting - tokens/sessions *before* they reach ``key_serializer`` so the token can't - land in the key. - - Writes are atomic (``mkstemp`` → ``chmod 0600`` → ``os.replace``) and the - containing ``cache/`` directory is forced to ``0700`` — ``mkdir``'s mode is - umask-subject, so the chmod is the reliable form. Both ``read`` and - ``write`` short-circuit when ``ttl_seconds <= 0``, so setting the TTL to - zero disables *both* cache layers symmetrically: a user opting out never - gets secret values written to disk at all. + The file holds only secret *values*, never raw auth material — backends + fingerprint tokens/sessions before they reach ``key_serializer``. Both + ``read`` and ``write`` short-circuit when ``ttl_seconds <= 0``, so a TTL of + zero disables both layers symmetrically: an opted-out user never gets + secret values written to disk at all. """ def __init__(self, basename: str, *, key_serializer: Callable[[K], str]) -> None: self._basename = basename self._key_serializer = key_serializer - # Temp-file prefix derived from the basename so concurrent writers for - # different backends in the same dir don't collide on the staging name. - stem = basename.split(".", 1)[0] - self._tmp_prefix = f".{stem}_" + # Per-backend temp prefix so concurrent writers in one dir never collide. + self._tmp_prefix = f".{basename.split('.', 1)[0]}_" def path(self, home_path: Optional[Path] = None) -> Path: return resolve_cache_home(home_path) / "cache" / self._basename @@ -128,36 +131,18 @@ class DiskCache(Generic[K]): ttl_seconds: float, home_path: Optional[Path] = None, ) -> Optional[CachedFetch]: - """Return a fresh cached entry for ``key``, or None. - - Best-effort: any I/O or parse error, a key mismatch, or a stale entry - all return None so the caller re-fetches. - """ + """Fresh cached entry for ``key``, or None (I/O error, mismatch, stale).""" if ttl_seconds <= 0: return None - path = self.path(home_path) try: - with open(path, "r", encoding="utf-8") as f: + with open(self.path(home_path), "r", encoding="utf-8") as f: payload = json.load(f) except (OSError, json.JSONDecodeError): return None - if not isinstance(payload, dict): + if not isinstance(payload, dict) or payload.get("key") != self._key_serializer(key): return None - if payload.get("key") != self._key_serializer(key): - return None - secrets = payload.get("secrets") - fetched_at = payload.get("fetched_at") - if not isinstance(secrets, dict) or not isinstance(fetched_at, (int, float)): - return None - # JSON permits non-string values; env vars need strings, so coerce by - # dropping anything that isn't a str→str pair. - typed: Dict[str, str] = { - k: v for k, v in secrets.items() if isinstance(k, str) and isinstance(v, str) - } - entry = CachedFetch(secrets=typed, fetched_at=float(fetched_at)) - if not entry.is_fresh(ttl_seconds): - return None - return entry + entry = entry_from_payload(payload) + return entry if entry is not None and entry.is_fresh(ttl_seconds) else None def write( self, @@ -166,44 +151,16 @@ class DiskCache(Generic[K]): ttl_seconds: float, home_path: Optional[Path] = None, ) -> None: - """Persist ``entry`` for ``key`` atomically at mode ``0600``. - - No-op when ``ttl_seconds <= 0`` (so caching is genuinely off) or on any - I/O error — the next invocation just re-fetches. - """ + """Persist ``entry`` atomically at mode 0600; no-op when ``ttl_seconds <= 0`` or on I/O error.""" if ttl_seconds <= 0: return - path = self.path(home_path) + payload = { + "key": self._key_serializer(key), + "secrets": entry.secrets, + "fetched_at": entry.fetched_at, + } try: - cache_dir = path.parent - cache_dir.mkdir(parents=True, exist_ok=True) - # mkdir's mode is umask-subject; chmod the dir to 0700 so cache - # metadata isn't exposed if HERMES_HOME is ever made traversable. - try: - os.chmod(cache_dir, 0o700) - except OSError: - pass - payload = { - "key": self._key_serializer(key), - "secrets": entry.secrets, - "fetched_at": entry.fetched_at, - } - # Write to a sibling temp file and atomic-rename. tempfile honours - # os.umask, so we explicitly chmod 0600 before the rename. - fd, tmp = tempfile.mkstemp( - prefix=self._tmp_prefix, suffix=".tmp", dir=str(cache_dir) - ) - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(payload, f) - os.chmod(tmp, 0o600) - os.replace(tmp, path) - except BaseException: - try: - os.unlink(tmp) - except OSError: - pass - raise + atomic_write_json(self.path(home_path), payload, tmp_prefix=self._tmp_prefix) except OSError: pass # best-effort — a disk-cache miss next invocation is fine @@ -213,3 +170,48 @@ class DiskCache(Generic[K]): self.path(home_path).unlink() except (FileNotFoundError, OSError): pass + + +class SecretCache(Generic[K]): + """Two-layer cache: in-process dict (L1) over a :class:`DiskCache` (L2). + + L1 saves repeated fetches WITHIN one process (CLI startup, gateway + hot-reload); L2 saves them ACROSS back-to-back short-lived processes. + """ + + def __init__(self, basename: str, *, key_serializer: Callable[[K], str]) -> None: + self.memory: Dict[K, CachedFetch] = {} + self.disk: DiskCache[K] = DiskCache(basename, key_serializer=key_serializer) + + def lookup( + self, + key: K, + ttl_seconds: float, + home_path: Optional[Path] = None, + read_disk: Optional[Callable[[], Optional[CachedFetch]]] = None, + ) -> Optional[CachedFetch]: + """Fresh entry from L1, else from L2 (promoted into L1), else None. + + ``read_disk`` swaps in an alternative L2 reader (e.g. an encrypted file). + """ + cached = self.memory.get(key) + if cached and cached.is_fresh(ttl_seconds): + return cached + disk_cached = read_disk() if read_disk else self.disk.read(key, ttl_seconds, home_path) + if disk_cached is not None: + self.memory[key] = disk_cached + return disk_cached + + def store( + self, + key: K, + entry: CachedFetch, + ttl_seconds: float, + home_path: Optional[Path] = None, + ) -> None: + self.memory[key] = entry + self.disk.write(key, entry, ttl_seconds, home_path) + + def clear(self, home_path: Optional[Path] = None) -> None: + self.memory.clear() + self.disk.clear(home_path) diff --git a/agent/secret_sources/base.py b/agent/secret_sources/base.py index a7bc345baa..56f6ba348c 100644 --- a/agent/secret_sources/base.py +++ b/agent/secret_sources/base.py @@ -1,37 +1,27 @@ """Secret-source contract: the ABC every secret backend implements. A *secret source* resolves credentials from an external secret manager -(Bitwarden Secrets Manager, 1Password, an OS keystore, a user script, ...) -into environment-variable-shaped values at process startup, AFTER -``~/.hermes/.env`` has loaded and BEFORE the rest of Hermes reads -``os.environ``. +(Bitwarden, 1Password, a user script, ...) into env-var-shaped values at +process startup, AFTER ``~/.hermes/.env`` has loaded and BEFORE the rest of +Hermes reads ``os.environ``. Scope of the contract (deliberate, please do not widen): -* **Read-only.** Sources resolve refs → values. There is no write-back - ("save this key to your vault"), no arbitrary secret objects, and no - mid-session secret API. If a future need for rotation/refresh appears - it will arrive as a versioned optional hook — do not bolt it on. -* **Startup-time, synchronous.** ``fetch()`` is called once per process - (per HERMES_HOME) by the orchestrator in - :mod:`agent.secret_sources.registry`, which enforces a wall-clock - timeout around it. Sources must not spawn background refreshers. -* **Never raises, never prompts.** ``fetch()`` returns a - :class:`FetchResult` — errors go in ``result.error`` with a - machine-readable :class:`ErrorKind`. Interactive auth belongs in the - source's CLI ``setup`` flow, never on the startup path (non-TTY - gateway/cron startup must never block on stdin). -* **Sources fetch; the orchestrator applies.** A source returns the - name→value mapping it *would* contribute. Precedence (mapped-beats-bulk, - first-wins, ``override_existing``, protected vars), conflict warnings, - provenance tracking, and the actual ``os.environ`` writes are owned by - the orchestrator so no backend can get them wrong. +* **Read-only.** Sources resolve refs → values; no write-back, no arbitrary + secret objects, no mid-session secret API. +* **Startup-time, synchronous.** ``fetch()`` runs once per process (per + HERMES_HOME) under a wall-clock timeout enforced by the registry. Sources + must not spawn background refreshers. +* **Never raises, never prompts.** Errors go in ``FetchResult.error`` with a + machine-readable :class:`ErrorKind`; interactive auth belongs in the CLI + ``setup`` flow (non-TTY gateway/cron startup must never block on stdin). +* **Sources fetch; the orchestrator applies.** Precedence, conflict warnings, + provenance and the ``os.environ`` writes live in ``registry.apply_all`` so + no backend can get them wrong. -Versioning: ``SECRET_SOURCE_API_VERSION`` gates plugin compatibility. -New *optional* hooks with default implementations do not bump it; -required-signature changes do, and the registry skips (with a warning) -sources built against a different major version instead of crashing -startup. +``SECRET_SOURCE_API_VERSION`` gates plugin compatibility: additive optional +hooks with defaults do NOT bump it; required-signature changes do, and the +registry skips (with a warning) sources built against another version. """ from __future__ import annotations @@ -39,18 +29,19 @@ from __future__ import annotations import os import re import subprocess -from contextvars import ContextVar, Token from abc import ABC, abstractmethod +from contextvars import ContextVar, Token from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Dict, FrozenSet, List, MutableMapping, Optional, Sequence +from typing import Any, Dict, FrozenSet, List, MutableMapping, Optional, Sequence, Tuple -# Bump ONLY for breaking changes to the required contract surface -# (abstract-method signatures, FetchResult required fields). Additive -# optional hooks must ship with defaults and must NOT bump this. SECRET_SOURCE_API_VERSION = 1 +# Generous: a first run may include a one-time CLI auto-install (bws download). +DEFAULT_FETCH_TIMEOUT_SECONDS = 120.0 +DEFAULT_CLI_TIMEOUT_SECONDS = 30.0 + _SOURCE_ENVIRONMENT: ContextVar[Optional[MutableMapping[str, str]]] _SOURCE_ENVIRONMENT = ContextVar("hermes_secret_source_environment", default=None) @@ -69,22 +60,28 @@ def get_source_environment() -> MutableMapping[str, str]: environ = _SOURCE_ENVIRONMENT.get() return environ if environ is not None else os.environ -# Timeout the orchestrator enforces around fetch() when the source's -# config section doesn't override it. Generous because a first run may -# include a one-time CLI binary auto-install (e.g. bws download+verify). -DEFAULT_FETCH_TIMEOUT_SECONDS = 120.0 -# Default timeout for run_secret_cli() subprocess invocations. -DEFAULT_CLI_TIMEOUT_SECONDS = 30.0 +def source_child_env() -> Dict[str, str]: + """Environment for a helper child that legitimately needs the caller's env. + + Single-profile startup keeps the legacy contract (full process env, minus + the terminal blocklist). A profile-local fetch (multiplex) gets ONLY the + per-fetch view so a child can never inherit sibling profiles' secrets. + """ + source_env = get_source_environment() + if source_env is os.environ: + from tools.environments.local import build_subprocess_env + + return build_subprocess_env(scrub_secrets=False, inherit_profile_home=False) + return dict(source_env) class ErrorKind(str, Enum): """Machine-readable failure taxonomy for :class:`FetchResult.error`. A fixed vocabulary keeps startup warnings and ``hermes secrets status`` - uniform across backends, and lets the orchestrator implement - kind-dependent policy (e.g. a future stale-cache fallback on - ``NETWORK``/``TIMEOUT`` but not on ``AUTH_FAILED``) exactly once. + uniform, and lets the orchestrator apply kind-dependent policy (e.g. + stale-cache fallback on NETWORK/TIMEOUT but never AUTH_FAILED) once. """ NOT_CONFIGURED = "not_configured" # enabled but missing token/project/map @@ -98,15 +95,35 @@ class ErrorKind(str, Enum): INTERNAL = "internal" # anything else (bug, unexpected shape) +# Ordered (kind, substrings) rules for mapping CLI failure text onto ErrorKind; +# first rule whose substring appears (case-insensitive) wins. +ErrorRules = Sequence[Tuple[ErrorKind, Sequence[str]]] + + +def classify_cli_error(message: str, rules: ErrorRules) -> ErrorKind: + """Best-effort mapping of helper-CLI failure text onto the taxonomy.""" + lowered = message.lower() + for kind, tokens in rules: + if any(tok in lowered for tok in tokens): + return kind + return ErrorKind.INTERNAL + + +def coerce_float(value: Any, default: float) -> float: + """``float(value)`` with ``default`` for malformed config values.""" + try: + return float(value) + except (TypeError, ValueError): + return default + + @dataclass class FetchResult: """Outcome of one source's fetch. - ``secrets`` holds what the source *would* contribute; whether each - var is actually applied is the orchestrator's decision. ``applied`` - and ``skipped`` exist for backward compatibility with the original - Bitwarden fetch-and-apply entry point and are left empty by - conforming ``fetch()`` implementations. + ``secrets`` is what the source *would* contribute; whether each var is + applied is the orchestrator's decision. ``applied``/``skipped`` exist for + the legacy fetch-and-apply entry points and stay empty in ``fetch()``. """ secrets: Dict[str, str] = field(default_factory=dict) @@ -115,40 +132,35 @@ class FetchResult: warnings: List[str] = field(default_factory=list) error: Optional[str] = None error_kind: Optional[ErrorKind] = None - # Path of the helper binary used, when the source is CLI-driven. - # Surfaced by status commands; None for SDK/API-driven sources. + # Helper binary used (CLI-driven sources); surfaced by status commands. binary_path: Optional[Path] = None @property def ok(self) -> bool: return self.error is None + def fail(self, error: str, kind: ErrorKind) -> "FetchResult": + self.error, self.error_kind = error, kind + return self + class SecretSource(ABC): - """One external secret backend. - - Subclasses set the class attributes and implement :meth:`fetch`. - Everything else has a sensible default. + """One external secret backend. Subclasses set attributes + ``fetch``. Attributes: - name: Config-section key under ``secrets:`` in config.yaml. - Lowercase ``[a-z0-9_]+``. Also the provenance label stored - for every var this source supplies. - label: Human-readable name used in startup messages and - ``hermes secrets status`` (e.g. ``"Bitwarden Secrets Manager"``). - shape: ``"mapped"`` when the user explicitly binds env-var names - to refs (1Password ``env:`` map, command source) or - ``"bulk"`` when the backend injects whole projects/folders - of secrets implicitly (Bitwarden BSM). The orchestrator - gives mapped sources precedence over bulk sources: an - explicit binding is stronger intent than a project dump. - scheme: Optional URI scheme this source owns for secret - references (``"op"`` for ``op://...``). Must be unique - across registered sources — refs may eventually appear - outside the ``secrets:`` block (e.g. credential-pool - ``api_key`` fields), so scheme collisions are rejected at - registration time to keep that future possible. - api_version: Contract version this source was built against. + name: Config-section key under ``secrets:`` (``[a-z0-9_]+``); also the + provenance label for every var this source supplies. + label: Human-readable name for startup messages / ``secrets status``. + shape: ``"mapped"`` (user binds env-var names to refs) or ``"bulk"`` + (backend injects whole projects). Mapped beats bulk: an explicit + binding is stronger intent than a project dump. + scheme: URI scheme this source owns for refs (``"op"``). Unique across + sources so refs can later appear outside the ``secrets:`` block. + token_env_key / default_token_env: config key naming the bootstrap-auth + env var, and its default. Drives :meth:`protected_env_vars` so a + vault holding its own access token can't clobber the credential + used to reach it. + override_existing_default: value of ``override_existing`` when unset. """ api_version: int = SECRET_SOURCE_API_VERSION @@ -156,112 +168,77 @@ class SecretSource(ABC): label: str = "" shape: str = "mapped" # "mapped" | "bulk" scheme: Optional[str] = None - - # -- required ---------------------------------------------------------- + token_env_key: Optional[str] = None + default_token_env: str = "" + override_existing_default: bool = False @abstractmethod def fetch(self, cfg: dict, home_path: Path) -> FetchResult: - """Resolve this source's secrets. MUST NOT raise or prompt. + """Resolve this source's secrets. MUST NOT raise or prompt. - ``cfg`` is the source's raw config section (``secrets.``) - from config.yaml — treat every field defensively, the section - may be malformed. ``home_path`` is the resolved HERMES_HOME. + ``cfg`` is the raw ``secrets.`` section — may be malformed. """ - # -- optional hooks (defaults are correct for most sources) ------------ - def is_enabled(self, cfg: dict) -> bool: - """Whether the user turned this source on.""" return bool(isinstance(cfg, dict) and cfg.get("enabled")) def override_existing(self, cfg: dict) -> bool: - """May this source overwrite vars that .env / the shell already set? + """May this source overwrite vars .env / the shell already set? - This NEVER extends to vars claimed by another secret source in the - same startup pass — cross-source overrides are a config error the - orchestrator warns about, not a knob. + Never extends to vars claimed by another source in the same pass — + cross-source overrides are a config error the orchestrator warns about. """ - return bool(isinstance(cfg, dict) and cfg.get("override_existing", False)) + return bool(isinstance(cfg, dict) + and cfg.get("override_existing", self.override_existing_default)) + + def token_env(self, cfg: dict) -> str: + """Name of the env var holding this source's bootstrap credential.""" + if isinstance(cfg, dict) and self.token_env_key: + return str(cfg.get(self.token_env_key) or self.default_token_env) + return self.default_token_env def protected_env_vars(self, cfg: dict) -> FrozenSet[str]: - """Env vars the orchestrator must never let ANY source overwrite. - - Typically the source's own bootstrap-auth var (e.g. - ``BWS_ACCESS_TOKEN``) so a vault that contains its own access - token can't clobber the credential used to reach it. - """ - return frozenset() + """Env vars the orchestrator must never let ANY source overwrite.""" + return frozenset({self.token_env(cfg)}) if self.token_env_key else frozenset() def fetch_timeout_seconds(self, cfg: dict) -> float: """Wall-clock budget the orchestrator enforces around fetch().""" - try: - val = float((cfg or {}).get("timeout_seconds", DEFAULT_FETCH_TIMEOUT_SECONDS)) - except (TypeError, ValueError): - return DEFAULT_FETCH_TIMEOUT_SECONDS + val = coerce_float((cfg or {}).get("timeout_seconds", DEFAULT_FETCH_TIMEOUT_SECONDS), + DEFAULT_FETCH_TIMEOUT_SECONDS) return val if val > 0 else DEFAULT_FETCH_TIMEOUT_SECONDS def config_schema(self) -> dict: - """Optional description of this source's config keys. - - Shape: ``{key: {"description": str, "default": Any}}``. Used by - setup surfaces to render config without hardcoding per-source - knowledge. Purely informational. - """ + """Informational ``{key: {"description": str, "default": Any}}`` for setup UIs.""" return {} def remediation(self, kind: Optional["ErrorKind"], cfg: dict) -> str: - """One-line, actionable next step for a failed fetch. + """One-line actionable next step for a failed fetch (pure, no I/O). - Called by the startup status printer (and ``hermes secrets ... - status``) right after a fetch error is surfaced, so the user sees - *what to run* next to fix it — not just what broke. Sources - should override this to point at their own CLI verbs (e.g. - ``hermes secrets bitwarden token`` for AUTH_FAILED). Return an - empty string to suppress the hint. - - Must never raise and must not perform I/O — it's a pure - kind→string mapping on the startup path. + Shown right after the fetch error by the startup status printer and + ``hermes secrets ... status``. Empty string suppresses the hint. """ - generic = { - ErrorKind.NOT_CONFIGURED: ( - f"Run `hermes secrets {self.name} setup` to finish configuration." - ), - ErrorKind.BINARY_MISSING: ( - f"Run `hermes secrets {self.name} setup` to install the helper CLI." - ), - ErrorKind.AUTH_FAILED: ( - f"Credentials rejected — run `hermes secrets {self.name} setup` " - "to re-authenticate." - ), - ErrorKind.AUTH_EXPIRED: ( - f"Credentials expired — run `hermes secrets {self.name} setup` " - "to re-authenticate." - ), - ErrorKind.NETWORK: ( - "Network problem reaching the secrets backend — check " - "connectivity and retry." - ), - ErrorKind.TIMEOUT: ( - f"Backend was slow — raise secrets.{self.name}.timeout_seconds " - "if this recurs." - ), - } - return generic.get(kind, "") if kind is not None else "" + return _GENERIC_REMEDIATION.get(kind, "").format(name=self.name) if kind is not None else "" + + +_GENERIC_REMEDIATION = { + ErrorKind.NOT_CONFIGURED: "Run `hermes secrets {name} setup` to finish configuration.", + ErrorKind.BINARY_MISSING: "Run `hermes secrets {name} setup` to install the helper CLI.", + ErrorKind.AUTH_FAILED: "Credentials rejected — run `hermes secrets {name} setup` to re-authenticate.", + ErrorKind.AUTH_EXPIRED: "Credentials expired — run `hermes secrets {name} setup` to re-authenticate.", + ErrorKind.NETWORK: "Network problem reaching the secrets backend — check connectivity and retry.", + ErrorKind.TIMEOUT: "Backend was slow — raise secrets.{name}.timeout_seconds if this recurs.", +} # --------------------------------------------------------------------------- # Shared helpers — use these instead of hand-rolling per backend # --------------------------------------------------------------------------- - _ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") -# ANSI CSI/OSC escape sequences — helper-CLI stderr often carries color -# codes that must not reach Hermes' own startup output. -# NOTE: intentionally NOT migrated to tools.ansi_strip.strip_ansi — the -# optional terminator here (``(?:\x07|\x1b\\)?``) also strips *unterminated* -# OSC sequences (common when a CLI is killed mid-write), which strip_ansi -# leaves untouched. strip_ansi is not a superset of this regex. +# Deliberately NOT tools.ansi_strip.strip_ansi: the optional terminator here +# also strips *unterminated* OSC sequences (a CLI killed mid-write), which +# strip_ansi leaves untouched. _ANSI_RE = re.compile(r"\x1b(?:\[[0-9;?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)?)") @@ -275,6 +252,35 @@ def scrub_ansi(text: str) -> str: return _ANSI_RE.sub("", text or "") +def run_cli( + argv: Sequence[str], + *, + env: Dict[str, str], + timeout: float, + label: str, + timeout_message: str, + stdin: Any = subprocess.DEVNULL, +) -> subprocess.CompletedProcess: + """``subprocess.run`` an argv list (never a shell), capturing utf-8 text. + + Timeout and spawn failure become ``RuntimeError`` with messages safe to + surface; callers own returncode interpretation. + """ + try: + return subprocess.run( # noqa: S603 — argv list, no shell + list(argv), + env=env, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + timeout=timeout, + stdin=stdin, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError(timeout_message) from exc + except OSError as exc: + raise RuntimeError(f"failed to invoke {label}: {exc}") from exc + + def run_secret_cli( argv: Sequence[str], *, @@ -284,53 +290,22 @@ def run_secret_cli( ) -> subprocess.CompletedProcess: """Run a secret-manager helper CLI with a minimal, allowlisted env. - Security posture shared by every subprocess-driven backend: - - * argv list only — never ``shell=True``. Callers pass user-supplied - reference strings AFTER a ``--`` option terminator in their argv. - * The child gets ``PATH``/``HOME``/locale basics plus only the env - vars named in ``allow_env`` (auth/session vars) and ``extra_env`` - — never a copy of the full post-dotenv ``os.environ``, which by - this point holds every credential Hermes knows about. - * ``NO_COLOR=1`` is set and stderr/stdout are ANSI-scrubbed so - helper diagnostics can't smuggle escape sequences into Hermes - output. - * stdin is ``/dev/null`` so a helper that decides to prompt fails - fast instead of hanging startup. - - Raises ``RuntimeError`` on spawn failure or timeout (message safe to - surface); returns the completed process otherwise — callers own - returncode interpretation. + The child gets PATH/HOME/locale basics plus only ``allow_env`` (auth/session + vars) and ``extra_env`` — never the full post-dotenv ``os.environ``, which + holds every credential Hermes knows. ``NO_COLOR=1`` plus ANSI-scrubbed + stderr keep helper diagnostics out of Hermes output; stdin is /dev/null so + a prompting helper fails fast. Pass user refs AFTER a ``--`` terminator. """ base_keep = ("PATH", "HOME", "USERPROFILE", "SYSTEMROOT", "TMPDIR", "TEMP", "LANG", "LC_ALL", "XDG_CONFIG_HOME", "XDG_DATA_HOME") - env: Dict[str, str] = {} - for key in (*base_keep, *allow_env): - val = os.environ.get(key) - if val is not None: - env[key] = val + env = {k: os.environ[k] for k in (*base_keep, *allow_env) if k in os.environ} if extra_env: env.update(extra_env) env.setdefault("NO_COLOR", "1") - try: - proc = subprocess.run( # noqa: S603 — argv list, no shell - list(argv), - env=env, - capture_output=True, - text=True, encoding="utf-8", errors="replace", - timeout=timeout, - stdin=subprocess.DEVNULL, - ) - except subprocess.TimeoutExpired as exc: - raise RuntimeError( - f"{Path(str(argv[0])).name} timed out after {timeout:.0f}s" - ) from exc - except OSError as exc: - raise RuntimeError( - f"failed to invoke {Path(str(argv[0])).name}: {exc}" - ) from exc - + name = Path(str(argv[0])).name + proc = run_cli(argv, env=env, timeout=timeout, label=name, + timeout_message=f"{name} timed out after {timeout:.0f}s") proc.stdout = proc.stdout or "" proc.stderr = scrub_ansi(proc.stderr or "") return proc