mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
8235f484c9
Follow-up on the cherry-picked #36896 commits, wiring 1Password into the new registry as the reference *mapped* source: - OnePasswordSource adapter (shape=mapped, scheme=op): fetch-only — precedence, override semantics, conflict warnings, and env writes move to the orchestrator; apply_onepassword_secrets kept as legacy shim like Bitwarden's. - Registered in _ensure_builtin_sources; mapped op:// bindings now outrank bulk Bitwarden project dumps on contested vars. - _cache.py FetchResult/is_valid_env_name re-exported from base so there is exactly one canonical definition; bitwarden.py re-adapted onto the contributor's DiskCache substrate. - ErrorKind classification for op failures (auth/binary/empty/network). - Registry + conformance coverage for OnePasswordSource, incl. the headline multi-source test: both vaults claim the same var, mapped 1Password wins, conflict surfaced, provenance correct. - env_loader tests migrated off the legacy apply_* mocks onto the fetch layer; AUTHOR_MAP entry for @hwrdprkns.
276 lines
9.3 KiB
Python
276 lines
9.3 KiB
Python
"""Tests for the secret-source tracking in ``hermes_cli.env_loader``.
|
|
|
|
These cover the small public surface that lets `hermes model` / `hermes setup`
|
|
label detected credentials with their origin ("from Bitwarden") so users
|
|
don't see an unexplained "credentials ✓" line when their .env is empty.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from hermes_cli import env_loader # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_sources():
|
|
"""Each test starts with a clean source map and applied-home guard."""
|
|
env_loader._SECRET_SOURCES.clear()
|
|
env_loader.reset_secret_source_cache()
|
|
yield
|
|
env_loader._SECRET_SOURCES.clear()
|
|
env_loader.reset_secret_source_cache()
|
|
|
|
|
|
def test_get_secret_source_returns_none_for_untracked_var():
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") is None
|
|
|
|
|
|
def test_get_secret_source_returns_label_for_tracked_var():
|
|
env_loader._SECRET_SOURCES["ANTHROPIC_API_KEY"] = "bitwarden"
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "bitwarden"
|
|
|
|
|
|
def test_format_secret_source_suffix_empty_for_untracked():
|
|
# Credentials from .env or the shell shouldn't add noise — the
|
|
# implicit case stays unlabeled.
|
|
assert env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY") == ""
|
|
|
|
|
|
def test_format_secret_source_suffix_bitwarden_uses_proper_name():
|
|
env_loader._SECRET_SOURCES["ANTHROPIC_API_KEY"] = "bitwarden"
|
|
assert (
|
|
env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY")
|
|
== " (from Bitwarden)"
|
|
)
|
|
|
|
|
|
def test_format_secret_source_suffix_generic_label_for_future_sources():
|
|
# Future-proofing: a new secret source (e.g. "vault") should still
|
|
# produce a sensible label without needing to edit every call site.
|
|
env_loader._SECRET_SOURCES["OPENAI_API_KEY"] = "vault"
|
|
assert (
|
|
env_loader.format_secret_source_suffix("OPENAI_API_KEY")
|
|
== " (from vault)"
|
|
)
|
|
|
|
|
|
def test_format_secret_source_suffix_onepassword_uses_proper_name():
|
|
env_loader._SECRET_SOURCES["OPENAI_API_KEY"] = "onepassword"
|
|
assert (
|
|
env_loader.format_secret_source_suffix("OPENAI_API_KEY")
|
|
== " (from 1Password)"
|
|
)
|
|
|
|
|
|
def test_apply_external_secret_sources_records_bitwarden_origin(tmp_path, monkeypatch):
|
|
"""End-to-end: when the Bitwarden source fetches keys, applied vars
|
|
end up in ``_SECRET_SOURCES`` so the UI can label them."""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
|
|
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"secrets:\n"
|
|
" bitwarden:\n"
|
|
" enabled: true\n"
|
|
" project_id: test-project\n"
|
|
" access_token_env: BWS_ACCESS_TOKEN\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Stub the fetch layer under the SecretSource adapter.
|
|
import agent.secret_sources.bitwarden as bw_module
|
|
|
|
monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
|
|
monkeypatch.setattr(
|
|
bw_module,
|
|
"fetch_bitwarden_secrets",
|
|
lambda **_kw: ({"ANTHROPIC_API_KEY": "sk-ant-test"}, []),
|
|
)
|
|
|
|
from agent.secret_sources import registry as reg_module
|
|
|
|
reg_module._reset_registry_for_tests()
|
|
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "bitwarden"
|
|
assert (
|
|
env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY")
|
|
== " (from Bitwarden)"
|
|
)
|
|
|
|
|
|
def test_apply_external_secret_sources_noop_when_disabled(tmp_path, monkeypatch):
|
|
"""Disabled Bitwarden config must not touch the source map."""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"secrets:\n"
|
|
" bitwarden:\n"
|
|
" enabled: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") is None
|
|
|
|
|
|
def test_apply_external_secret_sources_dedupes_within_process(tmp_path, monkeypatch):
|
|
"""``load_hermes_dotenv()`` is called at module-import time from several
|
|
hot modules (cli.py, hermes_cli/main.py, run_agent.py, ...). The
|
|
Bitwarden status line previously printed once per call — 3-5x per
|
|
startup. The applied-home guard must short-circuit subsequent calls
|
|
so the heavy work (config re-parse, Bitwarden lookup, status print)
|
|
runs exactly once per HERMES_HOME per process.
|
|
"""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.test-token")
|
|
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
|
|
config_path = tmp_path / "config.yaml"
|
|
config_path.write_text(
|
|
"secrets:\n"
|
|
" bitwarden:\n"
|
|
" enabled: true\n"
|
|
" project_id: test-project\n"
|
|
" access_token_env: BWS_ACCESS_TOKEN\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def _fake_fetch(**_kwargs):
|
|
call_count["n"] += 1
|
|
return {"ANTHROPIC_API_KEY": "sk-ant-test"}, []
|
|
|
|
import agent.secret_sources.bitwarden as bw_module
|
|
monkeypatch.setattr(bw_module, "find_bws", lambda **_kw: Path("/fake/bws"))
|
|
monkeypatch.setattr(bw_module, "fetch_bitwarden_secrets", _fake_fetch)
|
|
|
|
from agent.secret_sources import registry as reg_module
|
|
|
|
reg_module._reset_registry_for_tests()
|
|
|
|
# Five calls in a row, simulating module-import-time invocations from
|
|
# cli.py, hermes_cli/main.py, run_agent.py, trajectory_compressor.py,
|
|
# gateway/run.py. Only the first should actually call the backend.
|
|
for _ in range(5):
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
|
|
assert call_count["n"] == 1, (
|
|
"Bitwarden backend was called {} time(s); expected exactly 1 — "
|
|
"the applied-home guard is broken.".format(call_count["n"])
|
|
)
|
|
|
|
# Source tracking still works after dedup.
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "bitwarden"
|
|
|
|
# reset_secret_source_cache() forces a fresh pull on the next call.
|
|
env_loader.reset_secret_source_cache()
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
assert call_count["n"] == 2
|
|
|
|
|
|
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``."""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
|
|
(tmp_path / "config.yaml").write_text(
|
|
"secrets:\n"
|
|
" onepassword:\n"
|
|
" enabled: true\n"
|
|
" env:\n"
|
|
" ANTHROPIC_API_KEY: 'op://Private/Anthropic/credential'\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
import agent.secret_sources.onepassword as op_module
|
|
|
|
monkeypatch.setattr(op_module, "find_op", lambda *_a, **_kw: Path("/fake/op"))
|
|
monkeypatch.setattr(
|
|
op_module,
|
|
"fetch_onepassword_secrets",
|
|
lambda **_kw: ({"ANTHROPIC_API_KEY": "sk-ant-test"}, []),
|
|
)
|
|
|
|
from agent.secret_sources import registry as reg_module
|
|
|
|
reg_module._reset_registry_for_tests()
|
|
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
|
|
assert env_loader.get_secret_source("ANTHROPIC_API_KEY") == "onepassword"
|
|
assert (
|
|
env_loader.format_secret_source_suffix("ANTHROPIC_API_KEY")
|
|
== " (from 1Password)"
|
|
)
|
|
|
|
|
|
def test_apply_external_secret_sources_survives_non_dict_section(tmp_path, monkeypatch):
|
|
"""A malformed `secrets:` section must not abort startup (fail-open).
|
|
|
|
Both `onepassword: true` (non-dict) and a bad bitwarden section must be
|
|
coerced to empty config instead of raising AttributeError up through
|
|
load_hermes_dotenv().
|
|
"""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
(tmp_path / "config.yaml").write_text(
|
|
"secrets:\n"
|
|
" bitwarden: true\n"
|
|
" onepassword: true\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# Must not raise and must not record anything.
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
assert env_loader.get_secret_source("ANYTHING") is None
|
|
|
|
|
|
def test_apply_external_secret_sources_bad_ttl_does_not_crash(tmp_path, monkeypatch):
|
|
"""A non-numeric cache_ttl_seconds must be coerced, not crash startup."""
|
|
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
(tmp_path / "config.yaml").write_text(
|
|
"secrets:\n"
|
|
" onepassword:\n"
|
|
" enabled: true\n"
|
|
" cache_ttl_seconds: not-a-number\n"
|
|
" env:\n"
|
|
" K: 'op://V/I/F'\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
captured = {}
|
|
|
|
def _fake_fetch(**kwargs):
|
|
captured.update(kwargs)
|
|
return {}, []
|
|
|
|
import agent.secret_sources.onepassword as op_module
|
|
monkeypatch.setattr(op_module, "find_op", lambda *_a, **_kw: Path("/fake/op"))
|
|
monkeypatch.setattr(op_module, "fetch_onepassword_secrets", _fake_fetch)
|
|
|
|
from agent.secret_sources import registry as reg_module
|
|
|
|
reg_module._reset_registry_for_tests()
|
|
|
|
env_loader._apply_external_secret_sources(tmp_path)
|
|
|
|
# Coerced to the 300s default rather than raising ValueError.
|
|
assert captured["cache_ttl_seconds"] == 300
|