mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-22 16:25:58 +00:00
0faf4c838c
Rework the encrypted cache onto the fallback that landed in #69051: one transport-only gate, encrypted tier replaces (never accompanies) the plaintext tier when enabled, warning carries the failure + cache age, in-process cache promoted on a stale hit, and clear_caches() (token rotation) also removes the encrypted file since its key derives from the rotated token.
1351 lines
46 KiB
Python
1351 lines
46 KiB
Python
"""Hermetic tests for the Bitwarden Secrets Manager integration.
|
|
|
|
We never hit GitHub or Bitwarden in tests — subprocess + urllib are
|
|
mocked so the suite stays fast and offline-safe. The "live" pull and
|
|
binary download are exercised manually by `hermes secrets bitwarden
|
|
setup` outside of pytest.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import zipfile
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
|
|
# Make the worktree importable without depending on the installed wheel.
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from agent.secret_sources import bitwarden as bw # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_caches():
|
|
bw._reset_cache_for_tests()
|
|
yield
|
|
bw._reset_cache_for_tests()
|
|
|
|
|
|
@pytest.fixture
|
|
def hermes_home(tmp_path, monkeypatch):
|
|
"""Point Hermes at an isolated home directory."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
# Some modules cache get_hermes_home; clear if needed.
|
|
import hermes_constants
|
|
if hasattr(hermes_constants, "_HERMES_HOME_CACHE"):
|
|
hermes_constants._HERMES_HOME_CACHE = None # type: ignore[attr-defined]
|
|
return home
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _platform_asset_name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"system,machine,libc_text,expected",
|
|
[
|
|
("Darwin", "x86_64", "",
|
|
f"bws-macos-universal-{bw._BWS_VERSION}.zip"),
|
|
("Darwin", "arm64", "",
|
|
f"bws-macos-universal-{bw._BWS_VERSION}.zip"),
|
|
("Linux", "x86_64", "glibc",
|
|
f"bws-x86_64-unknown-linux-gnu-{bw._BWS_VERSION}.zip"),
|
|
("Linux", "x86_64", "musl libc",
|
|
f"bws-x86_64-unknown-linux-musl-{bw._BWS_VERSION}.zip"),
|
|
("Linux", "aarch64", "",
|
|
f"bws-aarch64-unknown-linux-gnu-{bw._BWS_VERSION}.zip"),
|
|
("Windows", "AMD64", "",
|
|
f"bws-x86_64-pc-windows-msvc-{bw._BWS_VERSION}.zip"),
|
|
("Windows", "ARM64", "",
|
|
f"bws-aarch64-pc-windows-msvc-{bw._BWS_VERSION}.zip"),
|
|
],
|
|
)
|
|
def test_platform_asset_name(system, machine, libc_text, expected):
|
|
with mock.patch.object(bw.platform, "system", return_value=system), \
|
|
mock.patch.object(bw.platform, "machine", return_value=machine), \
|
|
mock.patch.object(
|
|
bw.subprocess,
|
|
"run",
|
|
return_value=mock.Mock(stdout=libc_text, stderr=libc_text),
|
|
):
|
|
assert bw._platform_asset_name() == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# install_bws — fully mocked HTTP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_fake_zip(binary_bytes: bytes) -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr("bws", binary_bytes)
|
|
return buf.getvalue()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _safe_extract_member — zip-slip containment
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_safe_extract_member_extracts_normal_member(tmp_path):
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr("bws", b"hello")
|
|
buf.seek(0)
|
|
|
|
dest = tmp_path / "extract"
|
|
dest.mkdir()
|
|
with zipfile.ZipFile(buf) as zf:
|
|
out = bw._safe_extract_member(zf, "bws", dest)
|
|
|
|
assert out == (dest / "bws").resolve() or out == dest / "bws"
|
|
assert Path(out).read_bytes() == b"hello"
|
|
# Nothing escaped the destination directory.
|
|
assert Path(out).resolve().parent == dest.resolve()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"evil_name",
|
|
[
|
|
"../escape",
|
|
"../../escape",
|
|
"sub/../../escape",
|
|
],
|
|
)
|
|
def test_safe_extract_member_rejects_traversal(tmp_path, evil_name):
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr(evil_name, b"pwned")
|
|
buf.seek(0)
|
|
|
|
dest = tmp_path / "extract"
|
|
dest.mkdir()
|
|
outside = tmp_path / "escape"
|
|
|
|
with zipfile.ZipFile(buf) as zf:
|
|
with pytest.raises(RuntimeError, match="unsafe archive member"):
|
|
bw._safe_extract_member(zf, evil_name, dest)
|
|
|
|
# The traversal target must not have been written.
|
|
assert not outside.exists()
|
|
|
|
|
|
def test_safe_extract_member_rejects_absolute_path(tmp_path):
|
|
# An absolute member name should never resolve inside dest.
|
|
abs_member = "/etc/cron.d/evil" if os.name != "nt" else "C:/Windows/evil"
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr(abs_member, b"pwned")
|
|
buf.seek(0)
|
|
|
|
dest = tmp_path / "extract"
|
|
dest.mkdir()
|
|
with zipfile.ZipFile(buf) as zf:
|
|
# Absolute paths are reduced to a relative member by zipfile, but
|
|
# we exercise the guard directly with the raw escaping name too.
|
|
with pytest.raises(RuntimeError, match="unsafe archive member"):
|
|
bw._safe_extract_member(zf, "../../../etc/cron.d/evil", dest)
|
|
|
|
|
|
def test_install_bws_rejects_malicious_member(hermes_home, monkeypatch):
|
|
# Build an archive whose only matching member escapes the temp dir.
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr(f"../../{bw._platform_binary_name()}", b"pwned")
|
|
zip_bytes = buf.getvalue()
|
|
asset_name = bw._platform_asset_name()
|
|
checksum_text = f"{hashlib.sha256(zip_bytes).hexdigest()} {asset_name}\n"
|
|
|
|
def fake_download(url, dest):
|
|
if url.endswith(".zip"):
|
|
Path(dest).write_bytes(zip_bytes)
|
|
elif url.endswith(".txt"):
|
|
Path(dest).write_text(checksum_text)
|
|
else:
|
|
raise AssertionError(f"unexpected download url: {url}")
|
|
|
|
monkeypatch.setattr(bw, "_http_download", fake_download)
|
|
|
|
with pytest.raises(RuntimeError, match="unsafe archive member"):
|
|
bw.install_bws()
|
|
|
|
|
|
def test_install_bws_happy_path(hermes_home, monkeypatch):
|
|
fake_binary = b"#!/bin/sh\necho 'bws fake 2.0.0'\n"
|
|
zip_bytes = _make_fake_zip(fake_binary)
|
|
asset_name = bw._platform_asset_name()
|
|
checksum_text = (
|
|
f"{hashlib.sha256(zip_bytes).hexdigest()} {asset_name}\n"
|
|
"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff other-file\n"
|
|
)
|
|
|
|
def fake_download(url, dest):
|
|
if url.endswith(".zip"):
|
|
Path(dest).write_bytes(zip_bytes)
|
|
elif url.endswith(".txt"):
|
|
Path(dest).write_text(checksum_text)
|
|
else:
|
|
raise AssertionError(f"unexpected download url: {url}")
|
|
|
|
monkeypatch.setattr(bw, "_http_download", fake_download)
|
|
|
|
path = bw.install_bws()
|
|
assert path.exists()
|
|
assert path.read_bytes() == fake_binary
|
|
# Executable bit set
|
|
assert path.stat().st_mode & stat.S_IXUSR
|
|
|
|
|
|
def test_install_bws_checksum_mismatch(hermes_home, monkeypatch):
|
|
zip_bytes = _make_fake_zip(b"contents")
|
|
asset_name = bw._platform_asset_name()
|
|
wrong_checksum = "0" * 64
|
|
checksum_text = f"{wrong_checksum} {asset_name}\n"
|
|
|
|
def fake_download(url, dest):
|
|
if url.endswith(".zip"):
|
|
Path(dest).write_bytes(zip_bytes)
|
|
else:
|
|
Path(dest).write_text(checksum_text)
|
|
|
|
monkeypatch.setattr(bw, "_http_download", fake_download)
|
|
|
|
with pytest.raises(RuntimeError, match="Checksum mismatch"):
|
|
bw.install_bws()
|
|
|
|
|
|
def test_install_bws_missing_checksum_entry(hermes_home, monkeypatch):
|
|
zip_bytes = _make_fake_zip(b"x")
|
|
|
|
def fake_download(url, dest):
|
|
if url.endswith(".zip"):
|
|
Path(dest).write_bytes(zip_bytes)
|
|
else:
|
|
Path(dest).write_text("ffffffff some-other-file.zip\n")
|
|
|
|
monkeypatch.setattr(bw, "_http_download", fake_download)
|
|
|
|
with pytest.raises(RuntimeError, match="No checksum entry"):
|
|
bw.install_bws()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fetch_bitwarden_secrets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _fake_bws_payload(items):
|
|
return json.dumps(items)
|
|
|
|
|
|
def test_fetch_happy_path(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([
|
|
{"key": "OPENAI_API_KEY", "value": "sk-abc"},
|
|
{"key": "ANTHROPIC_API_KEY", "value": "sk-ant-xyz"},
|
|
])
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
assert cmd[0] == str(fake_binary)
|
|
assert "secret" in cmd and "list" in cmd
|
|
assert kwargs["env"]["BWS_ACCESS_TOKEN"] == "0.fake.token"
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
secrets, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.fake.token",
|
|
project_id="proj-uuid",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
assert secrets == {
|
|
"OPENAI_API_KEY": "sk-abc",
|
|
"ANTHROPIC_API_KEY": "sk-ant-xyz",
|
|
}
|
|
assert warnings == []
|
|
|
|
|
|
def test_fetch_skips_invalid_env_names(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([
|
|
{"key": "VALID_KEY", "value": "v1"},
|
|
{"key": "1BAD_START", "value": "v2"},
|
|
{"key": "has spaces", "value": "v3"},
|
|
{"key": "DASH-KEY", "value": "v4"},
|
|
])
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess,
|
|
"run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
|
|
secrets, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
assert secrets == {"VALID_KEY": "v1"}
|
|
assert len(warnings) == 3
|
|
|
|
|
|
def test_fetch_auth_failure(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess,
|
|
"run",
|
|
lambda *a, **kw: mock.Mock(
|
|
returncode=1, stdout="", stderr="Error: invalid access token"
|
|
),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="invalid access token"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.bad",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
|
|
|
|
def test_fetch_timeout(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
|
|
def fake_run(*a, **kw):
|
|
raise subprocess.TimeoutExpired(cmd="bws", timeout=30)
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
with pytest.raises(RuntimeError, match="timed out"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
|
|
|
|
def test_fetch_non_json(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess,
|
|
"run",
|
|
lambda *a, **kw: mock.Mock(
|
|
returncode=0, stdout="not json at all", stderr=""
|
|
),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="non-JSON"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
|
|
|
|
def test_fetch_cache_hits(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K", "value": "v"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
bw.fetch_bitwarden_secrets(access_token="0.t", project_id="p",
|
|
binary=fake_binary, cache_ttl_seconds=60)
|
|
bw.fetch_bitwarden_secrets(access_token="0.t", project_id="p",
|
|
binary=fake_binary, cache_ttl_seconds=60)
|
|
assert call_count["n"] == 1 # cached on second call
|
|
|
|
|
|
def test_fetch_server_url_sets_env(monkeypatch, tmp_path):
|
|
"""server_url must be plumbed into the subprocess as BWS_SERVER_URL."""
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K", "value": "v"}])
|
|
|
|
captured_env = {}
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
captured_env.update(kwargs["env"])
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
server_url="https://vault.bitwarden.eu",
|
|
)
|
|
assert captured_env.get("BWS_SERVER_URL") == "https://vault.bitwarden.eu"
|
|
|
|
|
|
def test_fetch_no_server_url_does_not_set_env(monkeypatch, tmp_path):
|
|
"""When server_url is empty, BWS_SERVER_URL must not be injected."""
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([])
|
|
# Make sure the inherited env doesn't already have BWS_SERVER_URL set.
|
|
monkeypatch.delenv("BWS_SERVER_URL", raising=False)
|
|
|
|
captured_env = {}
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
captured_env.update(kwargs["env"])
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="p",
|
|
binary=fake_binary,
|
|
use_cache=False,
|
|
)
|
|
assert "BWS_SERVER_URL" not in captured_env
|
|
|
|
|
|
def test_fetch_server_url_keyed_in_cache(monkeypatch, tmp_path):
|
|
"""Different server_url values must produce separate cache entries."""
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K", "value": "v"}])
|
|
|
|
call_count = {"n": 0}
|
|
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
# US (default empty) — fresh fetch.
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="p",
|
|
binary=fake_binary, cache_ttl_seconds=60,
|
|
)
|
|
# EU — different server_url, must NOT hit the US cache entry.
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="p",
|
|
binary=fake_binary, cache_ttl_seconds=60,
|
|
server_url="https://vault.bitwarden.eu",
|
|
)
|
|
# Second EU call hits cache.
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="p",
|
|
binary=fake_binary, cache_ttl_seconds=60,
|
|
server_url="https://vault.bitwarden.eu",
|
|
)
|
|
assert call_count["n"] == 2
|
|
|
|
|
|
def test_fetch_cache_disabled(monkeypatch, tmp_path):
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([])
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
|
|
bw.fetch_bitwarden_secrets(access_token="0.t", project_id="p",
|
|
binary=fake_binary, use_cache=False)
|
|
bw.fetch_bitwarden_secrets(access_token="0.t", project_id="p",
|
|
binary=fake_binary, use_cache=False)
|
|
assert call_count["n"] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_bitwarden_secrets — the public entry point used by env_loader
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_apply_disabled_returns_empty():
|
|
result = bw.apply_bitwarden_secrets(enabled=False, project_id="p")
|
|
assert result.ok
|
|
assert not result.applied
|
|
assert not result.error
|
|
|
|
|
|
def test_apply_missing_token(monkeypatch):
|
|
monkeypatch.delenv("BWS_ACCESS_TOKEN", raising=False)
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="p", auto_install=False
|
|
)
|
|
assert not result.ok
|
|
assert "BWS_ACCESS_TOKEN" in result.error
|
|
|
|
|
|
def test_apply_missing_project_id(monkeypatch):
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.t")
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="", auto_install=False
|
|
)
|
|
assert not result.ok
|
|
assert "project_id" in result.error
|
|
|
|
|
|
def test_apply_does_not_override_existing(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.t")
|
|
monkeypatch.setenv("OPENAI_API_KEY", "existing-value")
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([
|
|
{"key": "OPENAI_API_KEY", "value": "bsm-value"},
|
|
{"key": "NEW_KEY", "value": "new-value"},
|
|
])
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
monkeypatch.setattr(bw, "find_bws", lambda **kw: fake_binary)
|
|
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="p",
|
|
override_existing=False, auto_install=False,
|
|
)
|
|
assert result.ok
|
|
assert "NEW_KEY" in result.applied
|
|
assert "OPENAI_API_KEY" in result.skipped
|
|
assert os.environ["OPENAI_API_KEY"] == "existing-value"
|
|
assert os.environ["NEW_KEY"] == "new-value"
|
|
|
|
|
|
def test_apply_override_existing(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.t")
|
|
monkeypatch.setenv("OPENAI_API_KEY", "stale")
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "OPENAI_API_KEY", "value": "fresh"}])
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
monkeypatch.setattr(bw, "find_bws", lambda **kw: fake_binary)
|
|
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="p",
|
|
override_existing=True, auto_install=False,
|
|
)
|
|
assert result.ok
|
|
assert os.environ["OPENAI_API_KEY"] == "fresh"
|
|
|
|
|
|
def test_apply_never_overrides_bootstrap_token(monkeypatch, tmp_path):
|
|
"""Even with override_existing=True, the access-token var is preserved."""
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.original")
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([
|
|
{"key": "BWS_ACCESS_TOKEN", "value": "0.malicious-replacement"},
|
|
])
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
monkeypatch.setattr(bw, "find_bws", lambda **kw: fake_binary)
|
|
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="p",
|
|
override_existing=True, auto_install=False,
|
|
)
|
|
assert os.environ["BWS_ACCESS_TOKEN"] == "0.original"
|
|
assert "BWS_ACCESS_TOKEN" in result.skipped
|
|
|
|
|
|
def test_apply_swallows_fetch_errors(monkeypatch, tmp_path):
|
|
"""A fetch failure produces an error, NOT an exception."""
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.t")
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="", stderr="bad token"),
|
|
)
|
|
monkeypatch.setattr(bw, "find_bws", lambda **kw: fake_binary)
|
|
|
|
result = bw.apply_bitwarden_secrets(
|
|
enabled=True, project_id="p", auto_install=False,
|
|
)
|
|
assert not result.ok
|
|
assert "bad token" in result.error
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# env_loader integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_env_loader_skips_when_disabled(tmp_path, monkeypatch):
|
|
"""No config.yaml present → no BSM call, no crash."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
|
|
|
from hermes_cli.env_loader import _apply_external_secret_sources
|
|
# Should be a no-op (returns None).
|
|
assert _apply_external_secret_sources(home) is None
|
|
|
|
|
|
def test_env_loader_calls_bsm_when_enabled(tmp_path, monkeypatch):
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
(home / "config.yaml").write_text(
|
|
"secrets:\n"
|
|
" bitwarden:\n"
|
|
" enabled: true\n"
|
|
" project_id: 'proj-1'\n"
|
|
" access_token_env: 'BWS_ACCESS_TOKEN'\n"
|
|
" cache_ttl_seconds: 0\n"
|
|
" override_existing: false\n"
|
|
" auto_install: false\n"
|
|
)
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
monkeypatch.setenv("BWS_ACCESS_TOKEN", "0.t")
|
|
monkeypatch.delenv("MY_BSM_KEY", raising=False)
|
|
|
|
called = {"n": 0}
|
|
|
|
def fake_fetch(**kwargs):
|
|
called["n"] += 1
|
|
assert kwargs["project_id"] == "proj-1"
|
|
return {"MY_BSM_KEY": "from-bsm"}, []
|
|
|
|
monkeypatch.setattr(
|
|
"agent.secret_sources.bitwarden.find_bws",
|
|
lambda **_kw: Path("/fake/bws"),
|
|
)
|
|
monkeypatch.setattr(
|
|
"agent.secret_sources.bitwarden.fetch_bitwarden_secrets",
|
|
fake_fetch,
|
|
)
|
|
from agent.secret_sources import registry as reg_module
|
|
|
|
reg_module._reset_registry_for_tests()
|
|
|
|
from hermes_cli.env_loader import _apply_external_secret_sources
|
|
_apply_external_secret_sources(home)
|
|
|
|
assert called["n"] == 1
|
|
assert os.environ.get("MY_BSM_KEY") == "from-bsm"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Disk-persisted cache (cross-process — speeds up back-to-back CLI invocations)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_disk_cache_written_after_first_fetch(monkeypatch, tmp_path):
|
|
"""First fetch hits bws AND writes a 0600 file under hermes_home/cache/."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
secrets, _ = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert secrets == {"K1": "v1"}
|
|
assert call_count["n"] == 1
|
|
|
|
cache_path = bw._disk_cache_path(home)
|
|
assert cache_path.exists()
|
|
# Mode must be 0600 — disk cache contains plaintext secret values
|
|
mode = os.stat(cache_path).st_mode & 0o777
|
|
assert mode == 0o600, f"expected 0o600, got 0o{mode:o}"
|
|
|
|
# File contents: key (fingerprint not raw token), secrets dict, fetched_at
|
|
payload_disk = json.loads(cache_path.read_text())
|
|
assert set(payload_disk.keys()) == {"key", "secrets", "fetched_at"}
|
|
assert payload_disk["secrets"] == {"K1": "v1"}
|
|
# Critically, the raw access token must NOT appear anywhere in the file
|
|
assert "0.t" not in cache_path.read_text()
|
|
|
|
|
|
def test_disk_cache_short_circuits_bws_when_fresh(monkeypatch, tmp_path):
|
|
"""Second fetch (different process simulation) skips bws entirely."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# First call: hits bws, populates disk cache
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert call_count["n"] == 1
|
|
|
|
# Clear ONLY the in-process cache to simulate a fresh subprocess.
|
|
bw._CACHE.clear()
|
|
|
|
secrets2, _ = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert secrets2 == {"K1": "v1"}
|
|
# Critical: bws was NOT invoked the second time
|
|
assert call_count["n"] == 1
|
|
|
|
|
|
def test_disk_cache_expires_with_ttl(monkeypatch, tmp_path):
|
|
"""Stale disk cache (older than ttl) triggers a refetch."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# First call
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert call_count["n"] == 1
|
|
|
|
# Backdate the disk cache so the TTL window has passed
|
|
cache_path = bw._disk_cache_path(home)
|
|
payload_disk = json.loads(cache_path.read_text())
|
|
payload_disk["fetched_at"] = time.time() - 10_000
|
|
cache_path.write_text(json.dumps(payload_disk))
|
|
bw._CACHE.clear()
|
|
|
|
# Second call: stale disk → refetch
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert call_count["n"] == 2
|
|
|
|
|
|
def test_disk_cache_key_mismatch_triggers_refetch(monkeypatch, tmp_path):
|
|
"""Disk cache entry written by a different token/project is ignored."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# Write a cache entry for a DIFFERENT token/project pair
|
|
cache_path = bw._disk_cache_path(home)
|
|
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cache_path.write_text(json.dumps({
|
|
"key": "deadbeef00000000|other-project|",
|
|
"secrets": {"OTHER": "should-not-leak"},
|
|
"fetched_at": time.time(),
|
|
}))
|
|
|
|
secrets, _ = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
# We must NOT have used the foreign cache entry
|
|
assert secrets == {"K1": "v1"}
|
|
assert "OTHER" not in secrets
|
|
assert call_count["n"] == 1
|
|
|
|
|
|
def test_disk_cache_use_cache_false_skips_disk(monkeypatch, tmp_path):
|
|
"""use_cache=False must skip BOTH in-process and disk caches."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
call_count = {"n": 0}
|
|
def fake_run(*a, **kw):
|
|
call_count["n"] += 1
|
|
return mock.Mock(returncode=0, stdout=payload, stderr="")
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# First call WITH cache populates disk
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, use_cache=True, home_path=home,
|
|
)
|
|
assert call_count["n"] == 1
|
|
bw._CACHE.clear()
|
|
|
|
# Second call with use_cache=False MUST hit bws again even though disk is fresh
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, use_cache=False, home_path=home,
|
|
)
|
|
assert call_count["n"] == 2
|
|
|
|
|
|
def test_disk_cache_corrupt_file_falls_through(monkeypatch, tmp_path):
|
|
"""A garbage cache file must NOT crash startup — we refetch."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "v1"}])
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# Write a corrupt cache file
|
|
cache_path = bw._disk_cache_path(home)
|
|
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cache_path.write_text("not json {{{")
|
|
|
|
secrets, _ = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
# Refetched cleanly
|
|
assert secrets == {"K1": "v1"}
|
|
# And the corrupt file was replaced with a valid one
|
|
assert json.loads(cache_path.read_text())["secrets"] == {"K1": "v1"}
|
|
|
|
|
|
def test_encrypted_cache_writes_without_plaintext(monkeypatch, tmp_path):
|
|
"""Encrypted cache stores last-good secrets without raw values on disk."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
payload = _fake_bws_payload([{"key": "K1", "value": "secret-value"}])
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess,
|
|
"run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout=payload, stderr=""),
|
|
)
|
|
bw._reset_cache_for_tests(home)
|
|
# A successful encrypted write must remove a pre-existing legacy plaintext
|
|
# cache from the migration path.
|
|
legacy_key = (bw._token_fingerprint("0.t"), "proj-1", "")
|
|
bw._DISK_CACHE.write(
|
|
legacy_key,
|
|
bw._CachedFetch(secrets={"K1": "legacy"}, fetched_at=time.time()),
|
|
300,
|
|
home,
|
|
)
|
|
assert bw._disk_cache_path(home).exists()
|
|
|
|
secrets, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=604800, home_path=home,
|
|
)
|
|
|
|
assert secrets == {"K1": "secret-value"}
|
|
assert warnings == []
|
|
assert not bw._disk_cache_path(home).exists()
|
|
cache_path = bw._encrypted_disk_cache_path(home)
|
|
assert cache_path.exists()
|
|
mode = stat.S_IMODE(os.stat(cache_path).st_mode)
|
|
assert mode == 0o600, f"expected 0o600, got 0o{mode:o}"
|
|
text = cache_path.read_text()
|
|
assert "secret-value" not in text
|
|
assert "0.t" not in text
|
|
payload_disk = json.loads(text)
|
|
assert set(payload_disk.keys()) == {
|
|
"version", "key", "salt", "nonce", "ciphertext",
|
|
}
|
|
assert not bw._disk_cache_path(home).exists()
|
|
|
|
|
|
def test_encrypted_cache_enabled_never_writes_plaintext_when_stale_disabled(
|
|
monkeypatch, tmp_path
|
|
):
|
|
"""Encryption remains mandatory even when stale fallback is disabled."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
monkeypatch.setattr(
|
|
bw.subprocess,
|
|
"run",
|
|
lambda *a, **kw: mock.Mock(
|
|
returncode=0,
|
|
stdout=_fake_bws_payload([{"key": "K1", "value": "secret-value"}]),
|
|
stderr="",
|
|
),
|
|
)
|
|
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="proj-1",
|
|
binary=fake_binary,
|
|
cache_ttl_seconds=300,
|
|
encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=0,
|
|
home_path=home,
|
|
)
|
|
|
|
assert bw._encrypted_disk_cache_path(home).exists()
|
|
assert not bw._disk_cache_path(home).exists()
|
|
|
|
|
|
def test_encrypted_cache_timestamp_is_authenticated(monkeypatch, tmp_path):
|
|
"""An unauthenticated outer timestamp cannot make old ciphertext usable."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
calls = {"n": 0}
|
|
|
|
def fake_run(*a, **kw):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
return mock.Mock(
|
|
returncode=0,
|
|
stdout=_fake_bws_payload([{"key": "K1", "value": "cached"}]),
|
|
stderr="",
|
|
)
|
|
return mock.Mock(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="Error: network is unreachable",
|
|
)
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="proj-1",
|
|
binary=fake_binary,
|
|
cache_ttl_seconds=0,
|
|
encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=300,
|
|
home_path=home,
|
|
)
|
|
|
|
cache_path = bw._encrypted_disk_cache_path(home)
|
|
payload = json.loads(cache_path.read_text())
|
|
cache_key = (bw._token_fingerprint("0.t"), "proj-1", "")
|
|
serialized_key = bw._cache_key_str(cache_key)
|
|
key = bw._derive_encrypted_cache_key("0.t", bw._b64d(payload["salt"]))
|
|
inner = json.loads(
|
|
bw.AESGCM(key).decrypt(
|
|
bw._b64d(payload["nonce"]),
|
|
bw._b64d(payload["ciphertext"]),
|
|
serialized_key.encode("utf-8"),
|
|
).decode("utf-8")
|
|
)
|
|
inner["fetched_at"] = time.time() - 10_000
|
|
nonce = os.urandom(12)
|
|
payload["nonce"] = bw._b64e(nonce)
|
|
payload["ciphertext"] = bw._b64e(
|
|
bw.AESGCM(key).encrypt(
|
|
nonce,
|
|
json.dumps(inner, separators=(",", ":")).encode("utf-8"),
|
|
serialized_key.encode("utf-8"),
|
|
)
|
|
)
|
|
# Simulate the old vulnerable format: a fresh, unauthenticated outer
|
|
# timestamp alongside stale encrypted content.
|
|
payload["fetched_at"] = time.time()
|
|
cache_path.write_text(json.dumps(payload))
|
|
bw._CACHE.clear()
|
|
|
|
with pytest.raises(RuntimeError, match="network is unreachable"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t",
|
|
project_id="proj-1",
|
|
binary=fake_binary,
|
|
cache_ttl_seconds=0,
|
|
encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=300,
|
|
home_path=home,
|
|
)
|
|
def test_encrypted_cache_falls_back_on_network_error(monkeypatch, tmp_path):
|
|
"""A fresh-enough encrypted cache is used when BWS is unreachable."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
calls = {"n": 0}
|
|
|
|
def fake_run(*a, **kw):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
return mock.Mock(
|
|
returncode=0,
|
|
stdout=_fake_bws_payload([{"key": "K1", "value": "cached"}]),
|
|
stderr="",
|
|
)
|
|
return mock.Mock(
|
|
returncode=1,
|
|
stdout="",
|
|
stderr="Error: network is unreachable",
|
|
)
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
first, _ = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=604800, home_path=home,
|
|
)
|
|
assert first == {"K1": "cached"}
|
|
bw._CACHE.clear()
|
|
|
|
second, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=604800, home_path=home,
|
|
)
|
|
assert second == {"K1": "cached"}
|
|
assert calls["n"] == 2
|
|
assert len(warnings) == 1
|
|
assert "stale ENCRYPTED disk cache" in warnings[0]
|
|
assert "bws live fetch failed" in warnings[0]
|
|
|
|
|
|
def test_encrypted_cache_does_not_fallback_on_auth_failure(monkeypatch, tmp_path):
|
|
"""Auth failures must not bypass revocation by using stale secrets."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
calls = {"n": 0}
|
|
|
|
def fake_run(*a, **kw):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
return mock.Mock(
|
|
returncode=0,
|
|
stdout=_fake_bws_payload([{"key": "K1", "value": "cached"}]),
|
|
stderr="",
|
|
)
|
|
return mock.Mock(returncode=1, stdout="", stderr="Error: invalid access token")
|
|
|
|
monkeypatch.setattr(bw.subprocess, "run", fake_run)
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=604800, home_path=home,
|
|
)
|
|
bw._CACHE.clear()
|
|
|
|
with pytest.raises(RuntimeError, match="invalid access token"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, encrypted_cache_enabled=True,
|
|
encrypted_cache_max_stale_seconds=604800, home_path=home,
|
|
)
|
|
|
|
|
|
def test_reset_cache_for_tests_deletes_disk_file(tmp_path):
|
|
"""_reset_cache_for_tests(home_path) must also clean disk."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
cache_path = bw._disk_cache_path(home)
|
|
encrypted_cache_path = bw._encrypted_disk_cache_path(home)
|
|
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cache_path.write_text("{}")
|
|
encrypted_cache_path.write_text("{}")
|
|
assert cache_path.exists()
|
|
assert encrypted_cache_path.exists()
|
|
|
|
bw._reset_cache_for_tests(home)
|
|
assert not cache_path.exists()
|
|
assert not encrypted_cache_path.exists()
|
|
# Idempotent
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stale disk cache fallback when live bws fetch fails
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _seed_stale_disk_cache(home, *, secrets, age_seconds, project_id="proj-1",
|
|
access_token="0.t", server_url=""):
|
|
"""Populate the disk cache as if a successful fetch happened `age_seconds`
|
|
ago. Writes the JSON payload directly (same shape the shared DiskCache
|
|
reads/writes) rather than going through DiskCache.write, since that
|
|
would honor cache_ttl_seconds and refuse to persist an already-"stale"
|
|
entry — this needs to land on disk regardless of TTL."""
|
|
cache_key = (
|
|
bw._token_fingerprint(access_token), project_id, server_url,
|
|
)
|
|
cache_path = bw._disk_cache_path(home)
|
|
cache_path.parent.mkdir(parents=True, exist_ok=True)
|
|
cache_path.write_text(json.dumps({
|
|
"key": bw._cache_key_str(cache_key),
|
|
"secrets": secrets,
|
|
"fetched_at": time.time() - age_seconds,
|
|
}))
|
|
|
|
|
|
def test_stale_disk_cache_returned_when_bws_fails(monkeypatch, tmp_path):
|
|
"""When bws fails and the disk cache is stale, return the stale secrets
|
|
with a warning rather than raising."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# Seed a stale (older than TTL) disk cache from a previous successful fetch
|
|
_seed_stale_disk_cache(home, secrets={"OPENAI_API_KEY": "sk-old"},
|
|
age_seconds=3600)
|
|
|
|
# Now simulate a BWS network failure
|
|
def fail_run(*a, **kw):
|
|
return mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: dns resolution failed")
|
|
monkeypatch.setattr(bw.subprocess, "run", fail_run)
|
|
|
|
secrets, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
assert secrets == {"OPENAI_API_KEY": "sk-old"}
|
|
assert len(warnings) == 1
|
|
assert "stale disk cache" in warnings[0]
|
|
assert "dns resolution failed" in warnings[0]
|
|
|
|
|
|
def test_stale_fallback_warning_includes_cache_age(monkeypatch, tmp_path):
|
|
"""Operator-facing warning should report how old the cached secrets are."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=7200)
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: connection refused"),
|
|
)
|
|
|
|
_, warnings = bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
# 7200s == 2h; we don't pin exact integer seconds because time.time()
|
|
# drifts a bit between seed and call, but it must be within a small band.
|
|
assert "7200s old" in warnings[0] or "7199s old" in warnings[0]
|
|
|
|
|
|
def test_no_stale_fallback_when_disk_cache_missing(monkeypatch, tmp_path):
|
|
"""If bws fails and there's no disk cache at all, re-raise — no silent
|
|
success with empty secrets."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: network unreachable"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="network unreachable"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
|
|
|
|
def test_stale_fallback_skipped_when_use_cache_false(monkeypatch, tmp_path):
|
|
"""`use_cache=False` is an explicit opt-out from any cached value,
|
|
including the stale fallback path — must still raise on bws failure."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
# A stale cache exists — but use_cache=False should ignore it
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=3600)
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: connection refused"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="connection refused"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home, use_cache=False,
|
|
)
|
|
|
|
|
|
def test_stale_fallback_does_not_overwrite_disk_cache(monkeypatch, tmp_path):
|
|
"""Stale fallback must not bump the disk cache `fetched_at` — that would
|
|
falsely make future processes think the cache is fresh."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=3600)
|
|
cache_path = bw._disk_cache_path(home)
|
|
original_fetched_at = json.loads(cache_path.read_text())["fetched_at"]
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: connection refused"),
|
|
)
|
|
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
|
|
# Disk cache should still carry the old fetched_at — the live fetch
|
|
# failed and produced no new secrets to persist.
|
|
assert json.loads(cache_path.read_text())["fetched_at"] == original_fetched_at
|
|
|
|
|
|
def test_stale_fallback_skipped_on_auth_failure(monkeypatch, tmp_path):
|
|
"""An AUTH_FAILED bws error must raise, not serve stale secrets — a bad
|
|
access token indicates a real credential problem the caller needs to
|
|
see, not a transient outage worth papering over."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=3600)
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: unauthorized (401)"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="unauthorized"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
|
|
|
|
def test_stale_fallback_skipped_on_malformed_output(monkeypatch, tmp_path):
|
|
"""An INTERNAL-classified failure (unparseable bws output) must raise —
|
|
the fallback is scoped to transport failures only, not "anything went
|
|
wrong"."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=3600)
|
|
|
|
# returncode == 0 but unparseable stdout raises a ValueError-wrapping
|
|
# RuntimeError from _run_bws_list's JSON parsing — classifies INTERNAL.
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=0, stdout="not json", stderr=""),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=300, home_path=home,
|
|
)
|
|
|
|
|
|
def test_stale_fallback_skipped_when_cache_ttl_zero(monkeypatch, tmp_path):
|
|
"""cache_ttl_seconds=0 means the caller opted out of caching entirely —
|
|
the stale fallback must honor that even though it explicitly asks the
|
|
disk cache for a stale (not-fresh) hit via ttl_seconds=inf internally."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
fake_binary = tmp_path / "bws"
|
|
fake_binary.write_text("")
|
|
bw._reset_cache_for_tests(home)
|
|
|
|
_seed_stale_disk_cache(home, secrets={"K1": "v1"}, age_seconds=3600)
|
|
|
|
monkeypatch.setattr(
|
|
bw.subprocess, "run",
|
|
lambda *a, **kw: mock.Mock(returncode=1, stdout="",
|
|
stderr="Error: connection refused"),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="connection refused"):
|
|
bw.fetch_bitwarden_secrets(
|
|
access_token="0.t", project_id="proj-1", binary=fake_binary,
|
|
cache_ttl_seconds=0, home_path=home,
|
|
)
|