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.
217 lines
8.1 KiB
Python
217 lines
8.1 KiB
Python
"""Official Bitwarden CLI vault backend.
|
|
|
|
All vault cryptography (master-key derivation, EncString, sync format)
|
|
is delegated to the maintained `bw` client. The broker spawns it as a
|
|
subprocess with a dedicated, isolated profile:
|
|
|
|
vault server --(bw)--> private pipe --> broker --> adapters
|
|
|
|
Secret channels:
|
|
BW_CLIENTID / BW_CLIENTSECRET login child env only
|
|
AGENT_BROKER_MASTER_PW unlock child env only (--passwordenv)
|
|
BW_SESSION data ops child env only, held in memory,
|
|
never persisted, invalidated by `bw lock`
|
|
|
|
Secrets never appear on argv and `bw` stdout/stderr is never forwarded
|
|
raw -- errors surface sanitized excerpts only.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from .. import errors
|
|
from ..config import read_secret_file
|
|
from ..errors import BrokerError, backend_error
|
|
from ..items import metadata_projection, normalize_bw_item
|
|
from ..sanitize import sanitize_text
|
|
from . import VaultBackend
|
|
|
|
_PW_ENV = "AGENT_BROKER_MASTER_PW" # env var NAME only -- value lives in child env
|
|
_TIMEOUT = 90
|
|
|
|
|
|
class BitwardenCliBackend(VaultBackend):
|
|
name = "bitwarden-cli"
|
|
|
|
def __init__(self, config):
|
|
super().__init__(config)
|
|
self.bw = config.bw_path or shutil.which("bw")
|
|
self._session: str | None = None
|
|
self._meta: list[dict] = []
|
|
|
|
# -- plumbing --------------------------------------------------------
|
|
|
|
def configured(self) -> bool:
|
|
return (bool(self.bw) and Path(self.bw).exists()
|
|
and self.config.client_id_file.exists()
|
|
and self.config.client_secret_file.exists())
|
|
|
|
def _appdata(self) -> Path:
|
|
d = self.config.bw_appdata_dir
|
|
d.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(d, 0o700)
|
|
return d
|
|
|
|
def _env(self, extra: dict | None = None) -> dict:
|
|
"""Child-scoped environment. PATH puts the directory containing
|
|
`bw` first so its `#!/usr/bin/env node` shebang resolves the same
|
|
Node.js the CLI was installed with (a stale system node breaks
|
|
the WASM crypto layer). The npm `bw` is a symlink into lib/ --
|
|
do NOT resolve it: the bin dir itself holds the right `node`."""
|
|
bw_dir = str(Path(self.bw).parent)
|
|
env = {
|
|
"PATH": f"{bw_dir}:/usr/local/bin:/usr/bin:/bin",
|
|
"HOME": os.environ.get("HOME", "/"),
|
|
"LANG": "C.UTF-8",
|
|
"BITWARDENCLI_APPDATA_DIR": str(self._appdata()),
|
|
}
|
|
if extra:
|
|
env.update(extra)
|
|
return env
|
|
|
|
def _run(self, args: list[str], *, extra_env=None, secrets=(), stdin=None,
|
|
timeout=_TIMEOUT) -> tuple[int, str, str]:
|
|
"""Spawn `bw` with secrets confined to child env/stdin. Returns
|
|
RAW stdout/stderr -- callers must sanitize before surfacing."""
|
|
env = self._env(extra_env)
|
|
try:
|
|
proc = subprocess.run(
|
|
[self.bw, *args], env=env, input=stdin,
|
|
capture_output=True, text=True, timeout=timeout,
|
|
)
|
|
except FileNotFoundError:
|
|
raise backend_error(f"bitwarden CLI not found at {self.bw}")
|
|
except subprocess.TimeoutExpired:
|
|
raise backend_error(f"bw {args[0] if args else '?'} timed out")
|
|
return proc.returncode, proc.stdout, proc.stderr
|
|
|
|
def _fail(self, op: str, err: str, secrets) -> BrokerError:
|
|
detail = sanitize_text(err.strip(), secrets)[:300]
|
|
low = detail.lower()
|
|
if "vault is locked" in low or "not logged in" in low:
|
|
return errors.locked()
|
|
return backend_error(f"bw {op} failed: {detail or 'unknown error'}")
|
|
|
|
def _ok(self, op: str, rc: int, out: str, err: str, secrets=()) -> str:
|
|
if rc != 0:
|
|
raise self._fail(op, err, secrets)
|
|
return out
|
|
|
|
def _bw_status(self) -> dict:
|
|
rc, out, err = self._run(["status"])
|
|
self._ok("status", rc, out, err)
|
|
try:
|
|
return json.loads(out)
|
|
except json.JSONDecodeError:
|
|
raise backend_error("unparseable bw status output")
|
|
|
|
# -- VaultBackend -----------------------------------------------------
|
|
|
|
def _ensure_server(self, st: dict) -> dict:
|
|
if (st.get("serverUrl") or "").rstrip("/") != self.config.vault_url:
|
|
rc, out, err = self._run(["config", "server", self.config.vault_url])
|
|
self._ok("config", rc, out, err)
|
|
return st
|
|
|
|
def _login(self):
|
|
client_id = read_secret_file(self.config.client_id_file)
|
|
client_secret = read_secret_file(self.config.client_secret_file)
|
|
secrets = (client_id, client_secret)
|
|
try:
|
|
rc, out, err = self._run(
|
|
["login", "--apikey", "--nointeraction"],
|
|
extra_env={"BW_CLIENTID": client_id,
|
|
"BW_CLIENTSECRET": client_secret},
|
|
secrets=secrets)
|
|
if rc != 0:
|
|
raise self._fail("login", err, secrets)
|
|
# stdout may carry a session key -- captured, never surfaced
|
|
finally:
|
|
client_id = client_secret = None
|
|
|
|
def unlock(self, password: str) -> None:
|
|
st = self._ensure_server(self._bw_status())
|
|
if st.get("status") == "unauthenticated":
|
|
self._login()
|
|
secrets = (password,)
|
|
try:
|
|
rc, out, err = self._run(
|
|
["unlock", "--passwordenv", _PW_ENV, "--raw", "--nointeraction"],
|
|
extra_env={_PW_ENV: password}, secrets=secrets)
|
|
if rc != 0 or not out.strip():
|
|
clean = sanitize_text(err.strip(), secrets)
|
|
detail = ([l for l in clean.splitlines() if l.strip()] or [""])[-1][:200]
|
|
raise BrokerError(
|
|
"UNLOCK_FAILED",
|
|
f"unlock failed: {detail or 'master password rejected'}")
|
|
self._session = out.strip().splitlines()[-1].strip()
|
|
finally:
|
|
password = None
|
|
self._sync()
|
|
raw = self._list_items_raw()
|
|
self._meta = [metadata_projection(normalize_bw_item(i)) for i in raw]
|
|
raw = None
|
|
|
|
def lock(self) -> None:
|
|
try:
|
|
if self._session:
|
|
self._run(["lock", "--nointeraction"], secrets=(self._session,))
|
|
except errors.BrokerError:
|
|
pass
|
|
finally:
|
|
self._session = None
|
|
self._meta = []
|
|
|
|
def logout(self) -> None:
|
|
try:
|
|
self._run(["logout", "--nointeraction"], secrets=(self._session,) if self._session else ())
|
|
except errors.BrokerError:
|
|
pass
|
|
finally:
|
|
self._session = None
|
|
self._meta = []
|
|
|
|
def list_metadata(self) -> list[dict]:
|
|
return self._meta
|
|
|
|
def _data_env(self) -> dict:
|
|
if not self._session:
|
|
raise errors.locked()
|
|
return {"BW_SESSION": self._session}
|
|
|
|
def _sync(self):
|
|
secrets = (self._session,)
|
|
rc, out, err = self._run(["sync", "--nointeraction"],
|
|
extra_env=self._data_env(), secrets=secrets)
|
|
self._ok("sync", rc, out, err, secrets)
|
|
|
|
def _list_items_raw(self) -> list[dict]:
|
|
secrets = (self._session,)
|
|
rc, out, err = self._run(["list", "items", "--nointeraction"],
|
|
extra_env=self._data_env(), secrets=secrets)
|
|
self._ok("list items", rc, out, err, secrets)
|
|
try:
|
|
data = json.loads(out)
|
|
except json.JSONDecodeError:
|
|
raise backend_error("unparseable bw list output")
|
|
if not isinstance(data, list):
|
|
raise backend_error("unexpected bw list output")
|
|
return data
|
|
|
|
def get_item(self, item_id: str) -> dict:
|
|
secrets = (self._session,)
|
|
rc, out, err = self._run(
|
|
["get", "item", str(item_id), "--nointeraction"],
|
|
extra_env=self._data_env(), secrets=secrets)
|
|
if rc != 0:
|
|
raise self._fail("get item", err, secrets)
|
|
try:
|
|
return normalize_bw_item(json.loads(out))
|
|
except json.JSONDecodeError:
|
|
raise backend_error("unparseable bw get output")
|