fix(secrets): add encrypted Bitwarden stale cache

agent/secret_sources/bitwarden.py

@@ -29,6 +29,7 @@ is easier to lazy-install than a wheels-with-Rust-extension dependency.
 
 from __future__ import annotations
 
+import base64
 import hashlib
 import json
 import logging
@@ -46,6 +47,10 @@ import zipfile
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
 from agent.secret_sources._cache import (
     CachedFetch as _CachedFetch,
     DiskCache,
@@ -92,6 +97,9 @@ _CACHE: Dict[_CacheKey, _CachedFetch] = {}
 # accidentally commit BSM-sourced secrets. The atomic-write/0600/TTL mechanics
 # live in agent.secret_sources._cache.DiskCache, shared with the other backends.
 _DISK_CACHE_BASENAME = "bws_cache.json"
+_ENCRYPTED_CACHE_BASENAME = "bws_cache.enc.json"
+_ENCRYPTED_CACHE_VERSION = 1
+_ENCRYPTED_CACHE_INFO = b"hermes-bws-encrypted-cache-v1"
 
 
 def _cache_key_str(cache_key: _CacheKey) -> str:
@@ -114,6 +122,13 @@ def _disk_cache_path(home_path: Optional[Path] = None) -> Path:
     return _DISK_CACHE.path(home_path)
 
 
+def _encrypted_disk_cache_path(home_path: Optional[Path] = None) -> Path:
+    """Return the encrypted disk cache path under hermes_home/cache/."""
+    from agent.secret_sources._cache import resolve_cache_home
+
+    return resolve_cache_home(home_path) / "cache" / _ENCRYPTED_CACHE_BASENAME
+
+
 # ---------------------------------------------------------------------------
 # Binary discovery + lazy install
 # ---------------------------------------------------------------------------
@@ -349,6 +364,130 @@ def _token_fingerprint(token: str) -> str:
     return hashlib.sha256(token.encode("utf-8")).hexdigest()[:16]
 
 
+def _b64e(raw: bytes) -> str:
+    return base64.b64encode(raw).decode("ascii")
+
+
+def _b64d(text: str) -> bytes:
+    return base64.b64decode(text.encode("ascii"), validate=True)
+
+
+def _derive_encrypted_cache_key(access_token: str, salt: bytes) -> bytes:
+    """Derive the local cache encryption key from the bootstrap BWS token."""
+    return HKDF(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=salt,
+        info=_ENCRYPTED_CACHE_INFO,
+    ).derive(access_token.encode("utf-8"))
+
+
+def _write_encrypted_disk_cache(
+    *,
+    cache_key: _CacheKey,
+    access_token: str,
+    entry: _CachedFetch,
+    home_path: Optional[Path] = None,
+) -> None:
+    """Persist an encrypted last-good cache entry atomically.
+
+    Best-effort by design: cache write failure must never block a fresh BWS
+    fetch.  The raw BWS access token is not stored; it only derives the AES key.
+    """
+    path = _encrypted_disk_cache_path(home_path)
+    try:
+        cache_dir = path.parent
+        cache_dir.mkdir(parents=True, exist_ok=True)
+        try:
+            os.chmod(cache_dir, 0o700)
+        except OSError:
+            pass
+        salt = os.urandom(16)
+        nonce = os.urandom(12)
+        serialized_key = _cache_key_str(cache_key)
+        key = _derive_encrypted_cache_key(access_token, salt)
+        plaintext = json.dumps(
+            {"secrets": entry.secrets, "fetched_at": entry.fetched_at},
+            separators=(",", ":"),
+        ).encode("utf-8")
+        ciphertext = AESGCM(key).encrypt(
+            nonce, plaintext, serialized_key.encode("utf-8")
+        )
+        payload = {
+            "version": _ENCRYPTED_CACHE_VERSION,
+            "key": serialized_key,
+            "fetched_at": entry.fetched_at,
+            "salt": _b64e(salt),
+            "nonce": _b64e(nonce),
+            "ciphertext": _b64e(ciphertext),
+        }
+        fd, tmp = tempfile.mkstemp(
+            prefix=".bws_cache_enc_", suffix=".tmp", dir=str(cache_dir)
+        )
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                json.dump(payload, f)
+            os.chmod(tmp, 0o600)
+            os.replace(tmp, path)
+        except BaseException:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+            raise
+    except Exception:  # noqa: BLE001 — best-effort cache only
+        return
+
+
+def _read_encrypted_disk_cache(
+    *,
+    cache_key: _CacheKey,
+    access_token: str,
+    max_age_seconds: float,
+    home_path: Optional[Path] = None,
+) -> Optional[_CachedFetch]:
+    """Return a decrypted encrypted-cache entry if it matches and is in-window."""
+    if max_age_seconds <= 0:
+        return None
+    path = _encrypted_disk_cache_path(home_path)
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        if not isinstance(payload, dict):
+            return None
+        serialized_key = _cache_key_str(cache_key)
+        if payload.get("version") != _ENCRYPTED_CACHE_VERSION:
+            return None
+        if payload.get("key") != serialized_key:
+            return None
+        fetched_at = payload.get("fetched_at")
+        if not isinstance(fetched_at, (int, float)):
+            return None
+        entry_age = time.time() - float(fetched_at)
+        if entry_age < 0 or entry_age > max_age_seconds:
+            return None
+        salt = _b64d(str(payload.get("salt", "")))
+        nonce = _b64d(str(payload.get("nonce", "")))
+        ciphertext = _b64d(str(payload.get("ciphertext", "")))
+        key = _derive_encrypted_cache_key(access_token, salt)
+        raw = AESGCM(key).decrypt(
+            nonce, ciphertext, serialized_key.encode("utf-8")
+        )
+        inner = json.loads(raw.decode("utf-8"))
+        if not isinstance(inner, dict):
+            return None
+        secrets = inner.get("secrets")
+        inner_fetched_at = inner.get("fetched_at")
+        if not isinstance(secrets, dict) or not isinstance(inner_fetched_at, (int, float)):
+            return None
+        typed = {
+            k: v for k, v in secrets.items()
+            if isinstance(k, str) and isinstance(v, str)
+        }
+        return _CachedFetch(secrets=typed, fetched_at=float(inner_fetched_at))
+    except Exception:  # noqa: BLE001 — cache miss on parse/decrypt/I/O errors
+        return None
+
+
 def fetch_bitwarden_secrets(
     *,
     access_token: str,
@@ -358,6 +497,8 @@ def fetch_bitwarden_secrets(
     use_cache: bool = True,
     server_url: str = "",
     home_path: Optional[Path] = None,
+    encrypted_cache_enabled: bool = False,
+    encrypted_cache_max_stale_seconds: float = 0,
 ) -> Tuple[Dict[str, str], List[str]]:
     """Pull the secrets for ``project_id`` from Bitwarden Secrets Manager.
 
@@ -369,12 +510,13 @@ def fetch_bitwarden_secrets(
     (``https://vault.bitwarden.com``, US Cloud).  This is plumbed into
     the subprocess as ``BWS_SERVER_URL``.
 
-    Caching is a two-layer LRU: an in-process dict (for hot-reload paths
-    inside one process) and a disk-persisted JSON file under
-    ``<hermes_home>/cache/bws_cache.json`` (for back-to-back CLI invocations).
-    Both share the same TTL.  Pass ``home_path`` so disk cache lookups find
-    the right directory in tests / non-standard installs; otherwise we fall
-    back to ``$HERMES_HOME`` / ``~/.hermes``.
+    ``cache_ttl_seconds`` controls the normal fresh cache.  When
+    ``encrypted_cache_enabled`` is true, fresh cache entries are written as
+    AES-GCM encrypted JSON instead of plaintext, and a last-good encrypted
+    entry may be used after NETWORK/TIMEOUT failures for up to
+    ``encrypted_cache_max_stale_seconds``.  This stale fallback is separate
+    from the fresh-cache TTL so operators can set ``cache_ttl_seconds: 0``
+    while still keeping an encrypted break-glass cache for offline startup.
 
     Raises :class:`RuntimeError` for fatal conditions (missing binary,
     auth failure, unparseable output).  Callers in the env_loader path
@@ -387,12 +529,20 @@ def fetch_bitwarden_secrets(
         raise RuntimeError("Bitwarden project_id is empty")
 
     cache_key = (_token_fingerprint(access_token), project_id, server_url or "")
-    if use_cache:
+    if use_cache and cache_ttl_seconds > 0:
         cached = _CACHE.get(cache_key)
         if cached and cached.is_fresh(cache_ttl_seconds):
             return cached.secrets, []
         # L2: disk cache. ~5ms on cache hit vs ~380ms for `bws secret list`.
-        disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path)
+        if encrypted_cache_enabled:
+            disk_cached = _read_encrypted_disk_cache(
+                cache_key=cache_key,
+                access_token=access_token,
+                max_age_seconds=cache_ttl_seconds,
+                home_path=home_path,
+            )
+        else:
+            disk_cached = _DISK_CACHE.read(cache_key, cache_ttl_seconds, home_path)
         if disk_cached is not None:
             # Promote into in-process cache so subsequent fetches in the
             # same process skip the disk read too.
@@ -419,31 +569,57 @@ def fetch_bitwarden_secrets(
         # fallback a fleet of bots sharing one BWS project all stop working
         # on a single network blip.
         #
-        # `cache_ttl_seconds <= 0` means the caller opted out of caching
-        # entirely (DiskCache.read/write both short-circuit on it) — honor
-        # that on the fallback path too, so a zero-TTL caller never gets a
-        # secret value that didn't come from a live fetch just now.
-        # `ttl_seconds=inf` on the read call itself bypasses freshness (we
-        # explicitly want a stale hit here); the caller's real TTL is what
-        # gates whether we even attempt the read, via the check above.
-        if (
-            use_cache
-            and cache_ttl_seconds > 0
-            and _classify_bws_error(str(exc)) in (ErrorKind.NETWORK, ErrorKind.TIMEOUT)
-        ):
-            stale = _DISK_CACHE.read(cache_key, float("inf"), home_path)
-            if stale is not None:
-                age = max(0.0, time.time() - stale.fetched_at)
-                _CACHE[cache_key] = stale
-                return stale.secrets, [
-                    f"bws live fetch failed ({exc}); "
-                    f"falling back to stale disk cache ({int(age)}s old)"
-                ]
+        # Two fallback tiers share the transport-only gate:
+        # * encrypted cache (opt-in) — AES-GCM payload keyed off the
+        #   bootstrap token, with its own max_stale_seconds window.  When
+        #   enabled it is the ONLY fallback consulted: the whole point is
+        #   that the at-rest payload is never plaintext, so we don't
+        #   quietly serve the plaintext file alongside it.
+        # * plaintext disk cache (default) — the ordinary DiskCache file.
+        #   `cache_ttl_seconds <= 0` means the caller opted out of caching
+        #   entirely (DiskCache.read/write both short-circuit on it) —
+        #   honor that on the fallback path too.  `ttl_seconds=inf` on the
+        #   read bypasses freshness (we explicitly want a stale hit); the
+        #   caller's real TTL gates whether we even attempt the read.
+        kind = _classify_bws_error(str(exc))
+        if use_cache and kind in (ErrorKind.NETWORK, ErrorKind.TIMEOUT):
+            if encrypted_cache_enabled:
+                stale = _read_encrypted_disk_cache(
+                    cache_key=cache_key,
+                    access_token=access_token,
+                    max_age_seconds=encrypted_cache_max_stale_seconds,
+                    home_path=home_path,
+                )
+                if stale is not None:
+                    age = max(0.0, time.time() - stale.fetched_at)
+                    _CACHE[cache_key] = stale
+                    return stale.secrets, [
+                        f"bws live fetch failed ({exc}); falling back to "
+                        f"stale ENCRYPTED disk cache ({int(age)}s old)"
+                    ]
+            elif cache_ttl_seconds > 0:
+                stale = _DISK_CACHE.read(cache_key, float("inf"), home_path)
+                if stale is not None:
+                    age = max(0.0, time.time() - stale.fetched_at)
+                    _CACHE[cache_key] = stale
+                    return stale.secrets, [
+                        f"bws live fetch failed ({exc}); "
+                        f"falling back to stale disk cache ({int(age)}s old)"
+                    ]
         raise
     entry = _CachedFetch(secrets=secrets, fetched_at=time.time())
-    _CACHE[cache_key] = entry
     if use_cache:
-        _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path)
+        if cache_ttl_seconds > 0:
+            _CACHE[cache_key] = entry
+        if encrypted_cache_enabled and encrypted_cache_max_stale_seconds > 0:
+            _write_encrypted_disk_cache(
+                cache_key=cache_key,
+                access_token=access_token,
+                entry=entry,
+                home_path=home_path,
+            )
+        elif cache_ttl_seconds > 0:
+            _DISK_CACHE.write(cache_key, entry, cache_ttl_seconds, home_path)
     return secrets, warnings
 
 
@@ -569,6 +745,8 @@ def apply_bitwarden_secrets(
     auto_install: bool = True,
     server_url: str = "",
     home_path: Optional[Path] = None,
+    encrypted_cache_enabled: bool = False,
+    encrypted_cache_max_stale_seconds: float = 0,
 ) -> FetchResult:
     """Pull secrets from BSM and set them on ``os.environ``.
 
@@ -620,6 +798,8 @@ def apply_bitwarden_secrets(
             cache_ttl_seconds=cache_ttl_seconds,
             server_url=server_url,
             home_path=home_path,
+            encrypted_cache_enabled=encrypted_cache_enabled,
+            encrypted_cache_max_stale_seconds=encrypted_cache_max_stale_seconds,
         )
     except RuntimeError as exc:
         result.error = str(exc)
@@ -689,9 +869,16 @@ class BitwardenSource(SecretSource):
             },
             "project_id": {"description": "BSM project UUID", "default": ""},
             "cache_ttl_seconds": {
-                "description": "Disk+memory cache TTL; 0 disables",
+                "description": "Fresh disk+memory cache TTL; 0 disables fresh-cache reuse",
                 "default": 300,
             },
+            "encrypted_cache": {
+                "description": "Encrypted last-good cache for network/timeout fallback",
+                "default": {
+                    "enabled": False,
+                    "max_stale_seconds": 0,
+                },
+            },
             "override_existing": {
                 "description": "BSM values overwrite .env/shell values",
                 "default": True,
@@ -745,6 +932,14 @@ class BitwardenSource(SecretSource):
         except (TypeError, ValueError):
             ttl = 300.0
 
+        encrypted_cfg = cfg.get("encrypted_cache")
+        encrypted_cfg = encrypted_cfg if isinstance(encrypted_cfg, dict) else {}
+        encrypted_enabled = bool(encrypted_cfg.get("enabled", False))
+        try:
+            encrypted_max_stale = float(encrypted_cfg.get("max_stale_seconds", 0))
+        except (TypeError, ValueError):
+            encrypted_max_stale = 0.0
+
         try:
             secrets, warnings = fetch_bitwarden_secrets(
                 access_token=access_token,
@@ -753,6 +948,8 @@ class BitwardenSource(SecretSource):
                 cache_ttl_seconds=ttl,
                 server_url=str(cfg.get("server_url", "") or "").strip(),
                 home_path=home_path,
+                encrypted_cache_enabled=encrypted_enabled,
+                encrypted_cache_max_stale_seconds=encrypted_max_stale,
             )
         except RuntimeError as exc:
             result.error = str(exc)
@@ -810,14 +1007,19 @@ def _classify_bws_error(message: str) -> ErrorKind:
 
 
 def clear_caches(home_path: Optional[Path] = None) -> None:
-    """Drop in-process AND disk caches.
+    """Drop in-process AND disk caches (plaintext and encrypted).
 
     Used after a token rotation (`hermes secrets bitwarden token`) so the
     next startup fetches fresh with the new credential instead of serving
-    a pull cached under the old token's fingerprint.
+    a pull cached under the old token's fingerprint.  The encrypted cache
+    is keyed off the old token too, so it must go as well.
     """
     _CACHE.clear()
     _DISK_CACHE.clear(home_path)
+    try:
+        _encrypted_disk_cache_path(home_path).unlink()
+    except (FileNotFoundError, OSError):
+        pass
 
 
 def _reset_cache_for_tests(home_path: Optional[Path] = None) -> None:

hermes_cli/config.py

@@ -3374,8 +3374,18 @@ DEFAULT_CONFIG = {
             "access_token_env": "BWS_ACCESS_TOKEN",
             # UUID of the BSM project to sync from.
             "project_id": "",
-            # Seconds to cache fetched secrets in-process.  0 disables.
+            # Seconds to reuse a fresh disk/memory cache entry before contacting
+            # Bitwarden again. 0 disables normal fresh-cache reuse.
             "cache_ttl_seconds": 300,
+            # Optional encrypted last-good fallback for network/timeout outages.
+            # When enabled, successful BWS fetches write AES-GCM encrypted cache
+            # material under ~/.hermes/cache/. If a later startup cannot reach
+            # Bitwarden due to NETWORK/TIMEOUT, Hermes may use this encrypted
+            # cache for up to max_stale_seconds. Auth failures do not fall back.
+            "encrypted_cache": {
+                "enabled": False,
+                "max_stale_seconds": 0,
+            },
             # When True, BSM values overwrite existing env vars.  Default
             # True because the point of using BSM is centralized rotation —
             # if .env had the final say, rotating in Bitwarden wouldn't

tests/test_bitwarden_secrets.py

@@ -870,17 +870,139 @@ def test_disk_cache_corrupt_file_falls_through(monkeypatch, tmp_path):
     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)
+
+    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", "fetched_at", "salt", "nonce", "ciphertext",
+    }
+
+
+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 warnings == [
+        "Using stale encrypted Bitwarden cache after network fetching BWS secrets"
+    ]
+
+
+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)