me
be87900794
Mechanical rename (package agent_credential_broker; entry points
agent-broker{,-mcp,-plugin}; env AGENT_BROKER_*; paths agent-broker;
agent-browser credential provider 'broker'; MCP server name agent-broker).
New backend 'openbao': stdlib-only HTTP client for OpenBao/Vault KV v2
(approle login or token file/env; metadata-only ref index; lazy re-login
on 401/403; token renewal). bitwarden-cli backend kept as transition.
391 lines
16 KiB
Python
391 lines
16 KiB
Python
"""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)
|