"""Helpers for loading Hermes .env files consistently across entrypoints.""" from __future__ import annotations import codecs import io import itertools import logging import os import sys import threading from pathlib import Path # Kept at module level on purpose: importing this module must fail when the dotenv install is # wiped (#57828) so early recovery provably runs before third-party imports (test_early_recovery). # The parser internals are imported lazily below because gateway tests stub ``sys.modules["dotenv"]``. import dotenv # noqa: F401 from utils import atomic_replace, load_yaml_file_readonly, mkstemp_beside logger = logging.getLogger(__name__) # The ONLY env vars sanitized on load: credentials must be pure ASCII (they become HTTP header values); # arbitrary user env vars are never silently altered. _CREDENTIAL_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_KEY") # Once-per-process guards: load_hermes_dotenv() runs repeatedly (user + project env, gateway hot-reload, # lazy imports mid-turn, tests) so warnings/logs fire once per key/path/home. _WARNED_KEYS: set[str] = set() # credential names already given the non-ASCII warning _WARNED_UTF32_PATHS: set[str] = set() # .env paths already given the UTF-32 refuse-to-mangle warning _SCOPED_SKIP_LOGGED: set[str] = set() # routed profile homes whose multiplex dotenv skip was logged # env-var name → source label ("bitwarden", …) for externally injected credentials; setup / `hermes # model` tell users WHERE a key came from when .env lacks it. _SECRET_SOURCES: dict[str, str] = {} # Every env-var name an external source SUPPLIED for some home, whether it was applied or lost to a # pre-existing process value (``skipped_existing``). ``_SECRET_SOURCES`` is provenance metadata and only # names applied values; the scrub that keeps a launch profile's source-supplied names out of a routed # child must see the skipped ones too, or a name already in the process env leaks with the launch value. _SOURCE_SUPPLIED_NAMES: set[str] = set() # Every KEY name a dotenv file loaded into ``os.environ`` during this process's lifetime. A key removed # or renamed in the launch ``.env`` after boot stays in ``os.environ`` (dotenv never unsets), but a # re-parse of the current file no longer names it — so the launch-residue strip for a routed child must # work from what was LOADED, not from what the file says now. Additive for the process lifetime. _LOADED_DOTENV_KEYS: set[str] = set() # KEY names loaded from the administrator-managed ``.env`` (``_apply_managed_env``). Kept OUT of the launch # residue: those values are policy that beats the user's own ``.env`` for every profile, so a routed child # must keep them — and keep them LAST, over the routed profile's scope (review on f5f88d5058). _MANAGED_DOTENV_KEYS: set[str] = set() # Immutable per-home snapshots: os.environ is shared across profiles and a later home's apply may overwrite it. _SECRET_SOURCE_VALUES_BY_HOME: dict[str, dict[str, str]] = {} # Per home: the subset of the snapshot a dotenv reload may re-assert — see ``AppliedVar.authoritative`` (#74265). _SECRET_SOURCE_RESTORE_BY_HOME: dict[str, dict[str, str]] = {} # Per home: what the process-global path WROTE into ``os.environ`` for an external source — name → # (source name, value written, value the name held before the first write or None). The per-home # snapshot above is republished on every pass, but ``os.environ`` is not: without this record a value a # since-removed or disabled source injected stayed in the process env, and single-profile # ``get_secret()`` kept serving it through its ``os.environ`` fallback until restart. Deliberately NOT # cleared by a per-home ``reset_secret_source_cache``: it describes ``os.environ``, which a reset does not # touch, and the next pass needs it to revoke. _SECRET_SOURCE_WRITES_BY_HOME: dict[str, dict[str, tuple[str, str, str | None]]] = {} # HERMES_HOME paths already pulled external secrets for: load_hermes_dotenv() runs at import time from # several hot modules, so without this 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() # What THIS process has published from dotenv files, per variable: (value before our first publish, or # None if absent; last value we published; load pass that published it). Reloads run per gateway turn and # per cron fire, and a line like ``PATH=/x:${PATH}`` interpolated against an environ that already holds the # previous reload's output grows by ``/x:`` every time until child spawns fail with E2BIG (#109902). A new # pass resolves against the baseline instead — but only where the environ still holds exactly what we # published, so a value the shell, config bridge, or an external secret source changed since is not frozen. # One process-wide record (not per home/project scope): the environ is process-wide, so alternating # callers (gateway with a project .env, cron without; home A then B) must peel each other's output too. _DOTENV_PUBLISHED: dict[str, tuple[str | None, str, int]] = {} _DOTENV_PASSES = itertools.count() _DOTENV_LOCK = threading.RLock() # Per-process credentials a parent mints and injects into the child's environment (the Desktop shell / # a link-style launcher spawns `hermes dashboard` with a fresh HERMES_DASHBOARD_SESSION_TOKEN and keeps # the same token for its own /api probes). They are never .env configuration, so a persisted value in # ~/.hermes/.env must not replace an injected one — the parent would then 401 against its own child # (#115955). A value an earlier dotenv pass published is still reloaded normally. _SPAWN_CREDENTIAL_KEYS: frozenset[str] = frozenset({"HERMES_DASHBOARD_SESSION_TOKEN"}) # Behavioral routing keys a parent Hermes process injects into child env that silently redirect a profile # onto the wrong provider path; these — and ONLY these — are scrubbed at startup when absent from the # profile's .env. Credentials are excluded: shell exports are a 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=``), via the same tokenizer that installs profile scopes — a key the installer sees is a key the dashboard scrub sees (BOM'd first line included).""" from agent.secret_scope import load_env_file return set(load_env_file(path)) def _clear_known_keys_missing_from_dotenv(path: Path) -> None: """After ``.env`` loaded with override, delete inherited ``_PROFILE_MANAGED_ENV_KEYS`` it does not define. Deliberately NARROW: only keys that change *which provider path* is used. Does **not** run when the ``.env`` file does not exist (bare-profile case, which follows ``#66930`` / ``#67027`` semantics). """ 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: """Source label that supplied ``env_var`` (``"bitwarden"`` …), None for .env/shell keys. Metadata only — never authorization to persist the raw value.""" return _SECRET_SOURCES.get(env_var) def _record_supplied_names(report) -> set[str]: """Every name *report*'s sources supplied — applied, or skipped because a value already existed — recorded into ``_SOURCE_SUPPLIED_NAMES`` so the routed-child scrub knows the source owns it.""" supplied = set(report.provenance) for src in report.sources: supplied.update(src.skipped_existing) _SOURCE_SUPPLIED_NAMES.update(supplied) return supplied def secret_source_names() -> tuple[str, ...]: """Every env-var name some profile's external secret source APPLIED (names only — the map is process-wide, so a value must be resolved through the active profile's secret scope). Consumers that forward source values into a child (MCP stdio env) want exactly these; see ``source_supplied_names`` for the wider set the routed-child scrub needs.""" return tuple(_SECRET_SOURCES) def source_supplied_names() -> tuple[str, ...]: """Every env-var name an external source supplied for any home — applied, or lost to a pre-existing process value (``skipped_existing``). The launch value in ``os.environ`` is still not a routed profile's to inherit, so the strip must see the skipped names too.""" return tuple(sorted(set(_SECRET_SOURCES) | _SOURCE_SUPPLIED_NAMES)) def launch_dotenv_keys() -> frozenset[str]: """KEY names any NON-managed dotenv file loaded into this process's ``os.environ`` so far (see ``_LOADED_DOTENV_KEYS``); the launch profile's residue set for routed children.""" return frozenset(_LOADED_DOTENV_KEYS) def managed_dotenv_keys() -> frozenset[str]: """KEY names the administrator-managed ``.env`` loaded (see ``_MANAGED_DOTENV_KEYS``). Policy for every profile: never stripped from a routed child, and re-applied over the routed scope.""" return frozenset(_MANAGED_DOTENV_KEYS) 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 route turns to profiles that never ran the process-global dotenv path, so resolve against a private mapping seeded from that ``.env`` and record the per-home snapshot for ``build_profile_secret_scope()``. Fail-open / once-per-home like ``_apply_external_secret_sources``; never returns 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) # A retry must not keep serving a partial result after the source is removed, disabled, or can no # longer be evaluated. Publish only the snapshot established by this attempt. _SECRET_SOURCE_VALUES_BY_HOME.pop(home_key, None) _SECRET_SOURCE_RESTORE_BY_HOME.pop(home_key, None) 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 (1Password token lives in gitignored .op.env) # or cold profiles fail 1Password hydration. .env wins. # Without seeding it here a cold profile configured for the supported .op.env flow fails 1Password # hydration (sweeper review on #74549). .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 {} # Routed profiles have no runtime reset path. Keep a failed source retryable so correcting its # profile-local bootstrap credentials takes effect on the next turn; successful sources from a # mixed report are still snapshotted below and can be used while the failed source recovers. if all(src.result.ok for src in report.sources): _APPLIED_HOMES.add(home_key) # Same ownership bookkeeping as the process-global path: a name this profile's source supplied — applied, # or skipped because the private mapping already had it — is a source-owned name the routed-child scrub # must know about, or a sibling still inherits the launch value for it (review on f5f88d5058). _record_supplied_names(report) 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 _SECRET_SOURCE_VALUES_BY_HOME[home_key] = values return dict(values) def reset_secret_source_cache(hermes_home: str | os.PathLike | None = None) -> None: """Forget applied homes so the next load re-pulls (tests, long-running processes after config edits). ``hermes_home`` limits the reset to ONE home: a multiplex gateway keeps every profile's snapshot in this process, and a per-fire cron re-pull or a plugin-discovery refresh for one home must not wipe a sibling's hydrated snapshot — the sibling's next scope build would run empty until it re-hydrated (#102041).""" if hermes_home is None: _APPLIED_HOMES.clear() _SECRET_SOURCES.clear() _SOURCE_SUPPLIED_NAMES.clear() _SECRET_SOURCE_VALUES_BY_HOME.clear() _SECRET_SOURCE_RESTORE_BY_HOME.clear() _SECRET_SOURCE_WRITES_BY_HOME.clear() return home_key = str(Path(hermes_home).resolve()) _APPLIED_HOMES.discard(home_key) _SECRET_SOURCE_VALUES_BY_HOME.pop(home_key, None) _SECRET_SOURCE_RESTORE_BY_HOME.pop(home_key, None) def format_secret_source_suffix(env_var: str) -> str: """``" (from Bitwarden)"``-style suffix; ``""`` for .env/shell keys (only external sources are named).""" source = get_secret_source(env_var) if not source: return "" if source == "bitwarden": return " (from Bitwarden)" # Registry label (e.g. "1Password"); raw name for unknown sources (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 from credential env vars (``_CREDENTIAL_SUFFIXES``) so the codebase never sees them. Emits a one-line warning to stderr when characters are stripped. Silent stripping would mask copy-paste corruption (Unicode lookalike glyphs from PDFs / rich-text editors, ZWSP from web pages) as opaque provider-side "invalid API key" errors (see #6843). """ 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, load_pass: int | None = None, managed: bool = False, ) -> None: """Load one dotenv file into ``os.environ`` like ``dotenv.load_dotenv`` — same parser, same ``${VAR}`` / ``${VAR:-default}`` / precedence rules — except that ``${VAR}`` resolves against the value VAR had before this process's earlier passes published it (see ``_DOTENV_PUBLISHED``). ``load_pass`` groups the layered files of one ``load_hermes_dotenv`` call: within a pass a later layer (project, managed) still sees the earlier layer's output, as it always did; only OTHER passes' output is peeled. A bare call (``hermes send``'s direct reload) is its own pass.""" raw = path.read_bytes() try: # utf-8-sig strips a leading BOM (PowerShell 5.1 / Notepad); plain utf-8 would keep U+FEFF on the # first key name and silently drop it from os.environ under its canonical name. text = raw.decode("utf-8-sig") except UnicodeDecodeError: if raw.startswith(codecs.BOM_UTF8): # strip the BOM by hand: utf-8-sig can't once we decode latin-1 raw = raw[len(codecs.BOM_UTF8) :] text = raw.decode("latin-1") # Imported here, not at module level: gateway tests stub ``sys.modules["dotenv"]`` with a bare module # exposing only ``load_dotenv``, and ``gateway.run`` imports this module at import time. from dotenv.main import DotEnv from dotenv.variables import parse_variables assignments = list(DotEnv(dotenv_path=None, stream=io.StringIO(text), interpolate=False).parse()) with _DOTENV_LOCK: if load_pass is None: load_pass = next(_DOTENV_PASSES) lookup_env: dict[str, str | None] = dict(os.environ) for name, (baseline, published, published_pass) in _DOTENV_PUBLISHED.items(): if published_pass != load_pass and lookup_env.get(name) == published: if baseline is None: del lookup_env[name] # absent, so ``${VAR:-default}`` takes the default again else: lookup_env[name] = baseline resolved: dict[str, str | None] = {} for name, value in assignments: if value is not None: # mirrors dotenv.main.resolve_variables, minus the live os.environ lookup = {**lookup_env, **resolved} if override else {**resolved, **lookup_env} value = "".join(atom.resolve(lookup) for atom in parse_variables(value)) resolved[name] = value for name, value in resolved.items(): if value is None or (not override and name in os.environ): continue current = os.environ.get(name) record = _DOTENV_PUBLISHED.get(name) ours = record is not None and current == record[1] if name in _SPAWN_CREDENTIAL_KEYS and current and not ours: continue # parent-minted per-process credential: .env must not split it from the parent # Ours and untouched since → keep the original baseline; anything else is a newer outside value. baseline = record[0] if ours else current os.environ[name] = value _DOTENV_PUBLISHED[name] = (baseline, value, load_pass) # Every key this file defines, for the launch-residue strip: dotenv never unsets, so a key later # removed from the launch .env stays in os.environ and a re-parse of the file no longer names it. # Managed keys are recorded separately: they are administrator policy, not launch-profile residue. (_MANAGED_DOTENV_KEYS if managed else _LOADED_DOTENV_KEYS).update(name for name, _value in assignments) _sanitize_loaded_credentials() # httpx encodes headers as ASCII def _sanitize_env_file_if_needed(path: Path) -> None: """Pre-sanitize a .env file before python-dotenv reads it. Sniffs a leading BOM *before* any text decode: UTF-16 (Notepad "Unicode") is rewritten as clean UTF-8; UTF-32 is refused (left untouched) so we never fall through to the errors=replace corruption path.""" 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 # ORDER MATTERS: BOM_UTF32_LE (FF FE 00 00) startswith BOM_UTF16_LE (FF FE); UTF-16 first would mangle it. force_utf8_rewrite = False if raw.startswith(codecs.BOM_UTF32_LE) or raw.startswith(codecs.BOM_UTF32_BE): # Lazy import keeps the module import block identical to #65124's codecs/io additions so the two PRs # auto-merge either order. 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 for endianness and strips it; newline=None matches open()'s universal # newlines (not splitlines()'s extra boundaries like U+2028) so sanitize sees the same lines. try: with io.TextIOWrapper(io.BytesIO(raw), encoding="utf-16", newline=None) as f: original = f.readlines() except UnicodeDecodeError: return force_utf8_rewrite = True # always rewrite UTF-16 as UTF-8 so the dotenv load sees a canonical file 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 # errors=replace turns undecodable leading bytes into U+FFFD; persisting would glue them onto # the first key name permanently — leave the file untouched instead. if original and original[0].startswith("\ufffd"): return try: # Strip NULs (os.environ raises ValueError on them); also repairs BOM-less UTF-16 (NUL-padded ASCII). stripped = [line.replace("\x00", "") for line in original] sanitized = _sanitize_env_lines(stripped) if sanitized != original or force_utf8_rewrite: fd, tmp = mkstemp_beside(path, 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 env files: ``~/.hermes/.env`` overrides stale shell exports; project ``.env`` is a dev fallback that only fills gaps when the user env exists (and overrides shell vars when it does not).""" # Process home on purpose (never the per-turn override): a startup .env load must not follow a routed # profile — see the multiplex guard below. from hermes_constants import get_process_hermes_home home_path = Path(hermes_home) if hermes_home else get_process_hermes_home() # Multiplex gateway: 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 spawned child. The launch # home's own .env is process configuration and still loads: the launch profile's scoped bodies bind # an override naming the launch home too, and skipping it hid launch-only credentials such as a # fallback_providers key from the process env (#125530). Both the load's target AND the active # home must be the launch home: a launch-targeted load inside a FOREIGN turn re-bridges terminal.* # from the config the override resolves to, i.e. the routed profile's cwd into the shared env. # External sources still refresh against the profile mapping. # (``is_multiplex_active()`` is also true, context-locally, for a routed cron fire in the desktop # backend — see ``cron.scheduler_provider.routed_profile_fire``.) from agent.secret_scope import is_multiplex_active from hermes_constants import get_hermes_home, get_hermes_home_override launch_home = _process_hermes_home().resolve() if (is_multiplex_active() and get_hermes_home_override() is not None and (home_path.resolve() != launch_home or get_hermes_home().resolve() != launch_home)): 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 load_pass = next(_DOTENV_PASSES) # one pass: later layers below see the earlier layers' output if user_env.exists(): # normalize formatting / strip NULs before parsing _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, load_pass=load_pass) loaded.append(user_env) _clear_known_keys_missing_from_dotenv(user_env) # mirrors reload_env(): inherited keys must not leak # .op.env AFTER .env so .env wins, but the bootstrap OP_SERVICE_ACCOUNT_TOKEN reaches # apply_onepassword_secrets() even in cron with no shell state; gitignored so the token never enters # the committed .env. override=False lets a systemd `EnvironmentFile=-…/.op.env` token win. 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, load_pass=load_pass) if project_env_path and project_env_path.exists(): _load_dotenv_with_fallback(project_env_path, override=not loaded, load_pass=load_pass) loaded.append(project_env_path) # The override=True loads above wrote the raw .env line (``__BITWARDEN_MANAGED__`` placeholder, stale # token) back over a value an external source resolved on an earlier call, and the source pass below is a # once-per-home no-op — so the clobber stuck for the life of the process (#74265). Re-assert only what the # source is authoritative for; managed scope, applied last with override=True, still wins on purpose. if _SECRET_SOURCE_RESTORE_BY_HOME: for name, value in _SECRET_SOURCE_RESTORE_BY_HOME.get(str(home_path.resolve()), {}).items(): if os.environ.get(name) != value: os.environ[name] = value # External sources are skipped for the updater (dotenv + managed env still load): ``update`` must not # import optional secret-manager libs (Bitwarden → cryptography → _rust.pyd) into the process replacing # that env on Windows, and a fresh retry after a deferred dependency install would otherwise make the # self-lock preflight exit 2 again. from hermes_cli import _early_recovery # External secret sources are skipped in two updater situations: 1. ``load_external_secrets=False`` — # the caller is 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 (#73381, #86735). 2. A fresh ``hermes update`` retry just completed a deferred dependency # install before importing this module. Do not remap native secret-source dependencies in that same # updater process or the self-lock preflight will recreate the marker and exit 2 again. Dotenv and # managed env still load in both cases; only external source resolution is unnecessary for the updater. if load_external_secrets and not _early_recovery._should_skip_external_secret_sources(): _apply_external_secret_sources(home_path) _apply_managed_env(load_pass=load_pass) # config.yaml owns terminal.*, but the override=True loads above let a stale TERMINAL_ENV=docker in # ~/.hermes/.env win on every reload and flip the backend mid-session in long-lived processes. # Re-apply the explicit terminal keys LAST, after the managed overlay, so the merged config lands. # config.yaml is the documented 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. written by an older # `hermes setup` before the user switched terminal.backend in config.yaml) silently wins again on every # reload. Startup launchers bridge config→env once, but long-lived processes (gateway per-turn reload, # cron standalone runs) call load_hermes_dotenv() repeatedly and used to flip the effective backend back # to the stale .env value mid-session (#29186, #67323). _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 via the single shared bridge ``apply_terminal_config_to_env`` (also used by terminal_tool and the TUI/dashboard launchers) so the semantics can't drift between 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(*, load_pass: int | None = None) -> None: """Apply the managed-scope .env last, with override, so it beats user/shell. Does NOT stop the agent from later mutating os.environ (v1 relies on filesystem permissions). Fail-open: never blocks 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, load_pass=load_pass, managed=True) def _revoke_secret_source_writes(home_path: Path, *, keep) -> None: """Take back what a source wrote into ``os.environ`` for *home_path* once nothing backs it. ``keep(name, source)`` says which recorded writes are still owned. Every other write is revoked, but only while ``os.environ`` still holds exactly the value the source wrote: a value another owner replaced since (the profile's ``.env``, the administrator-managed ``.env``, a shell export, a later source) is theirs and stays. A revoked name gets back the value it held before the source's first write, or is removed when it had none.""" home_key = str(Path(home_path).resolve()) writes = _SECRET_SOURCE_WRITES_BY_HOME.get(home_key) if not writes: return try: from agent.secret_scope import load_env_file owned_by_dotenv = set(load_env_file(Path(home_path) / ".env")) | _MANAGED_DOTENV_KEYS except Exception: # noqa: BLE001 — unreadable .env: treat nothing as dotenv-owned owned_by_dotenv = set(_MANAGED_DOTENV_KEYS) kept: dict[str, tuple[str, str, str | None]] = {} for name, (source, value, prior) in writes.items(): if keep(name, source): kept[name] = (source, value, prior) continue if name not in owned_by_dotenv and os.environ.get(name) == value: if prior is None: os.environ.pop(name, None) else: os.environ[name] = prior still_attributed = any( other.get(name, ("",))[0] == source for key, other in _SECRET_SOURCE_WRITES_BY_HOME.items() if key != home_key) if _SECRET_SOURCES.get(name) == source and not still_attributed: _SECRET_SOURCES.pop(name, None) if kept: _SECRET_SOURCE_WRITES_BY_HOME[home_key] = kept else: _SECRET_SOURCE_WRITES_BY_HOME.pop(home_key, None) def _apply_external_secret_sources(home_path: Path) -> None: """Pull secrets from every enabled external source into env — AFTER dotenv (sources need .env bootstrap tokens), BEFORE Hermes reads credentials; failures never block startup. Precedence/conflicts/provenance live in ``registry.apply_all``; this wrapper owns the once-per-home guard, the post-apply ASCII sweep, the ``_SECRET_SOURCES`` map and status lines.""" home_key = str(Path(home_path).resolve()) if home_key in _APPLIED_HOMES: return # Neither early return marks the home applied: a malformed config.yaml would otherwise permanently # disable secret loading for this process, and an unmarked home picks up a config change on the next # load (the signature-cached read is cheap). try: cfg = _load_secrets_config(home_path) except Exception: # noqa: BLE001 — config errors must not block startup # See #40597. return # No source configured / enabled any more: whatever one wrote earlier is no longer backed (#126982 # review). A config READ failure above keeps them — fail-open, same as the fetch-failure path below. if not cfg: _revoke_secret_source_writes(home_path, keep=lambda _name, _source: False) return # Defer the registry import until a source is enabled — bitwarden eagerly loads cryptography._rust.pyd, # which makes the Windows updater self-lock before its preflight. Detect by *shape* (dict with enabled # flag), not names, so plugin/test sources pass and a plain dict entry never forces the crypto load. any_enabled = any(isinstance(v, dict) and v.get("enabled") is True for v in cfg.values()) if not any_enabled: _revoke_secret_source_writes(home_path, keep=lambda _name, _source: False) return try: from agent.secret_sources.registry import apply_all, enabled_source_names except ImportError: return # Revoke a removed/disabled source's writes BEFORE the pass: left in place, its stale value would make # a still-enabled source that supplies the same name skip it as pre-existing (``skipped_existing``). try: active = enabled_source_names(cfg, home_path) except Exception: # noqa: BLE001 — cannot tell which sources remain: keep everything (fail-open) active = None if active is not None: _revoke_secret_source_writes(home_path, keep=lambda _name, source: source in active) environ_before = dict(os.environ) try: report = apply_all(cfg, home_path) except Exception: # noqa: BLE001 — belt-and-braces; apply_all shouldn't raise return if not report.sources: # no source enabled: keep retrying cheaply so flipping one on takes effect return # A real fetch attempt happened (success OR error): mark the home so the 3-5 import-time calls per # startup don't re-fetch / re-print (error retries are opt-in via reset_secret_source_cache()). # Marking AFTER the attempt keeps the earlier failure paths retryable. _APPLIED_HOMES.add(home_key) if report.applied_any: _sanitize_loaded_credentials() # vault values carry the same copy-paste corruption risk as .env for name, applied in report.provenance.items(): _SECRET_SOURCES[name] = applied.source # Snapshot EVERY name a source supplied, not just the newly applied ones. A name the source supplied # but the pre-existing process value won (``skipped_existing``) is still this home's effective value # for it — and on the unscoped path os.environ IS this home's environment. Skipping those names # latched an EMPTY snapshot whenever the key was already in the env: after the first cron/plugin # re-pull (the previous apply's own write-back shadows every key) or from boot under systemd # ``EnvironmentFile=``. Under multiplex the scope is the only credential source, so an empty # snapshot failed every default-profile turn for the process lifetime (#102041). values: dict[str, str] = {} supplied = _record_supplied_names(report) for name in supplied: if name in os.environ: values[name] = os.environ[name] _record_secret_source_writes(home_key, report, supplied, environ_before) if values: _SECRET_SOURCE_VALUES_BY_HOME[home_key] = values _SECRET_SOURCE_RESTORE_BY_HOME[home_key] = { n: values[n] for n, a in report.provenance.items() if a.authoritative and n in 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 _record_secret_source_writes(home_key: str, report, supplied: set[str], environ_before: dict) -> None: """Refresh ``_SECRET_SOURCE_WRITES_BY_HOME`` after one process-global pass, revoking a write its still-enabled source has stopped supplying (rotated out, mapping removed). A source whose fetch FAILED this pass keeps its earlier writes: a transient vault outage is not a revocation.""" failed = {src.name for src in report.sources if not src.result.ok} previous = _SECRET_SOURCE_WRITES_BY_HOME.get(home_key, {}) _revoke_secret_source_writes( Path(home_key), keep=lambda name, source: name in report.provenance or name in supplied or source in failed) writes = {name: rec for name, rec in _SECRET_SOURCE_WRITES_BY_HOME.get(home_key, {}).items() if name not in report.provenance and os.environ.get(name) == rec[1]} for name, applied in report.provenance.items(): if name not in os.environ: continue before = environ_before.get(name) earlier = previous.get(name) # Re-applied over our own earlier write: the pre-source value is the one recorded then. prior = earlier[2] if earlier is not None and earlier[1] == before else before writes[name] = (applied.source, os.environ[name], prior) if writes: _SECRET_SOURCE_WRITES_BY_HOME[home_key] = writes else: _SECRET_SOURCE_WRITES_BY_HOME.pop(home_key, None) def _remediation_hint(source_name: str, error_kind, secrets_cfg: dict, *, scope: str | None = None) -> str: """The failed source's one-line fix-it hint; a plugin remediation() could raise and startup must not.""" 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 of config.yaml, isolated so a malformed config can't break dotenv.""" config_path = home_path / "config.yaml" if not config_path.exists(): return {} # Prefer the shared raw-config cache: this is the first config.yaml read of a normal startup, so # populating it lets main.py's early bridge and hermes_logging reuse one parse instead of 3-4. 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 # Routed profiles re-enter their scope on every poll/turn; only re-parse after the file changed. try: data = load_yaml_file_readonly(config_path) or {} except Exception: # noqa: BLE001 return {} return data.get("secrets") or {} def _process_hermes_home() -> Path: """The HERMES_HOME the running process was launched under. Must be the *true* process home, ignoring any context-local ``set_hermes_home_override`` a per-request task has installed. Both callers depend on that: * ``_reapply_terminal_config_bridge`` guards "only re-bridge config into the shared ``os.environ`` when THIS load is for the process's own profile". If this followed the task override, a per-turn handler scoped to a *secondary* profile (the multiplex dashboard serving every profile from one process) would satisfy the guard and bridge that profile's ``terminal.backend`` into the shared env — e.g. a ``local`` sibling profile clobbering the launch profile's ``TERMINAL_ENV=ssh`` while leaving its ``TERMINAL_SSH_*`` untouched, so the session silently runs commands locally (cross-profile terminal-backend leak). * ``_load_secrets_config`` uses it to decide whether the shared (mtime,size)-keyed config cache is safe to reuse; under an override it must fall through to an isolated parse of the scoped profile. ``hermes_constants.get_routing_process_hermes_home()`` is the override-immune resolver built for exactly this; delegate to it. It is also immune to a host that mirrors the served profile into the live ``HERMES_HOME`` env var (``pin_process_hermes_home``): without that, the mirrored profile satisfied the guard above and bridged ITS ``terminal.*`` into the shared env. """ try: from hermes_constants import get_routing_process_hermes_home return get_routing_process_hermes_home() except Exception: return Path.home() / ".hermes"