From 0583692c2d7ccc746893151f45796d5073a289d4 Mon Sep 17 00:00:00 2001 From: izumi0uu Date: Sat, 4 Jul 2026 13:41:41 +0800 Subject: [PATCH] fix(secrets): scope BWS-injected provider keys Snapshot values applied by external secret sources per resolved HERMES_HOME so a later profile cannot replace an earlier profile scope through shared os.environ. Keep provider and credential-pool fallback reads on the active secret scope, and fail closed on unscoped multiplex reads. Tests: scripts/run_tests.sh tests/test_env_loader_secret_sources.py tests/test_env_loader_op_bootstrap.py tests/agent/test_secret_scope.py tests/agent/test_credential_pool.py tests/tools/test_credential_pool_env_fallback.py tests/hermes_cli/test_xiaomi_provider.py tests/cron/test_run_one_job.py tests/hermes_cli/test_api_key_providers.py tests/gateway/test_multiplex_credential_isolation.py -q (395 passed) --- agent/credential_pool.py | 11 ++-- agent/secret_scope.py | 15 ++++- hermes_cli/config.py | 10 ++- hermes_cli/env_loader.py | 17 +++++ tests/agent/test_secret_scope.py | 34 ++++++++++ tests/hermes_cli/test_xiaomi_provider.py | 77 ++++++++++++++++++++++ tests/test_env_loader_secret_sources.py | 82 +++++++++++++++++++++++- 7 files changed, 238 insertions(+), 8 deletions(-) diff --git a/agent/credential_pool.py b/agent/credential_pool.py index a968b3ef968..15b1458860e 100644 --- a/agent/credential_pool.py +++ b/agent/credential_pool.py @@ -2309,9 +2309,10 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool def _get_env_prefer_dotenv(key: str) -> str: env_file = load_env() raw = env_file.get(key, "").strip() - env_val = os.environ.get(key, "").strip() + scoped_value = (_get_secret(key, "") or "").strip() # If .env contains an unresolved op:// reference, prefer the - # already-resolved value from os.environ (set by + # already-resolved value supplied by the active secret scope (or by + # os.environ in legacy single-profile mode), set by # load_hermes_dotenv() -> apply_onepassword_secrets()). The raw # "op://Vault/Item/field" string would otherwise win and every # provider auth attempt would receive a URL instead of a key. This @@ -2319,9 +2320,9 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool # references straight into .env rather than the secrets.onepassword # config block. For every non-op:// value the original # .env-takes-precedence behaviour is preserved unchanged. - if raw.startswith("op://") and env_val: - return env_val - return raw or _get_secret(key, "") or env_val + if raw.startswith("op://") and scoped_value: + return scoped_value + return raw or scoped_value # Honour user suppression — `hermes auth remove ` for an # env-seeded credential marks the env: source as suppressed so it diff --git a/agent/secret_scope.py b/agent/secret_scope.py index d8730db2bd6..8b376d5fcef 100644 --- a/agent/secret_scope.py +++ b/agent/secret_scope.py @@ -218,5 +218,18 @@ def build_profile_secret_scope(hermes_home: Path) -> Dict[str, str]: global vars are intentionally NOT copied in — ``get_secret`` reads those from ``os.environ`` directly, so the scope holds only profile secrets. """ - return load_env_file(Path(hermes_home) / ".env") + home = Path(hermes_home) + secrets = load_env_file(home / ".env") + try: + from hermes_cli.env_loader import get_secret_source_values + external_secrets = get_secret_source_values(home) + except Exception: + external_secrets = {} + + for key, value in external_secrets.items(): + if _is_global_env(key): + continue + secrets[key] = value + + return secrets diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 53de7286fe0..f17c6a0a550 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -8219,9 +8219,17 @@ def get_env_value_prefer_dotenv(key: str) -> Optional[str]: if val: return val try: - from agent.secret_scope import get_secret as _get_secret + from agent.secret_scope import ( + UnscopedSecretError, + get_secret as _get_secret, + ) + except Exception: + return os.environ.get(key) + try: return _get_secret(key) + except UnscopedSecretError: + raise except Exception: return os.environ.get(key) diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index cb234afc36a..e91c12adf7e 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -36,6 +36,9 @@ _WARNED_UTF32_PATHS: set[str] = set() # directly (otherwise the "credentials detected ✓" line looks identical to # the .env case and they don't know Bitwarden is wired up). _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 we've already pulled external secrets for during this # process. ``load_hermes_dotenv()`` is called at module-import time from @@ -60,6 +63,14 @@ def get_secret_source(env_var: str) -> str | None: 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``.""" + home_key = str(Path(hermes_home).resolve()) + return dict(_SECRET_SOURCE_VALUES_BY_HOME.get(home_key, {})) + + def reset_secret_source_cache() -> None: """Forget which HERMES_HOME paths have already had external secrets applied. @@ -71,6 +82,8 @@ def reset_secret_source_cache() -> None: that want to refresh 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: @@ -443,8 +456,12 @@ def _apply_external_secret_sources(home_path: Path) -> None: # flows can label detected credentials with "(from Bitwarden)" / # "(from 1Password)" — otherwise users see "credentials ✓" with # no hint the value came from a vault rather than .env. + 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: diff --git a/tests/agent/test_secret_scope.py b/tests/agent/test_secret_scope.py index 43dae703c4f..2a325aa693b 100644 --- a/tests/agent/test_secret_scope.py +++ b/tests/agent/test_secret_scope.py @@ -175,3 +175,37 @@ class TestEnvFileParsing: assert ss.build_profile_secret_scope(tmp_path) == { "ANTHROPIC_API_KEY": "sk-profile" } + + def test_build_profile_secret_scope_includes_home_external_secrets( + self, tmp_path, monkeypatch + ): + (tmp_path / ".env").write_text("XIAOMI_API_KEY=placeholder\n") + from hermes_cli import env_loader + + home_key = str(tmp_path.resolve()) + monkeypatch.setitem( + env_loader._SECRET_SOURCE_VALUES_BY_HOME, + home_key, + {"XIAOMI_API_KEY": "sk-from-bitwarden"}, + ) + + assert ss.build_profile_secret_scope(tmp_path) == { + "XIAOMI_API_KEY": "sk-from-bitwarden" + } + + def test_build_profile_secret_scope_ignores_other_home_external_secrets( + self, tmp_path, monkeypatch + ): + profile = tmp_path / "profile" + other = tmp_path / "other" + profile.mkdir() + other.mkdir() + from hermes_cli import env_loader + + monkeypatch.setitem( + env_loader._SECRET_SOURCE_VALUES_BY_HOME, + str(other.resolve()), + {"XIAOMI_API_KEY": "sk-other-profile"}, + ) + + assert ss.build_profile_secret_scope(profile) == {} diff --git a/tests/hermes_cli/test_xiaomi_provider.py b/tests/hermes_cli/test_xiaomi_provider.py index 4a5a7724ad0..f84631fe114 100644 --- a/tests/hermes_cli/test_xiaomi_provider.py +++ b/tests/hermes_cli/test_xiaomi_provider.py @@ -121,6 +121,83 @@ class TestXiaomiCredentials: creds = resolve_api_key_provider_credentials("xiaomi") assert creds["base_url"] == "https://custom.xiaomi.example/v1" + def test_resolve_credentials_reads_home_external_secret_scope( + self, tmp_path, monkeypatch + ): + """BWS-injected keys belong in the profile scope that loaded them.""" + from agent import secret_scope as ss + from hermes_cli import config as config_module + from hermes_cli import env_loader + + home = tmp_path / "hermes" + home.mkdir() + (home / ".env").write_text("", encoding="utf-8") + monkeypatch.setattr(config_module, "get_env_path", lambda: home / ".env") + config_module.invalidate_env_cache() + + monkeypatch.delenv("XIAOMI_BASE_URL", raising=False) + monkeypatch.setitem( + env_loader._SECRET_SOURCE_VALUES_BY_HOME, + str(home.resolve()), + {"XIAOMI_API_KEY": "sk-bws-xiaomi-12345678"}, + ) + + ss.set_multiplex_active(True) + token = ss.set_secret_scope(ss.build_profile_secret_scope(home)) + try: + creds = resolve_api_key_provider_credentials("xiaomi") + finally: + ss.reset_secret_scope(token) + ss.set_multiplex_active(False) + + assert creds["api_key"] == "sk-bws-xiaomi-12345678" + assert creds["source"] == "XIAOMI_API_KEY" + + def test_scoped_missing_key_does_not_fall_through_to_raw_env( + self, tmp_path, monkeypatch + ): + from agent import secret_scope as ss + from hermes_cli import config as config_module + + home = tmp_path / "hermes" + home.mkdir() + (home / ".env").write_text("", encoding="utf-8") + monkeypatch.setattr(config_module, "get_env_path", lambda: home / ".env") + config_module.invalidate_env_cache() + + monkeypatch.setenv("XIAOMI_API_KEY", "sk-other-profile-12345678") + monkeypatch.delenv("XIAOMI_BASE_URL", raising=False) + + ss.set_multiplex_active(True) + token = ss.set_secret_scope({}) + try: + creds = resolve_api_key_provider_credentials("xiaomi") + finally: + ss.reset_secret_scope(token) + ss.set_multiplex_active(False) + + assert creds["api_key"] == "" + + def test_unscoped_multiplex_read_fails_closed(self, tmp_path, monkeypatch): + from agent import secret_scope as ss + from hermes_cli import config as config_module + + home = tmp_path / "hermes" + home.mkdir() + (home / ".env").write_text("", encoding="utf-8") + monkeypatch.setattr(config_module, "get_env_path", lambda: home / ".env") + config_module.invalidate_env_cache() + + monkeypatch.setenv("XIAOMI_API_KEY", "sk-global-leak-12345678") + monkeypatch.delenv("XIAOMI_BASE_URL", raising=False) + + ss.set_multiplex_active(True) + try: + with pytest.raises(ss.UnscopedSecretError): + resolve_api_key_provider_credentials("xiaomi") + finally: + ss.set_multiplex_active(False) + # ============================================================================= # Model catalog (dynamic — no static list) diff --git a/tests/test_env_loader_secret_sources.py b/tests/test_env_loader_secret_sources.py index 2637f74690f..ffb5fcc0c10 100644 --- a/tests/test_env_loader_secret_sources.py +++ b/tests/test_env_loader_secret_sources.py @@ -7,6 +7,7 @@ don't see an unexplained "credentials ✓" line when their .env is empty. from __future__ import annotations +import os import sys from pathlib import Path @@ -24,9 +25,11 @@ from hermes_cli import env_loader # noqa: E402 def _reset_sources(): """Each test starts with a clean source map and applied-home guard.""" env_loader._SECRET_SOURCES.clear() + env_loader._SECRET_SOURCE_VALUES_BY_HOME.clear() env_loader.reset_secret_source_cache() yield env_loader._SECRET_SOURCES.clear() + env_loader._SECRET_SOURCE_VALUES_BY_HOME.clear() env_loader.reset_secret_source_cache() @@ -39,6 +42,27 @@ def test_get_secret_source_returns_label_for_tracked_var(): assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "bitwarden" +def test_get_secret_source_values_returns_home_snapshot_copy(tmp_path): + home_a = tmp_path / "profile-a" + home_b = tmp_path / "profile-b" + home_a.mkdir() + home_b.mkdir() + + env_loader._SECRET_SOURCE_VALUES_BY_HOME[str(home_a.resolve())] = { + "ANTHROPIC_API_KEY": "sk-profile-a" + } + + snapshot = env_loader.get_secret_source_values(home_a) + assert snapshot == { + "ANTHROPIC_API_KEY": "sk-profile-a" + } + assert env_loader.get_secret_source_values(home_b) == {} + snapshot["ANTHROPIC_API_KEY"] = "mutated" + assert env_loader.get_secret_source_values(home_a) == { + "ANTHROPIC_API_KEY": "sk-profile-a" + } + + def test_format_secret_source_suffix_empty_for_untracked(): # Credentials from .env or the shell shouldn't add noise — the # implicit case stays unlabeled. @@ -151,7 +175,6 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa ) call_count = {"n": 0} - def _fake_fetch(**_kwargs): call_count["n"] += 1 return {"ANTHROPIC_API_KEY": "sk-ant-test"}, [] @@ -177,6 +200,9 @@ def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypa # Source tracking still works after dedup. assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "bitwarden" + assert env_loader.get_secret_source_values(tmp_path) == { + "ANTHROPIC_API_KEY": "sk-ant-test" + } # reset_secret_source_cache() forces a fresh pull on the next call. env_loader.reset_secret_source_cache() @@ -224,6 +250,60 @@ def test_apply_external_secret_sources_status_line_suppresses_secret_names( assert "LEAK_THIS_TOKEN" not in err +def test_external_secret_values_are_isolated_between_homes(tmp_path, monkeypatch): + """A later apply for the same key must not mutate an earlier home snapshot.""" + from agent.secret_scope import build_profile_secret_scope + from agent.secret_sources.registry import AppliedVar, ApplyReport + from agent.secret_sources import registry as reg_module + + home_a = tmp_path / "profile-a" + home_b = tmp_path / "profile-b" + for home in (home_a, home_b): + home.mkdir() + (home / "config.yaml").write_text( + "secrets:\n test-source:\n enabled: true\n", + encoding="utf-8", + ) + + values = { + str(home_a.resolve()): "value-a", + str(home_b.resolve()): "value-b", + } + + def _fake_apply_all(_cfg, home_path): + value = values[str(Path(home_path).resolve())] + monkeypatch.setenv("SHARED_API_KEY", value) + return ApplyReport( + provenance={ + "SHARED_API_KEY": AppliedVar( + name="SHARED_API_KEY", + source="test-source", + shape="mapped", + overrode_env=True, + ) + } + ) + + monkeypatch.setattr(reg_module, "apply_all", _fake_apply_all) + + env_loader._apply_external_secret_sources(home_a) + env_loader._apply_external_secret_sources(home_b) + + assert os.environ["SHARED_API_KEY"] == "value-b" + assert env_loader.get_secret_source_values(home_a) == { + "SHARED_API_KEY": "value-a" + } + assert env_loader.get_secret_source_values(home_b) == { + "SHARED_API_KEY": "value-b" + } + assert build_profile_secret_scope(home_a) == { + "SHARED_API_KEY": "value-a" + } + assert build_profile_secret_scope(home_b) == { + "SHARED_API_KEY": "value-b" + } + + def test_apply_external_secret_sources_records_onepassword_origin(tmp_path, monkeypatch): """When the 1Password source resolves refs, applied vars end up in ``_SECRET_SOURCES`` labeled ``onepassword``."""