637 lines
28 KiB
Python
637 lines
28 KiB
Python
"""Helpers for loading Hermes .env files consistently across entrypoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import codecs
|
|
import io
|
|
import logging
|
|
import os
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from utils import atomic_replace, fast_safe_load
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Env var name suffixes that indicate credential values. These are the ONLY env vars whose
|
|
# values we sanitize on load — we must not silently alter arbitrary user env vars, but
|
|
# credentials are known to require pure ASCII (they become HTTP header values).
|
|
_CREDENTIAL_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY")
|
|
|
|
# Names already warned about this process, so repeated load_hermes_dotenv() calls (user env +
|
|
# project env, gateway hot-reload, tests) don't spam the same warning.
|
|
_WARNED_KEYS: set[str] = set()
|
|
|
|
# Paths already given the UTF-32 refuse-to-mangle warning (once per path is enough).
|
|
_WARNED_UTF32_PATHS: set[str] = set()
|
|
|
|
# env-var name → source label ("bitwarden", …) for credentials injected by an external secret
|
|
# source during load_hermes_dotenv(). Setup / `hermes model` use it to label detected credentials
|
|
# so users know WHERE a key came from when their .env doesn't contain it directly.
|
|
_SECRET_SOURCES: dict[str, str] = {}
|
|
# Applied values are immutable per-home snapshots. ``os.environ`` is shared across profiles and
|
|
# may be overwritten by a later home's source apply.
|
|
_SECRET_SOURCE_VALUES_BY_HOME: dict[str, dict[str, str]] = {}
|
|
|
|
# HERMES_HOME paths already pulled external secrets for this process. load_hermes_dotenv() runs
|
|
# at import time from several hot modules (cli.py, hermes_cli/main.py, run_agent.py, gateway/
|
|
# run.py, …); without this guard the Bitwarden status line prints 3-5x per startup and the config
|
|
# re-parse + ASCII sweep re-run each time (Bitwarden's own cache only saves the network call).
|
|
_APPLIED_HOMES: set[str] = set()
|
|
_SECRET_SOURCE_CACHE_LOCK = threading.RLock()
|
|
|
|
# Routed profile homes whose dotenv load was skipped under multiplex, so the skip is logged once
|
|
# per home rather than on every lazy import mid-turn.
|
|
_SCOPED_SKIP_LOGGED: set[str] = set()
|
|
|
|
|
|
# Behavioral routing keys a parent Hermes process injects into child env and that silently
|
|
# redirect a profile onto the wrong provider path (ACP auth method, copilot-ACP endpoints).
|
|
# These — and ONLY these — are scrubbed from os.environ at startup when absent from the
|
|
# profile's .env. Credential keys are excluded: shell exports are a legitimate, documented way
|
|
# to supply them, and read-time secret-scope checks (agent/secret_scope.py) own cross-profile
|
|
# credential isolation.
|
|
_PROFILE_MANAGED_ENV_KEYS: frozenset[str] = frozenset({
|
|
"HERMES_ACP_AUTH_METHOD",
|
|
"HERMES_ACP_AUTO_APPROVE",
|
|
"HERMES_COPILOT_ACP_COMMAND",
|
|
"HERMES_COPILOT_ACP_ARGS",
|
|
"COPILOT_CLI_PATH",
|
|
"COPILOT_ACP_BASE_URL",
|
|
})
|
|
|
|
|
|
def _env_keys_defined_in_dotenv(path: Path) -> set[str]:
|
|
"""KEY names assigned in a dotenv file (including empty ``KEY=``).
|
|
|
|
A fast line scanner rather than full dotenv parsing, so it works during early bootstrap without
|
|
importing python-dotenv. Decode errors fall back to ``latin-1``, matching ``_load_dotenv_with_fallback``.
|
|
"""
|
|
keys: set[str] = set()
|
|
try:
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
try:
|
|
text = path.read_text(encoding="latin-1", errors="replace")
|
|
except Exception:
|
|
return keys
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key = line.removeprefix("export ").split("=", 1)[0].strip()
|
|
if key:
|
|
keys.add(key)
|
|
return keys
|
|
|
|
|
|
def _clear_known_keys_missing_from_dotenv(path: Path) -> None:
|
|
"""Delete inherited ``_PROFILE_MANAGED_ENV_KEYS`` from ``os.environ`` that ``.env`` does not define.
|
|
|
|
Runs after the profile's ``.env`` loaded with ``override=True``. Scope is deliberately NARROW:
|
|
only the behavioral routing keys that change *which provider path* a profile uses.
|
|
"""
|
|
if not path.exists():
|
|
return
|
|
defined = _env_keys_defined_in_dotenv(path)
|
|
for key in _PROFILE_MANAGED_ENV_KEYS:
|
|
if key not in defined and key in os.environ:
|
|
del os.environ[key]
|
|
|
|
|
|
def get_secret_source(env_var: str) -> str | None:
|
|
"""Label of the secret source that supplied ``env_var`` (``"bitwarden"`` …), or None for .env/shell keys.
|
|
|
|
Metadata only -- credential-pool persistence may store it to explain a borrowed secret's origin
|
|
but must never treat it as authorization to persist the raw value.
|
|
"""
|
|
return _SECRET_SOURCES.get(env_var)
|
|
|
|
|
|
def get_secret_source_values(hermes_home: str | os.PathLike) -> dict[str, str]:
|
|
"""Return the external-secret value snapshot for ``hermes_home``."""
|
|
return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(str(Path(hermes_home).resolve()), {}))
|
|
|
|
|
|
def hydrate_profile_secret_sources(hermes_home: str | os.PathLike) -> dict[str, str]:
|
|
"""Resolve one profile's configured sources without mutating ``os.environ``.
|
|
|
|
Multiplex gateways can route a first turn to a secondary profile that never ran the process-
|
|
global dotenv startup path. Resolve its sources against a private mapping seeded from its own
|
|
``.env`` and record the usual per-home snapshot for ``build_profile_secret_scope()``.
|
|
|
|
Fail-open and once-per-home semantics intentionally mirror ``_apply_external_secret_sources``.
|
|
The result contains only values contributed by external sources, never plaintext ``.env`` entries.
|
|
"""
|
|
with _SECRET_SOURCE_CACHE_LOCK:
|
|
return _hydrate_profile_secret_sources(Path(hermes_home))
|
|
|
|
|
|
def _hydrate_profile_secret_sources(home: Path) -> dict[str, str]:
|
|
"""Locked implementation for :func:`hydrate_profile_secret_sources`."""
|
|
home_key = str(home.resolve())
|
|
if home_key in _APPLIED_HOMES:
|
|
return get_secret_source_values(home)
|
|
|
|
try:
|
|
cfg = _load_secrets_config(home)
|
|
except Exception: # noqa: BLE001 — external sources must not block routing
|
|
return {}
|
|
if not cfg:
|
|
return {}
|
|
|
|
try:
|
|
from agent.secret_scope import _is_global_env, load_env_file
|
|
from agent.secret_sources.registry import apply_all
|
|
|
|
local_env = {name: value for name, value in os.environ.items() if _is_global_env(name)}
|
|
local_env.update(load_env_file(home / ".env"))
|
|
# Mirror load_hermes_dotenv()'s .op.env bootstrap: the 1Password service-account token
|
|
# lives in <home>/.op.env (gitignored), not .env; without it a cold profile on the .op.env
|
|
# flow fails 1Password hydration. .env values win — never override an existing key.
|
|
op_env = home / ".op.env"
|
|
if op_env.exists():
|
|
for _name, _value in load_env_file(op_env).items():
|
|
local_env.setdefault(_name, _value)
|
|
local_env["HERMES_HOME"] = str(home)
|
|
report = apply_all(cfg, home, environ=local_env)
|
|
except Exception: # noqa: BLE001 — preserve fail-open startup behavior
|
|
return {}
|
|
|
|
if not report.sources:
|
|
return {}
|
|
|
|
_APPLIED_HOMES.add(home_key)
|
|
values: dict[str, str] = {}
|
|
for name, applied in report.provenance.items():
|
|
value = local_env.get(name)
|
|
if value is None:
|
|
continue
|
|
_SECRET_SOURCES[name] = applied.source
|
|
values[name] = value
|
|
if values:
|
|
_SECRET_SOURCE_VALUES_BY_HOME[home_key] = values
|
|
return dict(values)
|
|
|
|
|
|
def reset_secret_source_cache() -> None:
|
|
"""Forget which HERMES_HOME paths already had external secrets applied, forcing a re-pull
|
|
(tests, long-running processes after a config change)."""
|
|
_APPLIED_HOMES.clear()
|
|
_SECRET_SOURCES.clear()
|
|
_SECRET_SOURCE_VALUES_BY_HOME.clear()
|
|
|
|
|
|
def format_secret_source_suffix(env_var: str) -> str:
|
|
"""``" (from Bitwarden)"``-style suffix, or ``""`` for .env/shell credentials (the implicit
|
|
default users already understand — only external sources get called out)."""
|
|
source = get_secret_source(env_var)
|
|
if not source:
|
|
return ""
|
|
if source == "bitwarden":
|
|
return " (from Bitwarden)"
|
|
# Registry label (e.g. "1Password"); raw source name for labels the registry doesn't know
|
|
# (stale provenance from an uninstalled plugin, tests).
|
|
try:
|
|
from agent.secret_sources.registry import get_source
|
|
|
|
registered = get_source(source)
|
|
if registered is not None and registered.label:
|
|
return f" (from {registered.label})"
|
|
except Exception: # noqa: BLE001 — label lookup must never raise
|
|
pass
|
|
return f" (from {source})"
|
|
|
|
|
|
def _format_offending_chars(value: str, limit: int = 3) -> str:
|
|
"""Compact ``U+XXXX ('c'), ...`` summary of non-ASCII codepoints."""
|
|
seen: list[str] = []
|
|
for ch in value:
|
|
if ord(ch) > 127:
|
|
label = f"U+{ord(ch):04X}"
|
|
if ch.isprintable():
|
|
label += f" ({ch!r})"
|
|
if label not in seen:
|
|
seen.append(label)
|
|
if len(seen) >= limit:
|
|
break
|
|
return ", ".join(seen)
|
|
|
|
|
|
def _sanitize_loaded_credentials() -> None:
|
|
"""Strip non-ASCII characters from credential env vars (``_CREDENTIAL_SUFFIXES``) in os.environ.
|
|
|
|
Called after dotenv loads so the rest of the codebase never sees non-ASCII API keys.
|
|
"""
|
|
for key, value in list(os.environ.items()):
|
|
if not any(key.endswith(suffix) for suffix in _CREDENTIAL_SUFFIXES):
|
|
continue
|
|
if value.isascii():
|
|
continue
|
|
cleaned = value.encode("ascii", errors="ignore").decode("ascii")
|
|
os.environ[key] = cleaned
|
|
if key in _WARNED_KEYS:
|
|
continue
|
|
_WARNED_KEYS.add(key)
|
|
stripped = len(value) - len(cleaned)
|
|
detail = _format_offending_chars(value) or "non-printable"
|
|
print(
|
|
f" Warning: {key} contained {stripped} non-ASCII character"
|
|
f"{'s' if stripped != 1 else ''} ({detail}) — stripped so the "
|
|
f"key can be sent as an HTTP header.",
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
" This usually means the key was copy-pasted from a PDF, "
|
|
"rich-text editor, or web page that substituted lookalike\n"
|
|
" Unicode glyphs for ASCII letters. If authentication fails "
|
|
"(e.g. \"API key not valid\"), re-copy the key from the\n"
|
|
" provider's dashboard and run `hermes setup` (or edit the "
|
|
".env file in a plain-text editor).",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
|
|
def _load_dotenv_with_fallback(path: Path, *, override: bool) -> None:
|
|
try:
|
|
# utf-8-sig strips a leading UTF-8 BOM (PowerShell 5.1 Set-Content -Encoding UTF8 /
|
|
# Notepad) and is a no-op otherwise. Plain "utf-8" would keep U+FEFF on the first key
|
|
# name and silently drop it from os.environ under its canonical name.
|
|
load_dotenv(dotenv_path=path, override=override, encoding="utf-8-sig")
|
|
except UnicodeDecodeError:
|
|
# utf-8-sig can't strip a BOM once we fall back to latin-1 decode.
|
|
raw = path.read_bytes()
|
|
if raw.startswith(codecs.BOM_UTF8):
|
|
raw = raw[len(codecs.BOM_UTF8) :]
|
|
load_dotenv(stream=io.StringIO(raw.decode("latin-1")), override=override)
|
|
# API keys must be pure ASCII (httpx encodes headers as ASCII); non-ASCII typically comes
|
|
# from copy-pasting keys from PDFs / rich-text editors that substitute lookalike glyphs.
|
|
_sanitize_loaded_credentials()
|
|
|
|
|
|
def _sanitize_env_file_if_needed(path: Path) -> None:
|
|
"""Pre-sanitize a .env file before python-dotenv reads it.
|
|
|
|
Encoding: sniffs a leading BOM *before* any text decode. UTF-16 (Notepad "Unicode") is decoded
|
|
correctly and rewritten as clean UTF-8. UTF-32 is refused (left untouched) so we never fall
|
|
through to the errors=replace corruption path. Order of BOM checks matters: UTF-32-LE's BOM
|
|
starts with UTF-16-LE's FF FE.
|
|
|
|
``hermes_cli.config._sanitize_env_lines`` normalizes line endings while treating content after
|
|
the first ``=`` as opaque for boundary discovery.
|
|
"""
|
|
if not path.exists():
|
|
return
|
|
try:
|
|
from hermes_cli.config import _sanitize_env_lines
|
|
except ImportError:
|
|
return # early bootstrap — config module not available yet
|
|
|
|
try:
|
|
raw = path.read_bytes()
|
|
except Exception:
|
|
return
|
|
|
|
# Sniff BOM bytes BEFORE decoding. ORDER MATTERS: BOM_UTF32_LE (FF FE 00 00) startswith
|
|
# BOM_UTF16_LE (FF FE); checking UTF-16 first would misdetect UTF-32-LE and mangle the file.
|
|
force_utf8_rewrite = False
|
|
if raw.startswith(codecs.BOM_UTF32_LE) or raw.startswith(codecs.BOM_UTF32_BE):
|
|
path_key = str(path.resolve())
|
|
if path_key not in _WARNED_UTF32_PATHS:
|
|
_WARNED_UTF32_PATHS.add(path_key)
|
|
logger.warning(
|
|
"Skipping .env sanitize for %s: UTF-32 BOM detected; "
|
|
"leaving file untouched to avoid corruption",
|
|
path,
|
|
)
|
|
return
|
|
if raw.startswith(codecs.BOM_UTF16_LE) or raw.startswith(codecs.BOM_UTF16_BE):
|
|
# "utf-16" uses the BOM to select endianness and strips it. TextIOWrapper + newline=None
|
|
# matches open()'s universal-newlines splitting (\n/\r\n/\r only — not splitlines()'s
|
|
# extra Unicode boundaries like U+2028), so sanitize sees the same lines as the UTF-8 path.
|
|
try:
|
|
with io.TextIOWrapper(io.BytesIO(raw), encoding="utf-16", newline=None) as f:
|
|
original = f.readlines()
|
|
except UnicodeDecodeError:
|
|
return
|
|
# Source is UTF-16 on disk; always rewrite as clean UTF-8 so the utf-8 dotenv load sees
|
|
# a canonical file.
|
|
force_utf8_rewrite = True
|
|
else:
|
|
# utf-8-sig strips a UTF-8 BOM; errors=replace so embedded NULs can be stripped below.
|
|
try:
|
|
with open(path, encoding="utf-8-sig", errors="replace") as f:
|
|
original = f.readlines()
|
|
except Exception:
|
|
return
|
|
# Defense-in-depth: errors=replace turns undecodable leading bytes into U+FFFD; persisting
|
|
# that glues replacement chars onto the first key name and rewrites the file permanently.
|
|
# Leave the file untouched rather than write the mangling.
|
|
if original and original[0].startswith("\ufffd"):
|
|
return
|
|
|
|
try:
|
|
# Strip NULs before _sanitize_env_lines so they never reach python-dotenv (which passes
|
|
# them to os.environ and crashes with ValueError). Also intentionally repairs BOM-less
|
|
# UTF-16 (NUL-padded ASCII) into clean UTF-8.
|
|
stripped = [line.replace("\x00", "") for line in original]
|
|
sanitized = _sanitize_env_lines(stripped)
|
|
if sanitized != original or force_utf8_rewrite:
|
|
import tempfile
|
|
fd, tmp = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".env_")
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
f.writelines(sanitized)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
atomic_replace(tmp, path)
|
|
except BaseException:
|
|
try:
|
|
os.unlink(tmp)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
except Exception:
|
|
pass # best-effort — don't block gateway startup
|
|
|
|
|
|
def load_hermes_dotenv(
|
|
*,
|
|
hermes_home: str | os.PathLike | None = None,
|
|
project_env: str | os.PathLike | None = None,
|
|
load_external_secrets: bool = True,
|
|
) -> list[Path]:
|
|
"""Load Hermes environment files with user config taking precedence.
|
|
|
|
``~/.hermes/.env`` overrides stale shell-exported values when present; project ``.env`` is a dev
|
|
fallback that only fills missing values when the user env exists, and overrides stale shell vars
|
|
when it does not.
|
|
"""
|
|
home_path = Path(hermes_home or os.getenv("HERMES_HOME", Path.home() / ".hermes"))
|
|
|
|
# A multiplex gateway hosts every profile in one process. While a routed profile-home override
|
|
# is active, copying that profile's .env into os.environ would expose its credentials to
|
|
# sibling turns and every subsequently spawned child. An unscoped startup load remains process
|
|
# configuration and keeps the normal path. External secret sources still need their refresh,
|
|
# so resolve them against the profile-local mapping instead of returning before hydration.
|
|
from agent.secret_scope import is_multiplex_active
|
|
from hermes_constants import get_hermes_home_override
|
|
|
|
if is_multiplex_active() and get_hermes_home_override() is not None:
|
|
home_key = str(home_path.resolve())
|
|
if home_key not in _SCOPED_SKIP_LOGGED:
|
|
_SCOPED_SKIP_LOGGED.add(home_key)
|
|
logger.debug(
|
|
"multiplex: skipping process-global dotenv load for routed "
|
|
"profile home %s (credentials resolve via the profile scope)",
|
|
home_path,
|
|
)
|
|
if load_external_secrets:
|
|
from hermes_cli import _early_recovery
|
|
|
|
if not _early_recovery._should_skip_external_secret_sources():
|
|
hydrate_profile_secret_sources(home_path)
|
|
return []
|
|
|
|
loaded: list[Path] = []
|
|
user_env = home_path / ".env"
|
|
project_env_path = Path(project_env) if project_env else None
|
|
|
|
# Normalize safe formatting and remove invalid NUL bytes before parsing.
|
|
if user_env.exists():
|
|
_sanitize_env_file_if_needed(user_env)
|
|
if project_env_path and project_env_path.exists():
|
|
_sanitize_env_file_if_needed(project_env_path)
|
|
|
|
if user_env.exists():
|
|
_load_dotenv_with_fallback(user_env, override=True)
|
|
loaded.append(user_env)
|
|
# Mirror reload_env() known-key cleanup so inherited Hermes keys absent from this
|
|
# profile's .env do not leak into the runtime.
|
|
_clear_known_keys_missing_from_dotenv(user_env)
|
|
|
|
# Load .op.env AFTER .env so .env values win, but the bootstrap token (OP_SERVICE_ACCOUNT_TOKEN)
|
|
# is available for apply_onepassword_secrets() even in cron / subprocess environments that
|
|
# inherit no shell state. .op.env is gitignored — the token never enters the committed .env.
|
|
# A systemd `EnvironmentFile=-/path/to/.hermes/.op.env` takes precedence (override=False
|
|
# ensures .op.env never clobbers a token already in the environment).
|
|
op_env = home_path / ".op.env"
|
|
if op_env.exists() and not os.environ.get("OP_SERVICE_ACCOUNT_TOKEN"):
|
|
_load_dotenv_with_fallback(op_env, override=False)
|
|
|
|
if project_env_path and project_env_path.exists():
|
|
_load_dotenv_with_fallback(project_env_path, override=not loaded)
|
|
loaded.append(project_env_path)
|
|
|
|
# External secret sources are skipped in two updater situations (dotenv and managed env still
|
|
# load; only external source resolution is unnecessary for the updater):
|
|
# 1. ``load_external_secrets=False`` — an ``update`` invocation that must not import optional
|
|
# secret-manager libraries (Bitwarden → cryptography → ``_rust.pyd``) into the process that
|
|
# replaces that same environment on Windows.
|
|
# 2. A fresh ``hermes update`` retry just completed a deferred dependency install before
|
|
# importing this module. Mapping native secret-source deps in that same updater process
|
|
# would make the self-lock preflight recreate the marker and exit 2 again.
|
|
from hermes_cli import _early_recovery
|
|
|
|
if load_external_secrets and not _early_recovery._should_skip_external_secret_sources():
|
|
_apply_external_secret_sources(home_path)
|
|
_apply_managed_env()
|
|
|
|
# config.yaml is the source of truth for terminal.* settings, but the dotenv loads above run
|
|
# with override=True — so a stale TERMINAL_ENV=docker left in ~/.hermes/.env (e.g. by an older
|
|
# `hermes setup`) would silently win on every reload. Startup launchers bridge config→env
|
|
# once, but long-lived processes (gateway per-turn reload, cron standalone runs) call this
|
|
# repeatedly and used to flip the effective backend back to the stale .env value mid-session.
|
|
# Re-apply config.yaml's explicit terminal keys LAST, after _apply_managed_env(), so the merged
|
|
# config (already carrying the managed overlay) is what lands in the env.
|
|
_reapply_terminal_config_bridge(home_path)
|
|
|
|
return loaded
|
|
|
|
|
|
def _reapply_terminal_config_bridge(home_path: Path) -> None:
|
|
"""Re-assert config.yaml's explicit ``terminal.*`` keys over reloaded .env.
|
|
|
|
Delegates to ``hermes_cli.config.apply_terminal_config_to_env`` — the single shared bridge
|
|
(terminal_tool's fallback and the TUI/dashboard launchers use it too) — so key coverage,
|
|
explicit-keys-only override semantics, cwd placeholder handling, and the managed-scope overlay
|
|
can't drift between bridge sites.
|
|
"""
|
|
try:
|
|
if Path(home_path).resolve() != _process_hermes_home().resolve():
|
|
return
|
|
from hermes_cli.config import apply_terminal_config_to_env
|
|
|
|
apply_terminal_config_to_env(env=None)
|
|
except Exception: # noqa: BLE001 — early bootstrap / malformed config
|
|
pass
|
|
|
|
|
|
def _apply_managed_env() -> None:
|
|
"""Apply the managed-scope .env last, with override, so it beats user/shell.
|
|
|
|
This does NOT stop the agent from later mutating ``os.environ`` in-process or ``export``-ing in
|
|
a subprocess shell; that hard boundary is a documented v2 item (design §8.1) — v1 relies on
|
|
filesystem permissions only. Fail-open: a missing managed dir/.env is the common no-op case,
|
|
and any error is swallowed so managed scope can never block startup.
|
|
"""
|
|
try:
|
|
from hermes_cli import managed_scope
|
|
|
|
managed_dir = managed_scope.get_managed_dir()
|
|
except Exception: # noqa: BLE001 — managed scope must never block startup
|
|
return
|
|
if managed_dir is None:
|
|
return
|
|
managed_env = managed_dir / ".env"
|
|
if not managed_env.exists():
|
|
return
|
|
_sanitize_env_file_if_needed(managed_env)
|
|
_load_dotenv_with_fallback(managed_env, override=True)
|
|
|
|
|
|
def _apply_external_secret_sources(home_path: Path) -> None:
|
|
"""Pull secrets from every enabled external source into env.
|
|
|
|
Runs AFTER dotenv loads so .env values are visible (sources use them to locate bootstrap tokens)
|
|
but BEFORE the rest of Hermes reads ``os.environ`` for credentials. Any failure is logged and
|
|
swallowed — external secret sources must never block startup.
|
|
|
|
Source ordering, mapped-beats-bulk precedence, first-claim-wins conflicts, override semantics
|
|
and provenance live in ``agent.secret_sources.registry.apply_all``; this wrapper owns the
|
|
once-per-HERMES_HOME guard, the post-apply ASCII sweep, the ``_SECRET_SOURCES`` provenance map
|
|
that UI surfaces read, and the startup status lines.
|
|
"""
|
|
home_key = str(Path(home_path).resolve())
|
|
if home_key in _APPLIED_HOMES:
|
|
return
|
|
|
|
try:
|
|
cfg = _load_secrets_config(home_path)
|
|
except Exception: # noqa: BLE001 — config errors must not block startup
|
|
# Deliberately NOT marked applied: a malformed config.yaml would otherwise permanently
|
|
# disable secret loading for this process even after the user fixes the file.
|
|
return
|
|
if not cfg:
|
|
# No secrets section (or everything disabled at parse level). Not marked applied either —
|
|
# the re-parse is a cheap fast_safe_load, and leaving the home unmarked lets a process
|
|
# pick up a config change on its next load_hermes_dotenv() call instead of never.
|
|
return
|
|
|
|
# Defer the registry import until a source is enabled — agent.secret_sources.bitwarden eagerly
|
|
# loads cryptography._rust.pyd, which makes the Windows updater self-lock before its preflight.
|
|
# Enabled-ness is detected by *shape* (source dict with enabled flag) rather than hardcoded
|
|
# names, so plugin/test sources pass through unknown keys and a generic dict entry never
|
|
# forces the crypto load on every launch.
|
|
any_enabled = any(isinstance(v, dict) and v.get("enabled") is True for v in cfg.values())
|
|
if not any_enabled:
|
|
return
|
|
|
|
try:
|
|
from agent.secret_sources.registry import apply_all
|
|
except ImportError:
|
|
return
|
|
|
|
try:
|
|
report = apply_all(cfg, home_path)
|
|
except Exception: # noqa: BLE001 — belt-and-braces; apply_all shouldn't raise
|
|
return
|
|
|
|
if not report.sources:
|
|
# Config parsed but no source is enabled: keep retrying cheaply (no fetch happens for
|
|
# disabled sources) so flipping a source on mid-process takes effect on the next call.
|
|
return
|
|
|
|
# A real fetch attempt happened (success OR error). Mark the home now so the 3-5 import-time
|
|
# load_hermes_dotenv() calls per startup don't re-fetch / re-print — error retries within one
|
|
# process are opt-in via reset_secret_source_cache(). Marking AFTER the attempt (not before)
|
|
# is what keeps the earlier failure paths retryable.
|
|
_APPLIED_HOMES.add(home_key)
|
|
|
|
if report.applied_any:
|
|
# Vault values are user-supplied and may carry the same copy-paste corruption as a
|
|
# manually edited .env — re-run the ASCII sanitization pass.
|
|
_sanitize_loaded_credentials()
|
|
# Remember where each var came from so setup / `hermes model` can label detected
|
|
# credentials "(from Bitwarden)" / "(from 1Password)" instead of a bare "credentials ✓".
|
|
values: dict[str, str] = {}
|
|
for name, applied in report.provenance.items():
|
|
_SECRET_SOURCES[name] = applied.source
|
|
if name in os.environ:
|
|
values[name] = os.environ[name]
|
|
_SECRET_SOURCE_VALUES_BY_HOME[home_key] = values
|
|
|
|
for src in report.sources:
|
|
if src.applied:
|
|
print(
|
|
f" {src.label}: applied {len(src.applied)} "
|
|
f"secret{'s' if len(src.applied) != 1 else ''}",
|
|
file=sys.stderr,
|
|
)
|
|
if src.result.error:
|
|
print(f" {src.label}: {src.result.error}", file=sys.stderr)
|
|
hint = _remediation_hint(src.name, src.result.error_kind, cfg, scope=home_key)
|
|
if hint:
|
|
print(f" {src.label}: → {hint}", file=sys.stderr)
|
|
for warn in src.result.warnings:
|
|
print(f" {src.label}: {warn}", file=sys.stderr)
|
|
for conflict in report.conflicts:
|
|
print(f" Secret sources: {conflict}", file=sys.stderr)
|
|
|
|
|
|
def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict, *, scope: str | None = None) -> str:
|
|
"""Ask the failed source for its one-line fix-it hint.
|
|
|
|
remediation() is a pure mapping and shouldn't raise, but a plugin source could — and startup
|
|
must never break on a status line.
|
|
"""
|
|
try:
|
|
from agent.secret_sources.registry import get_source
|
|
|
|
source = get_source(source_name, scope=scope)
|
|
if source is None:
|
|
return ""
|
|
src_cfg = secrets_cfg.get(source_name)
|
|
src_cfg = src_cfg if isinstance(src_cfg, dict) else {}
|
|
return str(source.remediation(error_kind, src_cfg) or "").strip()
|
|
except Exception: # noqa: BLE001 — hints must never block startup
|
|
return ""
|
|
|
|
|
|
def _load_secrets_config(home_path: Path) -> dict:
|
|
"""Read just the ``secrets:`` section out of config.yaml, isolated from the main config loader
|
|
so a malformed config can't take down dotenv loading entirely."""
|
|
config_path = home_path / "config.yaml"
|
|
if not config_path.exists():
|
|
return {}
|
|
# Prefer the shared (mtime, size)-keyed raw-config cache — this is the first config.yaml read
|
|
# in a normal `hermes` startup, so populating it here lets main.py's early bridge and
|
|
# hermes_logging reuse the same parse (one per process instead of 3-4). Falls back to a direct
|
|
# isolated parse if the shared reader is unavailable (it also swallows parse errors -> {}).
|
|
if home_path == _process_hermes_home():
|
|
try:
|
|
from hermes_cli.config import read_raw_config
|
|
|
|
data = read_raw_config() or {}
|
|
return data.get("secrets") or {}
|
|
except Exception:
|
|
pass
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
data = fast_safe_load(f) or {}
|
|
except Exception: # noqa: BLE001
|
|
return {}
|
|
return data.get("secrets") or {}
|
|
|
|
|
|
def _process_hermes_home() -> Path:
|
|
"""The HERMES_HOME the shared config cache is keyed to."""
|
|
try:
|
|
from hermes_constants import get_hermes_home
|
|
|
|
return get_hermes_home()
|
|
except Exception:
|
|
return Path.home() / ".hermes"
|