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.
166 lines
6.6 KiB
Python
166 lines
6.6 KiB
Python
"""Broker configuration: files, paths and hardening defaults.
|
|
|
|
No secret material ever lives in this module -- only *paths* to secret
|
|
files and socket endpoints.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
from .errors import BrokerError
|
|
|
|
DEFAULT_VAULT_URL = "https://vault.paginica.com"
|
|
|
|
CREDENTIALS_DIR = Path.home() / ".config" / "devin" / "credentials"
|
|
CLIENT_ID_FILE = CREDENTIALS_DIR / "bitwarden-client-id.txt"
|
|
CLIENT_SECRET_FILE = CREDENTIALS_DIR / "bitwarden-client-secret.token"
|
|
|
|
# OpenBao defaults: loopback dev agent (see backlog 03); TLS via
|
|
# openbao_ca_cert / openbao_tls_skip_verify when the listener is exposed.
|
|
DEFAULT_OPENBAO_ADDR = "http://127.0.0.1:18200"
|
|
OPENBAO_ROLE_ID_FILE = CREDENTIALS_DIR / "openbao-role-id.txt"
|
|
OPENBAO_SECRET_ID_FILE = CREDENTIALS_DIR / "openbao-secret-id.txt"
|
|
OPENBAO_TOKEN_FILE = CREDENTIALS_DIR / "openbao-token.txt"
|
|
OPENBAO_TOKEN_ENV = "AGENT_BROKER_OPENBAO_TOKEN"
|
|
|
|
|
|
def runtime_dir() -> Path:
|
|
base = os.environ.get("XDG_RUNTIME_DIR") or f"/run/user/{os.getuid()}"
|
|
p = Path(base) / "agent-broker"
|
|
p.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(p, 0o700)
|
|
return p
|
|
|
|
|
|
def socket_path() -> Path:
|
|
return Path(os.environ.get("AGENT_BROKER_SOCKET", str(runtime_dir() / "broker.sock")))
|
|
|
|
|
|
def state_dir() -> Path:
|
|
base = os.environ.get("XDG_STATE_HOME") or str(Path.home() / ".local" / "state")
|
|
p = Path(base) / "agent-broker"
|
|
p.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(p, 0o700)
|
|
return p
|
|
|
|
|
|
def audit_log_path() -> Path:
|
|
p = state_dir() / "audit.jsonl"
|
|
if not p.exists():
|
|
p.touch(mode=0o600)
|
|
os.chmod(p, 0o600)
|
|
return p
|
|
|
|
|
|
def tmp_dir() -> Path:
|
|
"""Secret-bearing temp files live ONLY on the runtime tmpfs."""
|
|
p = runtime_dir() / "tmp"
|
|
p.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(p, 0o700)
|
|
return p
|
|
|
|
|
|
def config_path() -> Path:
|
|
return Path(os.environ.get(
|
|
"AGENT_BROKER_CONFIG",
|
|
str(Path.home() / ".config" / "agent-broker" / "config.json"),
|
|
))
|
|
|
|
|
|
def profiles_dir() -> Path:
|
|
env = os.environ.get("AGENT_BROKER_PROFILES_DIR")
|
|
if env:
|
|
return Path(env)
|
|
default = Path(__file__).resolve().parent.parent / "profiles"
|
|
if default.is_dir():
|
|
return default
|
|
return Path.home() / ".config" / "agent-broker" / "profiles"
|
|
|
|
|
|
class Config:
|
|
def __init__(self, data: dict):
|
|
self.vault_url = data.get("vault_url", DEFAULT_VAULT_URL).rstrip("/")
|
|
self.client_id_file = Path(data.get("client_id_file", str(CLIENT_ID_FILE)))
|
|
self.client_secret_file = Path(data.get("client_secret_file", str(CLIENT_SECRET_FILE)))
|
|
self.request_timeout = float(data.get("request_timeout", 15))
|
|
self.unlock_ttl_seconds = float(data.get("unlock_ttl_seconds", 3600))
|
|
self.confirmation_ttl_seconds = float(data.get("confirmation_ttl_seconds", 120))
|
|
self.max_response_bytes = int(data.get("max_response_bytes", 256 * 1024))
|
|
self.git_workdir_roots = data.get("git_workdir_roots", [])
|
|
# Optional non-interactive unlock: when set, `unlock` reads the
|
|
# master password from this file (same rules as credential files:
|
|
# 600 perms, owned by this user) instead of prompting via getpass.
|
|
self.master_password_file = (
|
|
Path(data["master_password_file"])
|
|
if data.get("master_password_file") else None
|
|
)
|
|
# Vault backend selection. "bitwarden-cli" (transition backend)
|
|
# and "openbao" are supported; anything else (including stale
|
|
# "legacy-crypto") fails loudly at daemon start -- no fallback.
|
|
self.vault_backend = data.get("vault_backend", "bitwarden-cli")
|
|
self.bw_path = data.get("bw_path") # None -> PATH lookup
|
|
self.bw_appdata_dir = Path(data.get(
|
|
"bw_appdata_dir",
|
|
str(Path.home() / ".local" / "share" / "agent-broker"
|
|
/ "bitwarden-cli")))
|
|
|
|
# --- OpenBao backend (vault_backend = "openbao") ---------------
|
|
# Items live at <mount>/data/<prefix>/<ref> (KV v2). Auth is
|
|
# non-interactive by design: an AppRole (role_id+secret_id read
|
|
# from 0600 files, same rules as client_id_file) or a direct
|
|
# token from a 0600 file / env var.
|
|
self.openbao_addr = data.get(
|
|
"openbao_addr", DEFAULT_OPENBAO_ADDR).rstrip("/")
|
|
self.openbao_mount = str(data.get("openbao_mount", "secret")).strip("/")
|
|
self.openbao_prefix = str(data.get("openbao_prefix", "agent")).strip("/")
|
|
# Optional namespace -> X-Vault-Namespace header on every request.
|
|
self.openbao_namespace = data.get("openbao_namespace") or None
|
|
# TLS: CA bundle path, or dev-only verification skip.
|
|
self.openbao_ca_cert = data.get("openbao_ca_cert") or None
|
|
self.openbao_tls_skip_verify = bool(
|
|
data.get("openbao_tls_skip_verify", False))
|
|
self.openbao_role_id_file = Path(data.get(
|
|
"openbao_role_id_file", str(OPENBAO_ROLE_ID_FILE)))
|
|
self.openbao_secret_id_file = Path(data.get(
|
|
"openbao_secret_id_file", str(OPENBAO_SECRET_ID_FILE)))
|
|
self.openbao_token_file = Path(data.get(
|
|
"openbao_token_file", str(OPENBAO_TOKEN_FILE)))
|
|
# Name of the env var holding a direct token (override with
|
|
# openbao_token_env; default AGENT_BROKER_OPENBAO_TOKEN).
|
|
self.openbao_token_env = data.get(
|
|
"openbao_token_env", OPENBAO_TOKEN_ENV) or OPENBAO_TOKEN_ENV
|
|
|
|
@classmethod
|
|
def load(cls) -> "Config":
|
|
path = config_path()
|
|
if path.exists():
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except Exception as e:
|
|
raise BrokerError("UNCONFIGURED", f"invalid config file {path}: {e}")
|
|
if not isinstance(data, dict):
|
|
raise BrokerError("UNCONFIGURED", f"invalid config file {path}")
|
|
else:
|
|
data = {}
|
|
return cls(data)
|
|
|
|
|
|
def read_secret_file(path: Path) -> str:
|
|
"""Read a local credential file for internal use only. Enforces that
|
|
the file is not group/other-readable."""
|
|
if not path.exists():
|
|
raise BrokerError("UNCONFIGURED", f"credential file not found: {path}")
|
|
st = path.stat()
|
|
if st.st_mode & (stat.S_IRWXG | stat.S_IRWXO):
|
|
raise BrokerError(
|
|
"UNCONFIGURED",
|
|
f"credential file {path} has unsafe permissions {oct(st.st_mode & 0o777)}; chmod 600",
|
|
)
|
|
if st.st_uid not in (os.getuid(), 0):
|
|
raise BrokerError("UNCONFIGURED", f"credential file {path} not owned by this user")
|
|
return path.read_text().strip()
|