"""BitwardenCliBackend tests -- run OFFLINE against a fake `bw` binary. The fake records, per invocation: argv, the *names* of secret-channel env vars it received (never values), and its own /proc/self/cmdline. Tests assert that API credentials, the master password and the session key never appear on argv/cmdline, that BW_SESSION lives only in child env, and that the broker socket surface stays metadata-only. """ from __future__ import annotations import json import os import subprocess import sys import tempfile import threading import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) sys.path.insert(0, str(Path(__file__).resolve().parent)) from fakebw import FAKE_BW from agent_credential_broker.config import Config from agent_credential_broker.core import BrokerCore from agent_credential_broker.errors import BrokerError from agent_credential_broker.server import _Server, call MASTER_PW = "fixture-master-password-DO-NOT-USE" PW = "example-password-DO-NOT-USE" TOKEN = "example-token-DO-NOT-USE" CLIENT_ID = "user.fixture-id" CLIENT_SECRET = "fixture-client-secret" ITEMS = [ {"id": "item-1", "type": 1, "name": "gitea admin", "login": {"username": "gitea-admin", "password": PW, "totp": None, "uris": [{"uri": "https://git2.paginica.com"}]}, "fields": [ {"name": "agent.ref", "value": "gitea.admin", "type": 0}, {"name": "agent.adapters", "value": "browser,http,git", "type": 0}, {"name": "agent.allowed_urls", "value": "https://git2.paginica.com", "type": 0}, {"name": "agent.auth_type", "value": "bearer", "type": 0}, {"name": "agent.confirmation", "value": "AUTO", "type": 0}, ], "notes": None}, {"id": "item-2", "type": 1, "name": "api token", "login": {"username": "", "password": TOKEN, "totp": None, "uris": [{"uri": "https://service.example"}]}, "fields": [ {"name": "agent.ref", "value": "test.service", "type": 0}, {"name": "agent.adapters", "value": "http,cli-profile", "type": 0}, {"name": "agent.allowed_urls", "value": "https://service.example", "type": 0}, {"name": "agent.auth_type", "value": "bearer", "type": 0}, {"name": "agent.confirmation", "value": "AUTO", "type": 0}, {"name": "agent.cli_profile", "value": "example-echo", "type": 0}, ], "notes": None}, {"id": "item-3", "type": 1, "name": "not for agents", "login": {"username": "x", "password": PW, "totp": None, "uris": []}, "fields": [], "notes": "private note"}, ] def _write(path: Path, text: str, mode=0o600): path.write_text(text) os.chmod(path, mode) def read_calls(appdata: Path) -> list[dict]: log = appdata / "calls.jsonl" if not log.exists(): return [] return [json.loads(l) for l in log.read_text().splitlines() if l.strip()] class CliBackendHarness(unittest.TestCase): """Daemon + socket + BitwardenCliBackend driving the fake `bw`.""" FAILING_BW = False @classmethod def setUpClass(cls): cls.tmp = tempfile.TemporaryDirectory() base = Path(cls.tmp.name) os.environ["XDG_RUNTIME_DIR"] = str(base / "rt") os.environ["XDG_STATE_HOME"] = str(base / "state") (base / "rt").mkdir() cred = base / "cred" cred.mkdir() _write(cred / "id.txt", CLIENT_ID) _write(cred / "secret.token", CLIENT_SECRET) bin_dir = base / "bin" bin_dir.mkdir() cls.bw = bin_dir / "bw" src = FAKE_BW.replace("die(\"unknown cmd", "sys.exit(1) # die(\"unknown cmd") \ if cls.FAILING_BW else FAKE_BW cls.bw.write_text(src) os.chmod(cls.bw, 0o755) cls.appdata = base / "bw-appdata" cls.appdata.mkdir(mode=0o700) (cls.appdata / "items.json").write_text(json.dumps(ITEMS)) cls.cfg_path = base / "broker-config.json" cls.cfg_path.write_text(json.dumps({ "vault_url": "https://vault.paginica.com", "client_id_file": str(cred / "id.txt"), "client_secret_file": str(cred / "secret.token"), "vault_backend": "bitwarden-cli", "bw_path": str(cls.bw), "bw_appdata_dir": str(cls.appdata), "unlock_ttl_seconds": 3600, "git_workdir_roots": [str(Path.home())], })) os.environ["AGENT_BROKER_CONFIG"] = str(cls.cfg_path) os.environ["AGENT_BROKER_SOCKET"] = str(base / "rt" / "agent-broker" / "broker.sock") os.environ["AGENT_BROKER_PROFILES_DIR"] = str(Path(__file__).resolve().parents[1] / "profiles") cls.socket_path = Path(os.environ["AGENT_BROKER_SOCKET"]) cls.socket_path.parent.mkdir(parents=True, exist_ok=True) cls.core = BrokerCore(Config.load()) cls.server = _Server(str(cls.socket_path), cls.core) os.chmod(cls.socket_path, 0o600) threading.Thread(target=cls.server.serve_forever, kwargs={"poll_interval": 0.1}, daemon=True).start() @classmethod def tearDownClass(cls): cls.server.shutdown() cls.tmp.cleanup() def call(self, method, params=None): return call(method, params or {}) class TestCliBackendLifecycle(CliBackendHarness): def test_locked_then_unlock_and_lock(self): st = self.call("status") self.assertTrue(st["locked"]) self.assertEqual(st["backend"], "bitwarden-cli") with self.assertRaises(BrokerError) as e: self.call("list_refs") self.assertEqual(e.exception.code, "BROKER_LOCKED") res = self.call("unlock", {"password": MASTER_PW}) self.assertTrue(res["unlocked"]) self.assertEqual(res["items"], len(ITEMS)) refs = {r["ref"]: r for r in self.call("list_refs")} self.assertEqual(set(refs), {"gitea.admin", "test.service"}) blob = json.dumps(refs) self.assertNotIn(PW, blob) self.assertNotIn(TOKEN, blob) self.call("lock") self.assertTrue(self.call("status")["locked"]) # bw lock ran and wiped the session server-side self.assertTrue(any(c["argv"][1] == "lock" for c in read_calls(self.appdata))) def test_wrong_password_unlock_failed(self): with self.assertRaises(BrokerError) as e: self.call("unlock", {"password": "not-the-password"}) self.assertEqual(e.exception.code, "UNLOCK_FAILED") def test_full_item_fetched_per_use(self): self.call("unlock", {"password": MASTER_PW}) calls_before = read_calls(self.appdata) self.assertFalse(any("item-1" in c["argv"] for c in calls_before)) cred = self.call("resolve", {"ref": "gitea.admin", "url": "https://git2.paginica.com/user/login"}) self.assertEqual(cred["password"], PW) calls = read_calls(self.appdata) get_calls = [c for c in calls if c["argv"][1:3] == ["get", "item"]] self.assertTrue(any("item-1" in c["argv"] for c in get_calls)) # only the requested item was fetched self.assertFalse(any("item-2" in c["argv"] for c in get_calls)) self.call("lock") def test_metadata_state_has_no_secrets(self): self.call("unlock", {"password": MASTER_PW}) blob = json.dumps(self.core.state.items) for secret in (PW, TOKEN, "private note", "gitea-admin"): self.assertNotIn(secret, blob) # only agent.* fields survive in metadata for meta in self.core.state.items: for k in meta["fields"]: self.assertTrue(k.startswith("agent.")) self.call("lock") class TestCliBackendIsolation(CliBackendHarness): """Secrets travel only through child env -- never argv, never the broker's own environment, never logs.""" @classmethod def setUpClass(cls): super().setUpClass() cls.call = staticmethod(call) call("unlock", {"password": MASTER_PW}) call("resolve", {"ref": "gitea.admin", "url": "https://git2.paginica.com/user/login"}) # capture the session key while live, then end the session st = json.loads((cls.appdata / "state.json").read_text()) cls.session = st.get("session") call("lock") def test_no_secret_on_broker_disk(self): """Broker-managed paths must never persist plaintext secrets. (bw_appdata_dir is CLI-managed state and excluded on purpose -- the official client's own data.json belongs to it.)""" base = Path(self.tmp.name) broker_paths = [base / "rt", base / "state", self.cfg_path] secrets = [s for s in (MASTER_PW, CLIENT_SECRET, PW, TOKEN, self.session) if s] for root in broker_paths: p = Path(root) files = [p] if p.is_file() else list(p.rglob("*")) for f in files: if f.is_file(): data = f.read_bytes() for s in secrets: self.assertNotIn(s.encode(), data, f"{s!r} persisted in {f}") def test_logout_via_socket(self): self.call("unlock", {"password": MASTER_PW}) res = self.call("logout") self.assertTrue(res["logged_out"]) calls = read_calls(self.appdata) self.assertTrue(any(c["argv"][1] == "logout" for c in calls)) st = self.call("status") self.assertTrue(st["locked"]) def test_no_secret_in_argv_or_cmdline(self): for c in read_calls(self.appdata): for secret in (CLIENT_SECRET, MASTER_PW, PW, TOKEN): self.assertNotIn(secret, " ".join(c["argv"])) self.assertNotIn(secret, c["cmdline"]) def test_session_key_only_in_child_env(self): st = json.loads((self.appdata / "state.json").read_text()) session = st.get("session") # gone after lock calls = read_calls(self.appdata) for c in calls: if c["argv"][1] == "login": self.assertIn("BW_CLIENTID", c["env_names"]) self.assertIn("BW_CLIENTSECRET", c["env_names"]) if c["argv"][1] == "unlock": self.assertNotIn("BW_SESSION", c["env_names"]) if c["argv"][1] in ("sync", "list", "get"): self.assertIn("BW_SESSION", c["env_names"]) # session never landed on argv if session: for c in calls: self.assertNotIn(session, " ".join(c["argv"])) self.assertNotIn(session, c["cmdline"]) self.assertNotIn("BW_SESSION", os.environ) def test_login_env_scoped_to_child(self): self.assertNotIn("BW_CLIENTID", os.environ) self.assertNotIn("BW_CLIENTSECRET", os.environ) def test_isolated_appdata_dir(self): self.assertEqual(oct(self.appdata.stat().st_mode & 0o777), "0o700") for c in read_calls(self.appdata): self.assertNotIn("Bitwarden CLI", c["cmdline"]) def test_config_server_called_for_vault(self): for c in read_calls(self.appdata): if c["argv"][1:3] == ["config", "server"]: self.assertEqual(c["argv"][3], "https://vault.paginica.com") return self.fail("bw config server never invoked") class TestCliBackendFailure(unittest.TestCase): """A failing `bw` must surface VAULT_BACKEND_ERROR -- never a silent fallback to legacy crypto.""" def test_backend_failure_no_fallback(self): with tempfile.TemporaryDirectory() as td: base = Path(td) cred = base / "cred" cred.mkdir() _write(cred / "id.txt", CLIENT_ID) _write(cred / "secret.token", CLIENT_SECRET) bw = base / "bw" bw.write_text("#!/bin/sh\nexit 3\n") os.chmod(bw, 0o755) appdata = base / "bw-appdata" appdata.mkdir(mode=0o700) cfg = Config({ "client_id_file": str(cred / "id.txt"), "client_secret_file": str(cred / "secret.token"), "vault_backend": "bitwarden-cli", "bw_path": str(bw), "bw_appdata_dir": str(appdata), }) core = BrokerCore(cfg) with self.assertRaises(BrokerError) as e: core.unlock(MASTER_PW) self.assertEqual(e.exception.code, "VAULT_BACKEND_ERROR") self.assertEqual(core.backend.name, "bitwarden-cli") def test_child_stderr_never_carries_secret(self): """A buggy/compromised child echoing the master password on stderr must still surface a sanitized error only.""" with tempfile.TemporaryDirectory() as td: base = Path(td) cred = base / "cred" cred.mkdir() _write(cred / "id.txt", CLIENT_ID) _write(cred / "secret.token", CLIENT_SECRET) # logged-in fake that leaks the unlock secret on stderr bw = base / "bw" bw.write_text( "#!/usr/bin/env python3\n" "import os, sys\n" "args = sys.argv[1:]\n" "if args[0] == 'status':\n" " print('{\"serverUrl\": \"https://vault.paginica.com\"," " \"status\": \"locked\"}'); sys.exit(0)\n" "if args[0] == 'unlock':\n" " var = args[args.index('--passwordenv') + 1]\n" " print('oops ' + os.environ.get(var, ''), file=sys.stderr)\n" " sys.exit(1)\n" "sys.exit(1)\n") os.chmod(bw, 0o755) appdata = base / "bw-appdata" appdata.mkdir(mode=0o700) cfg = Config({ "client_id_file": str(cred / "id.txt"), "client_secret_file": str(cred / "secret.token"), "vault_backend": "bitwarden-cli", "bw_path": str(bw), "bw_appdata_dir": str(appdata), }) core = BrokerCore(cfg) with self.assertRaises(BrokerError) as e: core.unlock(MASTER_PW) self.assertNotIn(MASTER_PW, e.exception.message) self.assertIn("[REDACTED]", e.exception.message) def test_unknown_backend_rejected(self): with tempfile.TemporaryDirectory() as td: base = Path(td) for name in ("id.txt", "secret.token"): _write(base / name, "x") cfg = Config({"vault_backend": "bogus", "client_id_file": str(base / "id.txt"), "client_secret_file": str(base / "secret.token")}) with self.assertRaises(BrokerError) as e: BrokerCore(cfg) self.assertEqual(e.exception.code, "UNCONFIGURED") def test_legacy_backend_rejected(self): """A stale 'legacy-crypto' config must fail loudly -- the custom Python crypto backend was removed and must never silently be replaced by another backend.""" with tempfile.TemporaryDirectory() as td: base = Path(td) for name in ("id.txt", "secret.token"): _write(base / name, "x") cfg = Config({"vault_backend": "legacy-crypto", "client_id_file": str(base / "id.txt"), "client_secret_file": str(base / "secret.token")}) with self.assertRaises(BrokerError) as e: BrokerCore(cfg) self.assertEqual(e.exception.code, "UNCONFIGURED") def test_missing_bw_binary(self): with tempfile.TemporaryDirectory() as td: base = Path(td) _write(base / "id.txt", CLIENT_ID) _write(base / "secret.token", CLIENT_SECRET) cfg = Config({"vault_backend": "bitwarden-cli", "bw_path": str(base / "no-such-bw"), "bw_appdata_dir": str(base / "app"), "client_id_file": str(base / "id.txt"), "client_secret_file": str(base / "secret.token")}) core = BrokerCore(cfg) self.assertFalse(core.status()["configured"]) with self.assertRaises(BrokerError) as e: core.unlock(MASTER_PW) self.assertEqual(e.exception.code, "UNCONFIGURED") if __name__ == "__main__": unittest.main(verbosity=2)