256 lines
11 KiB
Python
256 lines
11 KiB
Python
"""Secret-source contract: the ABC every secret backend implements.
|
|
|
|
A *secret source* resolves credentials from an external secret manager into
|
|
env-var-shaped values at process startup, AFTER ``~/.hermes/.env`` has loaded
|
|
and BEFORE the rest of Hermes reads ``os.environ``. The contract is deliberately
|
|
narrow: read-only; startup-time and synchronous (one ``fetch()`` per process per
|
|
HERMES_HOME, under a registry-enforced wall-clock timeout, no background
|
|
refreshers); never raises, never prompts (errors go in ``FetchResult.error``
|
|
with an :class:`ErrorKind`; interactive auth belongs in the CLI ``setup`` flow);
|
|
sources fetch, the orchestrator (``registry.apply_all``) applies.
|
|
|
|
``SECRET_SOURCE_API_VERSION`` gates plugin compatibility: additive optional
|
|
hooks with defaults do NOT bump it; required-signature changes do.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
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 Any, Dict, FrozenSet, List, MutableMapping, Optional, Sequence, Tuple
|
|
|
|
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)
|
|
|
|
|
|
def set_source_environment(environ: MutableMapping[str, str]) -> Token:
|
|
"""Install a per-fetch environment view without changing ``os.environ``."""
|
|
return _SOURCE_ENVIRONMENT.set(environ)
|
|
|
|
|
|
def reset_source_environment(token: Token) -> None:
|
|
_SOURCE_ENVIRONMENT.reset(token)
|
|
|
|
|
|
def get_source_environment() -> MutableMapping[str, str]:
|
|
"""Return the active per-fetch environment, or the process environment."""
|
|
environ = _SOURCE_ENVIRONMENT.get()
|
|
return environ if environ is not None else os.environ
|
|
|
|
|
|
def source_child_env() -> Dict[str, str]:
|
|
"""Environment for a helper child that legitimately needs the caller's env:
|
|
full process env (minus the terminal blocklist) in single-profile startup;
|
|
ONLY the per-fetch view under multiplex, so no sibling profile's secrets leak."""
|
|
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):
|
|
"""Failure taxonomy for :class:`FetchResult.error`; lets the orchestrator apply
|
|
kind-dependent policy once (stale-cache fallback on NETWORK/TIMEOUT, never AUTH_FAILED)."""
|
|
|
|
NOT_CONFIGURED = "not_configured" # enabled but missing token/project/map
|
|
BINARY_MISSING = "binary_missing" # helper CLI not found / not installed
|
|
AUTH_FAILED = "auth_failed" # bad credentials
|
|
AUTH_EXPIRED = "auth_expired" # credentials were valid, aren't now
|
|
REF_INVALID = "ref_invalid" # a secret reference failed validation
|
|
NETWORK = "network" # transport-level failure
|
|
EMPTY_VALUE = "empty_value" # backend returned nothing for a ref
|
|
TIMEOUT = "timeout" # fetch exceeded its wall-clock budget
|
|
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`` is what the source *would*
|
|
contribute; ``applied``/``skipped`` serve the legacy fetch-and-apply entry
|
|
points and stay empty in ``fetch()``."""
|
|
|
|
secrets: Dict[str, str] = field(default_factory=dict)
|
|
applied: List[str] = field(default_factory=list)
|
|
skipped: List[str] = field(default_factory=list)
|
|
warnings: List[str] = field(default_factory=list)
|
|
error: Optional[str] = None
|
|
error_kind: Optional[ErrorKind] = None
|
|
# 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
|
|
|
|
|
|
_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.",
|
|
}
|
|
|
|
|
|
class SecretSource(ABC):
|
|
"""One external secret backend. Subclasses set attributes + ``fetch``.
|
|
|
|
``name``: config-section key under ``secrets:`` (``[a-z0-9_]+``) and the
|
|
provenance label. ``shape``: ``"mapped"`` (user binds env-var names to refs)
|
|
or ``"bulk"`` (backend injects whole projects); mapped beats bulk because an
|
|
explicit binding is stronger intent. ``scheme``: URI scheme this source owns
|
|
for refs, unique across sources. ``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. ``remediation_hints``: per-kind
|
|
overrides of the generic remediation text (``{name}`` / ``{token_env}``).
|
|
"""
|
|
|
|
api_version: int = SECRET_SOURCE_API_VERSION
|
|
name: str = ""
|
|
label: str = ""
|
|
shape: str = "mapped" # "mapped" | "bulk"
|
|
scheme: Optional[str] = None
|
|
token_env_key: Optional[str] = None
|
|
default_token_env: str = ""
|
|
override_existing_default: bool = False
|
|
remediation_hints: Dict[ErrorKind, str] = {}
|
|
|
|
@abstractmethod
|
|
def fetch(self, cfg: dict, home_path: Path) -> FetchResult:
|
|
"""Resolve this source's secrets. MUST NOT raise or prompt; ``cfg`` is the
|
|
raw ``secrets.<name>`` section and may be malformed."""
|
|
|
|
def is_enabled(self, cfg: dict) -> bool:
|
|
return bool(isinstance(cfg, dict) and cfg.get("enabled"))
|
|
|
|
def override_existing(self, cfg: dict) -> bool:
|
|
"""May this source overwrite vars .env / the shell already set? Never extends
|
|
to vars claimed by another source (a config error the orchestrator warns about)."""
|
|
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."""
|
|
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()."""
|
|
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:
|
|
"""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 (pure); "" suppresses the hint."""
|
|
if kind is None:
|
|
return ""
|
|
template = self.remediation_hints.get(kind) or _GENERIC_REMEDIATION.get(kind, "")
|
|
return template.format(name=self.name, token_env=self.token_env(cfg))
|
|
|
|
|
|
# --- Shared helpers — use these instead of hand-rolling per backend ---------
|
|
|
|
_ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
|
|
# 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\\)?)")
|
|
|
|
|
|
def is_valid_env_name(name: str) -> bool:
|
|
"""True when ``name`` is a legal environment-variable name."""
|
|
return bool(name) and bool(_ENV_NAME_RE.match(name))
|
|
|
|
|
|
def scrub_ansi(text: str) -> str:
|
|
"""Strip ANSI escape sequences (whole CSI/OSC sequences, not just ESC)."""
|
|
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``. 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], *, allow_env: Sequence[str] = (), extra_env: Optional[Dict[str, str]] = None,
|
|
timeout: float = DEFAULT_CLI_TIMEOUT_SECONDS) -> subprocess.CompletedProcess:
|
|
"""Run a secret-manager helper CLI with a minimal, allowlisted env (never the
|
|
full post-dotenv ``os.environ``): PATH/HOME/locale basics plus ``allow_env``
|
|
and ``extra_env``. ``NO_COLOR=1`` + ANSI-scrubbed stderr; 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 = {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")
|
|
|
|
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
|