diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b0113b4..81b4670c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,28 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] ### Added +- **dedicated agent kbs — `vouch kb create --for-agent `** (#609): a + scheduled agent (a pr triager in ci, an incident summariser) produces a lot of + narrow, machine-shaped memory, and mixing it into the project kb buries what a + human curated. `vouch kb create` provisions an isolated kb outside any project + tree, registers it with an **owner and the agent it serves**, and issues a + credential **bound to that kb and only that kb** — printed once, never + persisted; what lands on disk is the 16-hex subject `trust.py` already derives. + The binding is what makes this worth more than `vouch init` in a subdirectory: + upward discovery makes which kb an agent hits a function of its working + directory, so a leaked ci secret otherwise reaches the project kb from the + wrong cwd. It is enforced at the existing transport chokepoint + (`agents.subject_is_active`) and lives machine-local in + `~/.config/vouch/credentials.yaml` (0600) — a binding stored *inside* the kb it + names is circular, since the project kb's copy has nothing to say about a token + issued for another kb and would still fall open. **Unbound tokens are + unaffected**, so this is opt-in rather than a migration; a bound one fails + closed everywhere but its own kb, including against a kb whose `kb.id` was + stripped. Ambient capture refuses to write into an agent kb reached by upward + discovery, the same guard the personal catch-all already gets. `vouch kb issue` + is the rotation path, `vouch kb list` shows what exists, `vouch agents revoke` + still retires a credential, and `vouch adopt` still promotes anything worth + keeping into the project kb. - **bench: composite guards** (#616): `efficiency`, `consistency` and `canary` as bounded multipliers over the composite, plus a `bench_version` stamp on every report. Reported **beside** the composite, never folded into it — diff --git a/docs/multi-agent.md b/docs/multi-agent.md index 2ea14edd..aeaabe92 100644 --- a/docs/multi-agent.md +++ b/docs/multi-agent.md @@ -27,6 +27,62 @@ A common convention: `-` so you can tell apart "Alice running Claude Code" from "Bob running Claude Code" from "Alice running Cursor". +## Dedicated KBs for scheduled agents + +`VOUCH_AGENT` separates *attribution* inside one KB. An agent that runs +on a schedule — a PR triager in CI, an incident summariser, a docs bot — +usually wants separation of *storage* too: it produces a lot of narrow, +machine-shaped memory, and a thousand run logs bury the knowledge a human +curated. + +```bash +vouch kb create ci-triage --for-agent pr-triager +``` + +That provisions an isolated KB under `$XDG_DATA_HOME/vouch/kbs/` (never +inside a project tree), registers it in the machine registry as owned by +you and provisioned for `pr-triager`, and prints a credential **once**: + +``` + VOUCH_TOKEN_PR_TRIAGER=… +``` + +vouch never stores that secret. What it keeps is the token's 16-hex +subject — in the KB's own committed `agents.yaml`, and in a machine-local +binding file at `~/.config/vouch/credentials.yaml` (mode 0600). The +binding is the part that matters: **that credential authenticates against +that KB and no other.** + +Why the binding rather than just running `vouch init` somewhere else: +which KB an agent hits is normally a function of its working directory, +because discovery walks upward. A CI job that starts in the wrong +directory — or a leaked token used from one — would otherwise reach the +project KB. A bound credential fails closed everywhere but its own KB, +regardless of cwd. + +Point the agent at it with both halves: + +```bash +export VOUCH_TOKEN_PR_TRIAGER=… # the credential +export VOUCH_KB_PATH=~/.local/share/vouch/kbs/ci-triage/.vouch +``` + +Housekeeping: + +```bash +vouch kb list # what exists, and how many credentials each has +vouch kb issue ci-triage --for-agent triager-2 # rotate / add a credential +vouch agents revoke pr-triager # retire one (terminal, by design) +vouch adopt … # promote what's worth keeping into the project KB +``` + +Two properties worth knowing. **Tokens with no binding are unaffected** — +an existing deployment keeps working exactly as before, so this is opt-in +rather than a migration. And ambient capture *refuses* to write into an +agent KB it reached by walking up from a project directory, the same +guard the personal catch-all KB gets; set `VOUCH_KB_PATH` when you mean +it deliberately. + ## Concurrency vouch is single-writer per file. Two agents proposing simultaneously diff --git a/src/vouch/agents.py b/src/vouch/agents.py index c0af370d..2a854b9d 100644 --- a/src/vouch/agents.py +++ b/src/vouch/agents.py @@ -275,10 +275,20 @@ def is_active(store: KBStore, subject: str) -> bool: def subject_is_active(subject: str) -> bool: """Store-resolving gate for the transport chokepoint. + Two questions, one gate: is this subject's row active *here*, and may this + subject reach this KB at all (`kb_binding`)? The second is what isolates a + credential issued for an agent's own KB — without it the token would still + authenticate against the project KB, because the answer below for a + subject this KB has never heard of is deliberately "yes". + Best-effort by design: a request that cannot resolve a KB has no registry to be denied by, and an unreadable registry must not lock every agent out of a running server. """ + # Imported here rather than at module scope for the same reason `trust.py` + # takes this gate as an injected callable: the auth path must not drag the + # storage layer into every importer of this module. + from . import kb_binding from .storage import KBStore, discover_root try: @@ -286,7 +296,7 @@ def subject_is_active(subject: str) -> bool: except Exception: return True try: - return is_active(store, subject) + return is_active(store, subject) and kb_binding.store_allows(store, subject) except Exception: # pragma: no cover - defensive logger.debug("agents: registry unreadable, allowing subject") return True diff --git a/src/vouch/cli.py b/src/vouch/cli.py index 1f1aebc8..e565b5d7 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -41,6 +41,7 @@ from . import hub as hub_mod from . import inbox as inbox_mod from . import install_adapter as install_mod +from . import kb_binding as kb_binding_mod from . import lifecycle as life from . import media as media_mod from . import metrics as metrics_mod @@ -124,6 +125,7 @@ def _cli_errors() -> Iterator[None]: codex_rollout_mod.CodexRolloutError, agents_mod.AgentError, pins_mod.PinError, + kb_binding_mod.BindingError, ) as e: raise click.ClickException(str(e)) from e @@ -395,6 +397,227 @@ def hub_unregister(token: str) -> None: click.echo(f"unregistered {removed.name} ({removed.kb_id})") +# --- dedicated agent KBs (#609) ------------------------------------------- + + +@cli.group(name="kb") +def kb_group() -> None: + """Provision KBs, including dedicated ones for scheduled agents. + + An agent that runs on a schedule produces a lot of narrow, machine-shaped + memory; mixing it into the project KB buries what a human curated. A + dedicated KB keeps it separate, and the credential issued alongside it + reaches that KB and no other — so a leaked CI secret cannot write to the + project KB no matter which directory the agent runs from. + """ + + +def _issue_agent_credential( + store: KBStore, *, agent: str, actor: str, note: str | None +) -> tuple[str, str]: + """Mint a credential for `agent`, bind it to `store`'s KB, register it. + + Returns (token, env_var). The token is returned for a single echo and is + never written anywhere by vouch: what lands on disk is its 16-hex subject, + in the KB's own committed agent registry and in the machine-local binding + file. Recovering it later is impossible by construction — issue a new one. + """ + identity = store.identity() + if identity is None: # pragma: no cover - init always mints + raise click.ClickException(f"KB at {store.root} has no identity to bind to") + kb_id, kb_name = identity + token = kb_binding_mod.issue_token() + subject = trust_mod.auth_subject_for_token(token) + agents_mod.register( + store, subject=subject, name=agent, actor=actor, + note=note or f"dedicated KB {kb_name}", + ) + kb_binding_mod.bind( + subject=subject, kb_id=kb_id, kb_name=kb_name, agent=agent, + ) + audit_mod.log_event( + store.kb_dir, event="agent.bind", actor=actor, + data={"subject": subject, "kb_id": kb_id, "agent": agent}, + ) + return token, kb_binding_mod.token_env_var(agent) + + +def _echo_credential(token: str, env_var: str, kb_dir: Path) -> None: + """Print a freshly issued credential once, with what to do with it.""" + click.echo("") + click.echo(f" {env_var}={token}") + click.echo("") + click.echo( + "This is the only time the token is shown — vouch stores its " + "fingerprint, never the secret." + ) + click.echo( + f"Export it where the agent runs, then point the agent at this KB:\n" + f" export {env_var}=...\n" + f" export VOUCH_KB_PATH={kb_dir}" + ) + + +@kb_group.command("create") +@click.argument("name") +@click.option( + "--for-agent", + "for_agent", + default=None, + help="Provision this KB for an agent and issue it a credential bound to " + "this KB alone.", +) +@click.option( + "--path", + default=None, + type=click.Path(file_okay=False), + help="Where to create it (default: $XDG_DATA_HOME/vouch/kbs/).", +) +@click.option("--note", default=None, help="What this agent is for.") +def kb_create( + name: str, for_agent: str | None, path: str | None, note: str | None +) -> None: + """Create a KB called NAME and register it on this machine. + + With --for-agent, the KB is provisioned for that agent: the registry row + records who owns it, ambient capture from a project directory refuses to + write into it, and a fresh credential is printed once, bound to this KB. + Promote anything worth keeping into the project KB with `vouch adopt`. + """ + if path is not None: + root = Path(path).expanduser().resolve() + else: + with _cli_errors(): + derived = hub_mod.agent_kb_root(name) + if derived is None: + raise click.ClickException( + "cannot determine a home for the KB — pass --path or set " + f"{hub_mod.AGENT_KBS_ENV} to a writable folder" + ) + root = derived + if (root / ".vouch").is_dir(): + raise click.ClickException( + f"a KB already exists at {root} — choose another name, or run " + f"`vouch kb issue {name}` to issue it another credential" + ) + with _cli_errors(): + try: + store, _seed, _tmpl = _bootstrap_kb(root) + except Exception as e: + # Same rollback as the personal KB: an unwritable path must not + # leave half a KB behind for a rerun to mistake for a finished one. + shutil.rmtree(root / ".vouch", ignore_errors=True) + raise click.ClickException(f"could not initialise the KB at {root}: {e}") from e + entry = hub_mod.register_kb( + root, name=name, actor=_whoami(), + owner=_whoami(), agent=for_agent or "", + ) + click.echo(f"Initialised KB at {store.kb_dir}") + click.echo(f"Registered in the machine registry: {entry.name} ({entry.kb_id})") + if for_agent is None: + return + with _cli_errors(): + token, env_var = _issue_agent_credential( + store, agent=for_agent, actor=_whoami(), note=note + ) + _write_serve_token_ref(store, env_var) + click.echo(f"Issued a credential for {for_agent}, bound to this KB only.") + _echo_credential(token, env_var, store.kb_dir) + + +@kb_group.command("issue") +@click.argument("name") +@click.option("--for-agent", "for_agent", required=True, help="The agent to issue for.") +@click.option("--note", default=None, help="What this agent is for.") +def kb_issue(name: str, for_agent: str, note: str | None) -> None: + """Issue another credential for the existing KB called NAME. + + The rotation path: a leaked credential is retired with `vouch agents + revoke`, which is terminal, and re-admitting an agent means giving it a + new token — which is a new subject, bound afresh to this KB. + """ + entry = _agent_kb_entry(name) + store = KBStore(Path(entry.path)) + with _cli_errors(): + token, env_var = _issue_agent_credential( + store, agent=for_agent, actor=_whoami(), note=note + ) + click.echo(f"Issued a credential for {for_agent}, bound to {entry.name} only.") + _echo_credential(token, env_var, store.kb_dir) + if env_var not in store.config_path.read_text(encoding="utf-8"): + click.echo( + f"note: add `env:{env_var}` to serve.bearer_tokens in " + f"{store.config_path} for this KB to accept it.", + err=True, + ) + + +@kb_group.command("list") +@click.option("--json", "as_json", is_flag=True, help="Emit JSON instead of a table.") +def kb_list(as_json: bool) -> None: + """List the KBs provisioned for agents on this machine.""" + entries = hub_mod.agent_entries() + if as_json: + _emit_json( + { + "registry": str(hub_mod.registry_path()), + "kbs": [ + { + "kb_id": e.kb_id, + "name": e.name, + "path": e.path, + "owner": e.owner, + "agent": e.agent, + "added_at": e.added_at, + "credentials": len(kb_binding_mod.bindings_for_kb(e.kb_id)), + } + for e in entries + ], + } + ) + return + if not entries: + click.echo("no agent KBs on this machine (vouch kb create --for-agent )") + return + for e in entries: + creds = len(kb_binding_mod.bindings_for_kb(e.kb_id)) + click.echo( + f"{e.name} agent={e.agent} owner={e.owner or '-'} " + f"credentials={creds} {e.path}" + ) + + +def _agent_kb_entry(name: str) -> hub_mod.RegistryEntry: + """The registry row for an agent KB called `name`, or a clean error.""" + entry = next((e for e in hub_mod.agent_entries() if e.name == name), None) + if entry is None: + raise click.ClickException( + f"no agent KB called {name!r} — `vouch kb list` shows what exists" + ) + if not (Path(entry.path) / ".vouch").is_dir(): + raise click.ClickException( + f"agent KB {name!r} is registered at {entry.path} but there is no " + ".vouch/ there — it moved or was deleted" + ) + return entry + + +def _write_serve_token_ref(store: KBStore, env_var: str) -> None: + """Point the fresh KB's accept-list at the credential's env var. + + Only ever called on a KB `_bootstrap_kb` just wrote, so a structural + rewrite is safe here — there are no hand-written comments to lose, which + is the reason `hub.set_personal_fallback` goes to textual lengths. + The `env:` indirection is what keeps the secret out of a committed file. + """ + loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) or {} + serve = loaded.setdefault("serve", {}) + serve["bearer_tokens"] = [f"env:{env_var}"] + store.config_path.write_text( + yaml.safe_dump(loaded, sort_keys=False, allow_unicode=True), encoding="utf-8" + ) + + def _init_personal_kb(fallback: bool | None) -> Path: """Create + register the personal catch-all KB; shared by the two entry points (`vouch hub init-personal` and `install-mcp --global`'s opt-in) diff --git a/src/vouch/hub.py b/src/vouch/hub.py index e891760c..1760beb0 100644 --- a/src/vouch/hub.py +++ b/src/vouch/hub.py @@ -2,7 +2,8 @@ `~/.config/vouch/registry.yaml` (override with VOUCH_REGISTRY_PATH; honours XDG_CONFIG_HOME) lists known KBs — one row per KB instance: id, display -name, role (project | personal | team), path. The registry is advisory +name, role (project | personal | team), path, and who it belongs to (owner, +and the agent it was provisioned for, if any). The registry is advisory routing state, never authority: identity and content live in each KB's own `.vouch/`, so a stale or deleted registry degrades to today's per-project behaviour instead of breaking anything. It is machine-local and never @@ -13,10 +14,16 @@ daemon start here as plain functions over a YAML file. `resolve()` wraps `storage.discover_root` with the registry-aware safety -check: a KB registered with role `personal` is never an ambient capture -target for a directory below it — capture refuses, reads warn. This is the -second belt on top of the structural $HOME walk-stop in `discover_root` -(which needs no registry state at all). +check: a KB that belongs to someone other than the directory it was found +from — a `personal` catch-all, or a KB provisioned for an agent — is never +an ambient capture target for a directory below it, so capture refuses and +reads warn. This is the second belt on top of the structural $HOME walk-stop +in `discover_root` (which needs no registry state at all). + +Ownership here is routing metadata, and only that. It records *whose* KB this +is so `vouch kb list` can say so and so the guard above knows a machine-owned +KB when it sees one. What a credential may actually reach is a separate, +enforced question — see `kb_binding.py`. """ from __future__ import annotations @@ -38,6 +45,7 @@ REGISTRY_ENV = "VOUCH_REGISTRY_PATH" PERSONAL_KB_ENV = "VOUCH_PERSONAL_KB" +AGENT_KBS_ENV = "VOUCH_AGENT_KBS_DIR" REGISTRY_VERSION = 1 ROLES = ("project", "personal", "team") @@ -59,6 +67,18 @@ class RegistryEntry: role: str path: str added_at: str + # Who this KB belongs to. `owner` is the human accountable for it; + # `agent` names the agent it was provisioned for, and is what makes a KB + # machine-owned. Both default empty so every row written before they + # existed reads back as a human-owned KB with an unrecorded owner — + # which is exactly what those rows are. + owner: str = "" + agent: str = "" + + @property + def machine_owned(self) -> bool: + """Whether this KB exists to hold one agent's memory.""" + return bool(self.agent) def _parse_entry(raw: object) -> RegistryEntry | None: @@ -79,6 +99,8 @@ def _parse_entry(raw: object) -> RegistryEntry | None: role=str(role), path=path, added_at=str(raw.get("added_at") or ""), + owner=str(raw.get("owner") or ""), + agent=str(raw.get("agent") or ""), ) @@ -128,22 +150,30 @@ def _registry_lock(p: Path) -> Iterator[None]: os.close(fd) +def _row(e: RegistryEntry) -> dict[str, Any]: + """One serialized registry row. Empty ownership fields are omitted so a + registry full of ordinary project KBs reads the same as it always did.""" + row: dict[str, Any] = { + "kb_id": e.kb_id, + "name": e.name, + "role": e.role, + "path": e.path, + "added_at": e.added_at, + } + if e.owner: + row["owner"] = e.owner + if e.agent: + row["agent"] = e.agent + return row + + def save_registry(entries: list[RegistryEntry], path: Path | None = None) -> Path: """Atomically write the registry (unique tmp file + rename).""" p = path or registry_path() p.parent.mkdir(parents=True, exist_ok=True) body: dict[str, Any] = { "version": REGISTRY_VERSION, - "kbs": [ - { - "kb_id": e.kb_id, - "name": e.name, - "role": e.role, - "path": e.path, - "added_at": e.added_at, - } - for e in entries - ], + "kbs": [_row(e) for e in entries], } # A per-writer tempfile (not a shared fixed name) so two concurrent # writers can never truncate or rename-steal each other's staging file. @@ -187,9 +217,16 @@ def register_kb( role: str = "project", name: str | None = None, actor: str, + owner: str | None = None, + agent: str | None = None, path: Path | None = None, ) -> RegistryEntry: - """Add (or refresh) one KB in the machine registry. Idempotent on kb_id.""" + """Add (or refresh) one KB in the machine registry. Idempotent on kb_id. + + ``owner`` and ``agent`` are sticky: passing None on a refresh keeps what + the row already said, so re-running plain `vouch hub register` over an + agent-owned KB does not quietly launder it into a human-owned one. + """ root = root.resolve() if not (root / KB_DIRNAME).is_dir(): raise KBNotFoundError(f"no {KB_DIRNAME}/ at {root} — run `vouch init` there first") @@ -197,27 +234,21 @@ def register_kb( raise ValueError(f"role must be one of {ROLES}, got {role!r}") store = KBStore(root) kb_id, kb_name = ensure_kb_identity(store, actor=actor) - entry = RegistryEntry( - kb_id=kb_id, - name=name or kb_name, - role=role, - path=str(root), - added_at=utcnow_iso(), - ) with _registry_lock(path or registry_path()): existing = load_registry(path) entries = [e for e in existing if e.kb_id != kb_id] # A moved/re-registered KB keeps one row: the kb_id is the key, the # path is metadata. Preserve the original added_at on refresh. previous = next((e for e in existing if e.kb_id == kb_id), None) - if previous is not None and previous.added_at: - entry = RegistryEntry( - kb_id=entry.kb_id, - name=entry.name, - role=entry.role, - path=entry.path, - added_at=previous.added_at, - ) + entry = RegistryEntry( + kb_id=kb_id, + name=name or kb_name, + role=role, + path=str(root), + added_at=(previous.added_at if previous and previous.added_at else utcnow_iso()), + owner=owner if owner is not None else (previous.owner if previous else ""), + agent=agent if agent is not None else (previous.agent if previous else ""), + ) entries.append(entry) save_registry(entries, path) return entry @@ -292,15 +323,23 @@ def resolve(start: Path | None = None) -> Resolution: return Resolution(root=root, why=trace) entry = entry_for_root(root) - if entry is not None and entry.role == "personal": + if entry is not None and (entry.role == "personal" or entry.machine_owned): origin = (start or Path.cwd()).resolve() if os.environ.get("VOUCH_PROJECT_DIR") and start is None: candidate = Path(os.environ["VOUCH_PROJECT_DIR"]) if candidate.is_dir(): origin = candidate.resolve() if origin != root.resolve(): + # An agent KB caught by upward discovery is the same hazard as the + # personal catch-all: a project's sessions would land in a store + # curated for something else entirely. + kind = ( + f"provisioned for agent {entry.agent!r}" + if entry.machine_owned + else "registered as a personal KB" + ) guard = ( - f"KB at {root} is registered as a personal KB; refusing ambient " + f"KB at {root} is {kind}; refusing ambient " f"capture from {origin}. Run `vouch init` in the project root, or " f"set VOUCH_KB_PATH={root / KB_DIRNAME} to target it deliberately." ) @@ -342,6 +381,39 @@ def personal_kb_root() -> Path | None: return home / ".local" / "share" / "vouch" / "personal" +def agent_kb_root(name: str) -> Path | None: + """Where a KB provisioned for an agent lives, by name. + + ``VOUCH_AGENT_KBS_DIR`` > ``$XDG_DATA_HOME/vouch/kbs`` > + ``~/.local/share/vouch/kbs``, mirroring `personal_kb_root`: content, so a + data path. Deliberately *outside* any project tree — a KB that sits above + a project is one that upward discovery can capture into by accident, and + the whole point of a dedicated agent KB is that it cannot be reached + without asking for it. None when no home can be determined (containers). + """ + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-.") + if not slug: + raise ValueError(f"cannot derive a directory name from {name!r}") + forced = os.environ.get(AGENT_KBS_ENV) + if forced: + return Path(forced).expanduser() / slug + xdg = os.environ.get("XDG_DATA_HOME") + if xdg: + return Path(xdg) / "vouch" / "kbs" / slug + try: + home = Path.home() + except RuntimeError: + return None + return home / ".local" / "share" / "vouch" / "kbs" / slug + + +def agent_entries(*, path: Path | None = None) -> list[RegistryEntry]: + """Every machine-owned row, newest-registered first.""" + rows = [e for e in load_registry(path) if e.machine_owned] + rows.sort(key=lambda e: e.added_at, reverse=True) + return rows + + def personal_entries(*, path: Path | None = None) -> list[RegistryEntry]: """Every personal-role row, live ones (a real ``.vouch/`` on disk) first. diff --git a/src/vouch/kb_binding.py b/src/vouch/kb_binding.py new file mode 100644 index 00000000..3d658032 --- /dev/null +++ b/src/vouch/kb_binding.py @@ -0,0 +1,279 @@ +"""Credential-to-KB binding: a token that reaches exactly one KB (#609). + +A scheduled agent — a PR triager in CI, an incident summariser — gets its own +KB so a thousand machine-written memories never bury the knowledge a human +curated. Provisioning that KB is only half the isolation. The other half is +the credential: without a binding, a token issued for the agent's KB still +authenticates against the project KB, because `agents.is_active` deliberately +fails open for subjects it has never heard of (an existing deployment whose +token predates the registry must keep working). + +So the binding lives here, keyed on the ``auth_subject`` — the sha256 prefix +``trust.py`` already derives — and never on the credential itself. Registering +a binding does not require storing, echoing, or even seeing the secret again, +which is the same split `agents.py` draws and the reason the agent registry +can live in committed YAML. + +Two properties follow, and they are the whole point: + +* **Unbound subjects are unaffected.** A token with no binding row + authenticates exactly as before, against whatever KB it is presented to. + Binding is opt-in, not a migration. +* **A bound subject reaches one KB and fails closed everywhere else.** That + is what makes a leaked CI secret harmless to the project KB *regardless of + the agent's working directory* — the failure mode `vouch init` in a + subdirectory cannot protect against, because upward discovery makes the + target a function of cwd. + +The file is machine-local (``~/.config/vouch/credentials.yaml``, 0600) rather +than committed inside a KB, for a reason worth stating: a binding stored in +the KB it names is circular — KB-B's copy has nothing to say about a token +issued for KB-A, so presenting that token to KB-B would still fall open. The +question "which KB may this subject reach" is machine-scoped, so the answer +has to be too. This is the same trust level as the bearer accept-list it +guards: anyone who can rewrite this file can already rewrite `serve. +bearer_tokens`. + +Unlike the hub registry — advisory routing state that degrades to per-project +behaviour when absent — this file is consulted for an authorization decision. +A missing or unreadable one therefore degrades to *today's* behaviour (no +bindings, nothing denied), never to "deny everything": an unreadable file must +not lock every agent out of a running server. +""" + +from __future__ import annotations + +import os +import re +import secrets +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import yaml + +from .hub import _registry_lock as _file_lock # one lock implementation, not two +from .models import utcnow_iso + +if TYPE_CHECKING: # pragma: no cover - typing only + from .storage import KBStore + +BINDINGS_ENV = "VOUCH_CREDENTIALS_PATH" +BINDINGS_VERSION = 1 + +# Long enough that guessing is hopeless, url-safe so it survives an env var, +# a CI secret store and a shell without quoting games. +TOKEN_BYTES = 32 + + +class BindingError(RuntimeError): + """A binding could not be created or removed.""" + + +def bindings_path() -> Path: + """Where the binding file lives (env > XDG_CONFIG_HOME > ~/.config). + + A sibling of the hub registry: both are machine-local config about KBs + rather than content belonging to any one of them. + """ + forced = os.environ.get(BINDINGS_ENV) + if forced: + return Path(forced) + xdg = os.environ.get("XDG_CONFIG_HOME") + base = Path(xdg) if xdg else Path.home() / ".config" + return base / "vouch" / "credentials.yaml" + + +@dataclass(frozen=True) +class Binding: + """One credential pinned to one KB, identified by its token's subject.""" + + subject: str + kb_id: str + kb_name: str = "" + agent: str = "" + bound_at: str = "" + + def to_dict(self) -> dict[str, Any]: + out: dict[str, Any] = {"subject": self.subject, "kb_id": self.kb_id} + if self.kb_name: + out["kb_name"] = self.kb_name + if self.agent: + out["agent"] = self.agent + if self.bound_at: + out["bound_at"] = self.bound_at + return out + + +def _parse_binding(raw: object) -> Binding | None: + """One row, or None if malformed — a bad row never breaks the rest.""" + if not isinstance(raw, dict): + return None + subject = raw.get("subject") + kb_id = raw.get("kb_id") + if not isinstance(subject, str) or not subject: + return None + if not isinstance(kb_id, str) or not kb_id: + return None + return Binding( + subject=subject, + kb_id=kb_id, + kb_name=str(raw.get("kb_name") or ""), + agent=str(raw.get("agent") or ""), + bound_at=str(raw.get("bound_at") or ""), + ) + + +def load_bindings(path: Path | None = None) -> list[Binding]: + """Read the binding file defensively: missing/corrupt -> no bindings. + + Degrading to "nothing is bound" is the only safe direction here — see the + module docstring. A corrupt file must not lock out every agent. + """ + p = path or bindings_path() + try: + loaded = yaml.safe_load(p.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return [] + if not isinstance(loaded, dict): + return [] + rows = loaded.get("bindings") + if not isinstance(rows, list): + return [] + out: list[Binding] = [] + seen: set[str] = set() + for raw in rows: + entry = _parse_binding(raw) + # First row wins on a duplicated subject: two answers to "which KB may + # this reach" is exactly the ambiguity the binding exists to remove. + if entry is not None and entry.subject not in seen: + seen.add(entry.subject) + out.append(entry) + return out + + +def save_bindings(bindings: list[Binding], path: Path | None = None) -> Path: + """Atomically write the binding file, owner-readable only.""" + p = path or bindings_path() + p.parent.mkdir(parents=True, exist_ok=True) + body = { + "version": BINDINGS_VERSION, + "bindings": [b.to_dict() for b in bindings], + } + fd, tmp_name = tempfile.mkstemp(dir=p.parent, prefix=p.name + ".") + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(yaml.safe_dump(body, sort_keys=False, allow_unicode=True)) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_name, p) + except BaseException: + Path(tmp_name).unlink(missing_ok=True) + raise + return p + + +def binding_for(subject: str, *, path: Path | None = None) -> Binding | None: + """The binding pinning `subject` to a KB, or None if it is unbound.""" + subject = subject.strip() + return next((b for b in load_bindings(path) if b.subject == subject), None) + + +def allows(subject: str, kb_id: str, *, path: Path | None = None) -> bool: + """Whether `subject` may authenticate against the KB with id `kb_id`. + + Unbound subjects are allowed everywhere (today's behaviour); a bound one + is allowed against its own KB and refused against every other. + """ + bound = binding_for(subject, path=path) + return bound is None or bound.kb_id == kb_id + + +def store_allows(store: KBStore, subject: str, *, path: Path | None = None) -> bool: + """`allows`, resolving the KB id from the store being served. + + A KB with no minted identity cannot be the target of any binding, so a + *bound* subject is refused there rather than falling through to it — the + binding would otherwise be escapable by pointing an agent at a KB whose + `kb.id` had been removed. + """ + identity = store.identity() + if identity is None: + return binding_for(subject, path=path) is None + return allows(subject, identity[0], path=path) + + +def bind( + *, + subject: str, + kb_id: str, + kb_name: str = "", + agent: str = "", + path: Path | None = None, +) -> Binding: + """Pin `subject` to `kb_id`. Refuses to move an existing binding. + + Re-binding in place would silently move a live credential between KBs, + which is the one transition this file exists to make impossible. Retiring + the old binding is an explicit `unbind`. + """ + subject = subject.strip() + if not subject: + raise BindingError("bind needs the token's auth subject") + if not kb_id: + raise BindingError("bind needs the target kb id") + entry = Binding( + subject=subject, + kb_id=kb_id, + kb_name=kb_name, + agent=agent, + bound_at=utcnow_iso(), + ) + p = path or bindings_path() + with _file_lock(p): + existing = load_bindings(p) + for b in existing: + if b.subject == subject and b.kb_id != kb_id: + raise BindingError( + f"subject {subject} is already bound to kb {b.kb_id}; " + "unbind it first" + ) + kept = [b for b in existing if b.subject != subject] + kept.append(entry) + save_bindings(kept, p) + return entry + + +def unbind(subject: str, *, path: Path | None = None) -> Binding | None: + """Drop `subject`'s binding. Returns the removed row, or None.""" + subject = subject.strip() + p = path or bindings_path() + with _file_lock(p): + existing = load_bindings(p) + removed = next((b for b in existing if b.subject == subject), None) + if removed is not None: + save_bindings([b for b in existing if b.subject != subject], p) + return removed + + +def bindings_for_kb(kb_id: str, *, path: Path | None = None) -> list[Binding]: + """Every credential pinned to one KB.""" + return [b for b in load_bindings(path) if b.kb_id == kb_id] + + +def issue_token() -> str: + """A fresh credential. Returned once and never persisted by vouch.""" + return secrets.token_urlsafe(TOKEN_BYTES) + + +def token_env_var(agent: str) -> str: + """The env var name a KB's accept-list references for `agent`'s token. + + `serve.bearer_tokens` supports an ``env:VAR`` indirection precisely so a + committed config never carries the secret; this picks the variable name so + the operator does not have to. + """ + slug = re.sub(r"[^A-Za-z0-9]+", "_", agent).strip("_").upper() + return f"VOUCH_TOKEN_{slug}" if slug else "VOUCH_TOKEN" diff --git a/tests/test_kb_binding.py b/tests/test_kb_binding.py new file mode 100644 index 00000000..b8c494d2 --- /dev/null +++ b/tests/test_kb_binding.py @@ -0,0 +1,525 @@ +"""Dedicated agent KBs and credentials bound to one KB — issue #609.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml +from click.testing import CliRunner + +from vouch import agents, audit, hub, kb_binding, trust +from vouch.cli import cli +from vouch.kb_binding import Binding, BindingError +from vouch.storage import KBStore + + +@pytest.fixture(autouse=True) +def _isolated_machine(tmp_path_factory, monkeypatch): + """Fake $HOME plus registry/binding paths so tests never touch the machine. + + Mirrors test_hub's isolator, and adds the binding file: an authorization + decision keyed on a real developer's ~/.config would be a nasty surprise. + """ + fake_home = tmp_path_factory.mktemp("home") + monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home)) + monkeypatch.setenv(hub.REGISTRY_ENV, str(fake_home / "registry.yaml")) + monkeypatch.setenv(kb_binding.BINDINGS_ENV, str(fake_home / "credentials.yaml")) + monkeypatch.setenv("XDG_DATA_HOME", str(fake_home / "data")) + for var in ("VOUCH_KB_PATH", "VOUCH_PROJECT_DIR", hub.PERSONAL_KB_ENV, + hub.AGENT_KBS_ENV, "XDG_CONFIG_HOME"): + monkeypatch.delenv(var, raising=False) + return fake_home + + +@pytest.fixture +def project(tmp_path: Path) -> KBStore: + """A human's project KB, registered the ordinary way.""" + root = tmp_path / "proj" + store = KBStore.init(root) + hub.register_kb(root, actor="human") + return store + + +def _create(*args: str): + return CliRunner().invoke(cli, ["kb", "create", *args]) + + +def _bound_subject() -> str: + return kb_binding.load_bindings()[0].subject + + +# --- the binding file ----------------------------------------------------- + + +def test_a_bound_subject_reaches_its_own_kb_and_no_other( + project: KBStore, tmp_path: Path +) -> None: + """The whole point: one credential, one KB.""" + agent_kb = KBStore.init(tmp_path / "agent") + kb_binding.bind(subject="abc123", kb_id=agent_kb.identity()[0], agent="ci") # type: ignore[index] + + assert kb_binding.store_allows(agent_kb, "abc123") is True + assert kb_binding.store_allows(project, "abc123") is False + + +def test_an_unbound_subject_is_unaffected(project: KBStore) -> None: + """Binding is opt-in: a token that predates it keeps working everywhere.""" + assert kb_binding.allows("never-bound", "any-kb-id") is True + assert kb_binding.store_allows(project, "never-bound") is True + + +def test_a_bound_subject_is_refused_by_a_kb_with_no_identity( + tmp_path: Path, +) -> None: + """Stripping `kb.id` must not become an escape hatch from the binding.""" + anonymous = KBStore.init(tmp_path / "anon") + kb_binding.bind(subject="abc123", kb_id="somewhere-else") + config = yaml.safe_load(anonymous.config_path.read_text(encoding="utf-8")) + config.pop("kb") + anonymous.config_path.write_text(yaml.safe_dump(config), encoding="utf-8") + + assert anonymous.identity() is None + assert kb_binding.store_allows(anonymous, "abc123") is False + # …while an unbound subject still authenticates against it as before. + assert kb_binding.store_allows(anonymous, "unbound") is True + + +def test_binding_file_is_owner_readable_and_never_holds_the_secret() -> None: + token = kb_binding.issue_token() + subject = trust.auth_subject_for_token(token) + kb_binding.bind(subject=subject, kb_id="kb-1", kb_name="ci", agent="bot") + + path = kb_binding.bindings_path() + raw = path.read_text(encoding="utf-8") + assert subject in raw + assert token not in raw + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_bind_refuses_to_move_a_live_credential() -> None: + kb_binding.bind(subject="s1", kb_id="kb-a") + with pytest.raises(BindingError, match="already bound to kb kb-a"): + kb_binding.bind(subject="s1", kb_id="kb-b") + # Re-binding to the same KB is a harmless refresh, not a move. + assert kb_binding.bind(subject="s1", kb_id="kb-a").kb_id == "kb-a" + assert len(kb_binding.load_bindings()) == 1 + + +def test_bind_needs_a_subject_and_a_kb_id() -> None: + with pytest.raises(BindingError, match="auth subject"): + kb_binding.bind(subject=" ", kb_id="kb-a") + with pytest.raises(BindingError, match="target kb id"): + kb_binding.bind(subject="s1", kb_id="") + + +def test_unbind_removes_one_row_and_is_quiet_about_the_rest() -> None: + kb_binding.bind(subject="s1", kb_id="kb-a") + kb_binding.bind(subject="s2", kb_id="kb-a") + + assert kb_binding.unbind("s1") is not None + assert kb_binding.unbind("s1") is None + assert [b.subject for b in kb_binding.load_bindings()] == ["s2"] + assert [b.subject for b in kb_binding.bindings_for_kb("kb-a")] == ["s2"] + assert kb_binding.bindings_for_kb("kb-z") == [] + + +def test_binding_for_finds_by_subject() -> None: + kb_binding.bind(subject="s1", kb_id="kb-a") + assert kb_binding.binding_for(" s1 ") is not None + assert kb_binding.binding_for("s9") is None + + +@pytest.mark.parametrize( + "body", + [ + "", # not a mapping + "bindings: not-a-list", + "bindings: [null, 3, {}, {subject: s}, {kb_id: k}, {subject: '', kb_id: k}]", + "{{{ not yaml", + ], +) +def test_an_unreadable_binding_file_denies_nobody(body: str) -> None: + """Fail-open on corruption: this file must never lock a fleet out.""" + kb_binding.bindings_path().parent.mkdir(parents=True, exist_ok=True) + kb_binding.bindings_path().write_text(body, encoding="utf-8") + assert kb_binding.load_bindings() == [] + assert kb_binding.allows("anyone", "any-kb") is True + + +def test_a_missing_binding_file_denies_nobody() -> None: + assert not kb_binding.bindings_path().exists() + assert kb_binding.load_bindings() == [] + + +def test_a_duplicated_subject_resolves_to_one_answer() -> None: + """Two answers to "which KB may this reach" is the ambiguity to remove.""" + kb_binding.bindings_path().parent.mkdir(parents=True, exist_ok=True) + kb_binding.bindings_path().write_text( + yaml.safe_dump( + { + "version": 1, + "bindings": [ + {"subject": "s1", "kb_id": "kb-a"}, + {"subject": "s1", "kb_id": "kb-b"}, + ], + } + ), + encoding="utf-8", + ) + assert [b.kb_id for b in kb_binding.load_bindings()] == ["kb-a"] + + +def test_a_failed_write_leaves_no_stray_temp_file( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A half-written credential file is worse than no write at all.""" + path = kb_binding.bindings_path() + path.parent.mkdir(parents=True, exist_ok=True) + monkeypatch.setattr( + kb_binding.yaml, "safe_dump", lambda *a, **k: (_ for _ in ()).throw(OSError("disk")) + ) + with pytest.raises(OSError): + kb_binding.save_bindings([Binding(subject="s", kb_id="k")]) + assert list(path.parent.iterdir()) == [] + + +def test_optional_binding_fields_are_omitted_when_empty() -> None: + assert Binding(subject="s", kb_id="k").to_dict() == {"subject": "s", "kb_id": "k"} + full = Binding( + subject="s", kb_id="k", kb_name="n", agent="a", bound_at="t" + ).to_dict() + assert full["kb_name"] == "n" and full["agent"] == "a" and full["bound_at"] == "t" + + +def test_bindings_path_prefers_env_then_xdg_then_home( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv(kb_binding.BINDINGS_ENV, str(tmp_path / "forced.yaml")) + assert kb_binding.bindings_path() == tmp_path / "forced.yaml" + monkeypatch.delenv(kb_binding.BINDINGS_ENV) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + assert kb_binding.bindings_path() == tmp_path / "xdg" / "vouch" / "credentials.yaml" + monkeypatch.delenv("XDG_CONFIG_HOME") + assert kb_binding.bindings_path() == Path.home() / ".config" / "vouch" / "credentials.yaml" + + +def test_issued_tokens_are_unguessable_and_distinct() -> None: + tokens = {kb_binding.issue_token() for _ in range(20)} + assert len(tokens) == 20 + assert all(len(t) >= 40 for t in tokens) + + +@pytest.mark.parametrize( + ("agent", "expected"), + [ + ("pr-triager", "VOUCH_TOKEN_PR_TRIAGER"), + ("incident.summariser", "VOUCH_TOKEN_INCIDENT_SUMMARISER"), + ("---", "VOUCH_TOKEN"), + ], +) +def test_token_env_var_is_shell_safe(agent: str, expected: str) -> None: + assert kb_binding.token_env_var(agent) == expected + + +# --- the transport chokepoint --------------------------------------------- + + +def test_the_gate_refuses_a_bound_token_against_the_project_kb( + project: KBStore, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The leaked-CI-secret case: right token, wrong working directory.""" + agent_kb = KBStore.init(tmp_path / "agent") + token = kb_binding.issue_token() + subject = trust.auth_subject_for_token(token) + kb_binding.bind(subject=subject, kb_id=agent_kb.identity()[0], agent="ci") # type: ignore[index] + + monkeypatch.chdir(agent_kb.root) + assert agents.subject_is_active(subject) is True + monkeypatch.chdir(project.root) + assert agents.subject_is_active(subject) is False + + +def test_the_gate_still_denies_a_revoked_agent_in_its_own_kb( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Binding is an extra gate, not a replacement for revocation.""" + agent_kb = KBStore.init(tmp_path / "agent") + subject = trust.auth_subject_for_token(kb_binding.issue_token()) + agents.register(agent_kb, subject=subject, name="ci", actor="human") + kb_binding.bind(subject=subject, kb_id=agent_kb.identity()[0]) # type: ignore[index] + agents.set_status(agent_kb, "ci", agents.AgentStatus.REVOKED, actor="human") + + monkeypatch.chdir(agent_kb.root) + assert agents.subject_is_active(subject) is False + + +def test_the_gate_is_unchanged_for_deployments_with_no_bindings( + project: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(project.root) + assert agents.subject_is_active(trust.auth_subject_for_token("legacy")) is True + + +# --- the registry gains an owner ------------------------------------------ + + +def test_ownership_round_trips_and_older_rows_read_as_human_owned( + tmp_path: Path, +) -> None: + root = tmp_path / "kb" + KBStore.init(root) + entry = hub.register_kb(root, actor="human", owner="alice-example", agent="ci-bot") + assert entry.machine_owned is True + + [reloaded] = hub.load_registry() + assert (reloaded.owner, reloaded.agent) == ("alice-example", "ci-bot") + assert hub.agent_entries() == [reloaded] + + plain = hub.RegistryEntry( + kb_id="k", name="n", role="project", path="/p", added_at="t" + ) + assert plain.machine_owned is False + assert "owner" not in hub._row(plain) and "agent" not in hub._row(plain) + + +def test_ownership_survives_a_plain_re_register(tmp_path: Path) -> None: + """`vouch hub register` must not launder an agent KB into a human one.""" + root = tmp_path / "kb" + KBStore.init(root) + first = hub.register_kb(root, actor="human", owner="alice-example", agent="ci-bot") + again = hub.register_kb(root, actor="human") + + assert (again.owner, again.agent) == ("alice-example", "ci-bot") + assert again.added_at == first.added_at + + +def test_ambient_capture_refuses_to_land_in_an_agent_kb(tmp_path: Path) -> None: + """An agent KB caught by upward discovery is the personal-KB hazard.""" + root = tmp_path / "outer" + KBStore.init(root) + hub.register_kb(root, actor="human", owner="alice-example", agent="ci-bot") + nested = root / "project" + nested.mkdir() + + res = hub.resolve(nested) + assert res.root == root.resolve() + assert res.guard is not None and "provisioned for agent 'ci-bot'" in res.guard + assert hub.resolve_for_capture(nested) is None + # From the KB's own root it is not ambient — it is the deliberate target. + assert hub.resolve(root).guard is None + + +def test_agent_kb_root_prefers_env_then_xdg_then_home( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv(hub.AGENT_KBS_ENV, str(tmp_path / "forced")) + assert hub.agent_kb_root("ci") == tmp_path / "forced" / "ci" + monkeypatch.delenv(hub.AGENT_KBS_ENV) + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg")) + assert hub.agent_kb_root("ci") == tmp_path / "xdg" / "vouch" / "kbs" / "ci" + monkeypatch.delenv("XDG_DATA_HOME") + assert hub.agent_kb_root("ci") == Path.home() / ".local/share/vouch/kbs/ci" + + +def test_agent_kb_root_sanitises_the_name_and_refuses_an_empty_one( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv(hub.AGENT_KBS_ENV, str(tmp_path)) + assert hub.agent_kb_root("../../etc/passwd") == tmp_path / "etc-passwd" + with pytest.raises(ValueError, match="cannot derive a directory name"): + hub.agent_kb_root("///") + + +def test_agent_kb_root_is_none_without_a_home(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("XDG_DATA_HOME", raising=False) + + def _no_home(cls): + raise RuntimeError("no home") + + monkeypatch.setattr(Path, "home", classmethod(_no_home)) + assert hub.agent_kb_root("ci") is None + + +# --- cli ------------------------------------------------------------------ + + +def test_cli_create_provisions_registers_and_issues_one_credential( + project: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(project.root) + res = _create("ci-triage", "--for-agent", "pr-triager") + assert res.exit_code == 0, res.output + + entry = hub.agent_entries()[0] + assert (entry.name, entry.agent) == ("ci-triage", "pr-triager") + agent_kb = KBStore(Path(entry.path)) + + # The token is echoed exactly once and never written to disk. + env_var = kb_binding.token_env_var("pr-triager") + token = next( + line.split("=", 1)[1].strip() + for line in res.output.splitlines() + if line.strip().startswith(env_var) + ) + assert trust.auth_subject_for_token(token) == _bound_subject() + assert token not in kb_binding.bindings_path().read_text(encoding="utf-8") + assert token not in (agent_kb.kb_dir / "agents.yaml").read_text(encoding="utf-8") + + # It authenticates against its own KB and not against the project. + assert kb_binding.store_allows(agent_kb, _bound_subject()) is True + assert kb_binding.store_allows(project, _bound_subject()) is False + + # The agent is named in its own KB, and the accept-list points at the + # env var rather than carrying the secret. + assert [a.name for a in agents.load_registry(agent_kb)] == ["pr-triager"] + config = yaml.safe_load(agent_kb.config_path.read_text(encoding="utf-8")) + assert config["serve"]["bearer_tokens"] == [f"env:{env_var}"] + assert any( + e.event == "agent.bind" + for e in audit.read_events(agent_kb.kb_dir) + ) + + +def test_cli_create_without_an_agent_is_just_a_registered_kb(tmp_path: Path) -> None: + res = _create("scratch", "--path", str(tmp_path / "scratch")) + assert res.exit_code == 0, res.output + assert "credential" not in res.output + assert kb_binding.load_bindings() == [] + [entry] = hub.load_registry() + assert entry.machine_owned is False and entry.owner + + +def test_cli_create_refuses_to_clobber_an_existing_kb(tmp_path: Path) -> None: + dest = tmp_path / "taken" + KBStore.init(dest) + res = _create("taken", "--path", str(dest)) + assert res.exit_code != 0 + assert "already exists" in res.output + assert "Traceback" not in res.output + + +def test_cli_create_reports_a_homeless_machine_cleanly( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("XDG_DATA_HOME", raising=False) + + def _no_home(cls): + raise RuntimeError("no home") + + monkeypatch.setattr(Path, "home", classmethod(_no_home)) + res = _create("ci") + assert res.exit_code != 0 + assert hub.AGENT_KBS_ENV in res.output + assert "Traceback" not in res.output + + +def test_cli_create_rejects_a_name_it_cannot_turn_into_a_directory() -> None: + res = _create("///") + assert res.exit_code != 0 + assert "cannot derive a directory name" in res.output + assert "Traceback" not in res.output + + +def test_cli_create_leaves_nothing_behind_when_init_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A half-built KB is one a rerun would mistake for a finished one.""" + def _boom(root, **kwargs): + (root / ".vouch").mkdir(parents=True, exist_ok=True) + raise OSError("read-only filesystem") + + monkeypatch.setattr("vouch.cli._bootstrap_kb", _boom) + dest = tmp_path / "doomed" + res = _create("doomed", "--path", str(dest)) + + assert res.exit_code != 0 + assert "could not initialise the KB" in res.output + assert "Traceback" not in res.output + assert not (dest / ".vouch").exists() + + +def test_cli_issue_rotates_a_credential_for_an_existing_kb() -> None: + assert _create("ci-triage", "--for-agent", "pr-triager").exit_code == 0 + first = _bound_subject() + + res = CliRunner().invoke( + cli, ["kb", "issue", "ci-triage", "--for-agent", "pr-triager-2"] + ) + assert res.exit_code == 0, res.output + subjects = {b.subject for b in kb_binding.load_bindings()} + assert first in subjects and len(subjects) == 2 + # Both are bound to the same KB, and the accept-list already names the + # first agent's var — so only the second one needs a nudge. + assert len({b.kb_id for b in kb_binding.load_bindings()}) == 1 + assert "add `env:VOUCH_TOKEN_PR_TRIAGER_2`" in res.output + + +def test_cli_issue_says_nothing_when_the_accept_list_already_names_the_var() -> None: + assert _create("ci-triage", "--for-agent", "bot").exit_code == 0 + res = CliRunner().invoke(cli, ["kb", "issue", "ci-triage", "--for-agent", "bot2"]) + assert res.exit_code == 0, res.output + assert "add `env:" in res.output + + entry = hub.agent_entries()[0] + store = KBStore(Path(entry.path)) + config = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) + config["serve"]["bearer_tokens"].append("env:VOUCH_TOKEN_BOT3") + store.config_path.write_text(yaml.safe_dump(config), encoding="utf-8") + + res = CliRunner().invoke(cli, ["kb", "issue", "ci-triage", "--for-agent", "bot3"]) + assert res.exit_code == 0, res.output + assert "add `env:" not in res.output + + +def test_cli_issue_errors_cleanly_on_an_unknown_or_moved_kb(tmp_path: Path) -> None: + runner = CliRunner() + res = runner.invoke(cli, ["kb", "issue", "nope", "--for-agent", "bot"]) + assert res.exit_code != 0 + assert "no agent KB called 'nope'" in res.output + + assert _create("gone", "--for-agent", "bot", "--path", str(tmp_path / "gone")).exit_code == 0 + import shutil as _shutil + + _shutil.rmtree(tmp_path / "gone") + res = runner.invoke(cli, ["kb", "issue", "gone", "--for-agent", "bot"]) + assert res.exit_code != 0 + assert "there is no .vouch/ there" in res.output + assert "Traceback" not in res.output + + +def test_cli_issue_refuses_a_duplicate_agent_name() -> None: + assert _create("ci-triage", "--for-agent", "bot").exit_code == 0 + res = CliRunner().invoke(cli, ["kb", "issue", "ci-triage", "--for-agent", "bot"]) + assert res.exit_code != 0 + assert "Traceback" not in res.output + + +def test_cli_list_reports_agent_kbs_only() -> None: + runner = CliRunner() + empty = runner.invoke(cli, ["kb", "list"]) + assert empty.exit_code == 0 and "no agent KBs" in empty.output + + assert _create("ci-triage", "--for-agent", "pr-triager").exit_code == 0 + table = runner.invoke(cli, ["kb", "list"]) + assert table.exit_code == 0, table.output + assert "ci-triage" in table.output and "agent=pr-triager" in table.output + assert "credentials=1" in table.output + + res = runner.invoke(cli, ["kb", "list", "--json"]) + assert res.exit_code == 0, res.output + payload = json.loads(res.output) + [row] = payload["kbs"] + assert row["agent"] == "pr-triager" + assert row["credentials"] == 1 + assert row["owner"] + + +def test_cli_list_shows_a_dash_for_an_unrecorded_owner(tmp_path: Path) -> None: + root = tmp_path / "kb" + KBStore.init(root) + hub.register_kb(root, actor="human", agent="ci-bot") + res = CliRunner().invoke(cli, ["kb", "list"]) + assert res.exit_code == 0, res.output + assert "owner=-" in res.output