ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(agent/creds): unify secret-source CLI/cache/error plumbing in base and _cache

Browse Source 
- 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.
This commit is contained in:
Teknium
2026-09-02 10:07:21 -07:00
parent a3d33fe22f
commit d4f2933262
3 changed files with 290 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/secret_sources/__init__.py
+10 -27
View File
@@ -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
agent/secret_sources/_cache.py
+122 -120
View File
@@ -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 ``<hermes_home>/cache/<basename>``::
One JSON object per backend at ``<hermes_home>/cache/<basename>``::
{"key": "<serialized 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)
agent/secret_sources/base.py
+158 -183
View File
@@ -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.<name>``)
from config.yaml — treat every field defensively, the section
may be malformed. ``home_path`` is the resolved HERMES_HOME.
``cfg`` is the raw ``secrets.<name>`` 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