mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
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)
This commit is contained in:
parent
c6f9e0c748
commit
0583692c2d
7 changed files with 238 additions and 8 deletions
|
|
@ -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 <provider> <N>` for an
|
||||
# env-seeded credential marks the env:<VAR> source as suppressed so it
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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) == {}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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``."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue