Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions agent/credential_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -2309,19 +2309,20 @@ 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
# happens during a partial migration, or when the user wrote op://
# 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
Expand Down
15 changes: 14 additions & 1 deletion agent/secret_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
10 changes: 9 additions & 1 deletion hermes_cli/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
17 changes: 17 additions & 0 deletions hermes_cli/env_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,9 @@
# 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
Expand All @@ -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.

Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down
34 changes: 34 additions & 0 deletions tests/agent/test_secret_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,3 +175,37 @@ def test_build_profile_secret_scope(self, tmp_path):
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) == {}
77 changes: 77 additions & 0 deletions tests/hermes_cli/test_xiaomi_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,6 +121,83 @@ def test_custom_base_url_override(self, monkeypatch):
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)
Expand Down
98 changes: 97 additions & 1 deletion tests/test_env_loader_secret_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from __future__ import annotations

import os
import sys
from pathlib import Path

Expand All @@ -24,9 +25,11 @@
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()


Expand All @@ -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.
Expand Down Expand Up @@ -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"}, []
Expand All @@ -177,6 +200,9 @@ def _fake_fetch(**_kwargs):

# 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()
Expand Down Expand Up @@ -224,6 +250,76 @@ 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.base import FetchResult
from agent.secret_sources.registry import (
AppliedVar,
ApplyReport,
SourceReport,
)
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(
# Real apply_all always appends a SourceReport per enabled
# source; the env_loader guard (#40597) early-returns on an
# empty sources list, so the fake must match the real shape.
sources=[
SourceReport(
name="test-source",
label="Test Source",
result=FetchResult(),
applied=["SHARED_API_KEY"],
)
],
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``."""
Expand Down
Loading