From 1aebe40716346bc0a16cc025b509885e0b691e30 Mon Sep 17 00:00:00 2001 From: minion1227 Date: Thu, 30 Jul 2026 13:10:07 -0700 Subject: [PATCH 1/3] =?UTF-8?q?feat(trust):=20scoped=20tokens=20=E2=80=94?= =?UTF-8?q?=20kb:read=20/=20kb:propose=20/=20kb:approve=20/=20kb:admin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit a bearer token was all-or-nothing: hold it and you could call all 73 kb.* methods, kb.approve included. fine for a solo human, wrong for a ci job that should only read or a triage bot that should only propose. the config-level trusted-agent flag can only widen the gate; this is the first thing in vouch that can narrow it — withholding kb:approve by default *is* the review gate, expressed as a credential. four coarse scopes over the method list rather than a per-method allowlist, because a per-method grammar makes every new kb.* method a config migration for every deployment. two rules keep this safe to ship into existing deployments. an unscoped credential means all scopes, so every token issued before this keeps working exactly as it did — an empty scope set is "unrestricted", never "denied". and every method must be classified: METHOD_SCOPES is exhaustive over capabilities.METHODS with a test enforcing it, so a newly-added method cannot silently land unreachable for scoped callers. an unclassified method is denied to a scoped caller — fails closed, because a deny-list in a trust-centric system fails open. enforcement lives at the two dispatch points that already exist: handle_request for jsonl and http, and wrap_tool_fn for mcp, both routed through trust.require_scope so the three surfaces inherit one implementation. the check runs before the handler, so a refused call cannot have side effects on its way to being refused. the agent registry supplies the scopes: a registered subject's scopes ride onto VouchTrust at the http chokepoint, and an unregistered subject stays unscoped. registration now validates scopes, so a typo cannot mint a credential with powers nobody asked for. kb.capabilities reports the caller's effective scopes and allowed methods, so an agent discovers what it may do instead of failing method by method. the trust block only grows a scopes key when the credential is actually scoped. stacked on #607: the registry is where scopes are stored. Closes #608 --- src/vouch/agents.py | 32 +++ src/vouch/capabilities.py | 3 + src/vouch/cli.py | 3 +- src/vouch/http_server.py | 14 +- src/vouch/jsonl_server.py | 9 + src/vouch/models.py | 4 + src/vouch/scopes.py | 211 +++++++++++++++++++ src/vouch/trust.py | 57 +++++- tests/test_agents.py | 13 +- tests/test_scopes.py | 416 ++++++++++++++++++++++++++++++++++++++ 10 files changed, 750 insertions(+), 12 deletions(-) create mode 100644 src/vouch/scopes.py create mode 100644 tests/test_scopes.py diff --git a/src/vouch/agents.py b/src/vouch/agents.py index c0af370d..1b208393 100644 --- a/src/vouch/agents.py +++ b/src/vouch/agents.py @@ -41,6 +41,7 @@ import yaml from . import audit as audit_mod +from . import scopes as scopes_mod if TYPE_CHECKING: # pragma: no cover - typing only from .storage import KBStore @@ -190,6 +191,10 @@ def register( raise AgentError("register needs the token's auth subject") if not name: raise AgentError("register needs a name") + try: + scopes = scopes_mod.parse_scopes(scopes) + except scopes_mod.ScopeError as e: + raise AgentError(str(e)) from e agents = load_registry(store) for existing in agents: @@ -272,6 +277,33 @@ def is_active(store: KBStore, subject: str) -> bool: return agent is None or agent.status is AgentStatus.ACTIVE +def scopes_for_subject(store: KBStore, subject: str) -> tuple[str, ...]: + """The scopes a registered subject holds; empty (= unscoped) if unknown. + + Unknown subjects stay unscoped so a token that predates the registry keeps + every power it had — the same back-compat rule `is_active` follows. + """ + agent = next( + (a for a in load_registry(store) if a.subject == subject.strip()), None + ) + return agent.scopes if agent is not None else () + + +def subject_scopes(subject: str) -> tuple[str, ...]: + """Store-resolving scope lookup for the transport chokepoint.""" + from .storage import KBStore, discover_root + + try: + store = KBStore(discover_root()) + except Exception: + return () + try: + return scopes_for_subject(store, subject) + except Exception: # pragma: no cover - defensive + logger.debug("agents: registry unreadable, treating subject as unscoped") + return () + + def subject_is_active(subject: str) -> bool: """Store-resolving gate for the transport chokepoint. diff --git a/src/vouch/capabilities.py b/src/vouch/capabilities.py index 5fecb9dc..858a0d50 100644 --- a/src/vouch/capabilities.py +++ b/src/vouch/capabilities.py @@ -13,6 +13,8 @@ from . import __version__ from . import hot_memory as hot_mod +from . import scopes as scopes_mod +from . import trust as trust_mod from .models import Capabilities from .openclaw.context_engine import describe_engine @@ -152,6 +154,7 @@ def capabilities(*, publish_skills: bool = True) -> Capabilities: context_engines=[describe_engine()], mcp={"publish_skills": publish_skills}, host_compat=_load_host_compat(), + scopes=scopes_mod.describe(trust_mod.current().scopes), hot_memory={ "sidebar_key": "vouch_hot_memory", "list_envelope": True, diff --git a/src/vouch/cli.py b/src/vouch/cli.py index ccabbc9a..1663b260 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -3772,7 +3772,8 @@ def agents_group() -> None: @click.option("--subject", required=True, help="The token's auth subject (vouch agents subject ).") @click.option("--scope", "scopes", multiple=True, - help="Advisory scope to record (repeatable).") + help="Grant a scope (repeatable): kb:read, kb:propose, " + "kb:approve, kb:admin. No --scope means unscoped (all).") @click.option("--note", default=None, help="What this agent is for.") def agents_register(name: str, subject: str, scopes: tuple[str, ...], note: str | None) -> None: diff --git a/src/vouch/http_server.py b/src/vouch/http_server.py index 6baf1527..fbd24a00 100644 --- a/src/vouch/http_server.py +++ b/src/vouch/http_server.py @@ -210,7 +210,12 @@ async def _rpc(request: Request) -> JSONResponse: tuple(getattr(request.app.state, "vouch_bearer_tokens", ()) or ()), gate=agents_mod.subject_is_active, ) - trust = trust_mod.with_auth_subject(trust_mod.JSONL_HTTP, bearer) + trust = trust_mod.with_auth_subject( + trust_mod.JSONL_HTTP, bearer, + scopes=agents_mod.subject_scopes( + trust_mod.auth_subject_for_token(bearer) + ) if bearer else (), + ) def _dispatch() -> dict[str, Any]: # The actor and the trust marker are both ContextVars. They're set @@ -293,7 +298,12 @@ async def __call__(self, scope: dict, receive: Receive, send: Send) -> None: self._accepted, gate=agents_mod.subject_is_active, ) - trust = trust_mod.with_auth_subject(trust_mod.MCP_HTTP, bearer) + trust = trust_mod.with_auth_subject( + trust_mod.MCP_HTTP, bearer, + scopes=agents_mod.subject_scopes( + trust_mod.auth_subject_for_token(bearer) + ) if bearer else (), + ) token = trust_mod.set_trust_context(trust) try: await self._app(scope, receive, send) diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index 868ed906..da3a469b 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -1011,12 +1011,21 @@ def handle_request(envelope: dict) -> dict: "error": {"code": "method_not_found", "message": f"unknown method: {method}"}, } try: + # Scope check before dispatch: a credential that may not call this + # must not reach the handler, so a denied call cannot have side + # effects on the way to being refused. + trust_mod.require_scope(method) result = HANDLERS[method](params) return { "id": req_id, "ok": True, "result": trust_mod.finish_kb_result(result), } + except trust_mod.ScopeDenied as e: + return { + "id": req_id, "ok": False, + "error": {"code": "permission_denied", "message": str(e)}, + } except skills_mod.SkillsDisabledError as e: return { "id": req_id, "ok": False, diff --git a/src/vouch/models.py b/src/vouch/models.py index 99bf27e6..472e3fc8 100644 --- a/src/vouch/models.py +++ b/src/vouch/models.py @@ -521,6 +521,10 @@ class Capabilities(BaseModel): "audit_log": True, } ) + # Per-credential scopes (#608). Additive: an unscoped caller gets + # `unscoped: true` and the full method list, which is what every + # pre-#608 deployment sees. + scopes: dict[str, Any] = Field(default_factory=dict) scoping: dict[str, Any] = Field( default_factory=lambda: { "enabled": True, diff --git a/src/vouch/scopes.py b/src/vouch/scopes.py new file mode 100644 index 00000000..78a98c60 --- /dev/null +++ b/src/vouch/scopes.py @@ -0,0 +1,211 @@ +"""Scoped credentials — kb:read / kb:propose / kb:approve / kb:admin (#608). + +A bearer token used to be all-or-nothing: hold it and you could call every +``kb.*`` method, ``kb.approve`` included. That is fine for a solo human and +wrong for anything else — a CI job that should only read, a triage bot that +should only propose, an agent that should never approve its own work. + +Withholding ``kb:approve`` *is* the review gate, expressed as a credential. +The config-level ``trusted-agent`` flag can only widen the gate; this is the +first thing in vouch that can narrow it. + +Four coarse scopes over the method list, not a per-method allowlist. A +per-method grammar is more flexible and makes every new ``kb.*`` method a +config migration for every deployment; four buckets keep that cost at zero. + +* ``kb:read`` — search, context, read_*, list_*, and every other analysis + that cannot change durable state. +* ``kb:propose`` — register_source, propose_*, cite, session_*. Everything + that files work *into* the review queue. +* ``kb:approve`` — approve/reject and the lifecycle verbs. Deciding what the + KB believes. +* ``kb:admin`` — destructive or index-wide maintenance: clear_claims, + wipe_dead_refs, index_rebuild, provenance_rebuild. + +Two rules make this safe to ship into existing deployments: + +**An unscoped credential means all scopes.** Every token issued before this +existed keeps working exactly as it did. Scoping is opt-in, and an empty scope +set is "unrestricted", never "denied" — the alternative would break every +deployment on upgrade. + +**Every method must be classified.** ``METHOD_SCOPES`` is exhaustive over +``capabilities.METHODS`` and ``test_every_method_is_classified`` enforces it, +so a newly-added method cannot silently land in no scope (unreachable for +scoped callers) or in all of them (a hole). Unknown methods deny by default — +fails closed, because a deny-list in a trust-centric system fails open. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover - typing only + from collections.abc import Iterable + +READ = "kb:read" +PROPOSE = "kb:propose" +APPROVE = "kb:approve" +ADMIN = "kb:admin" + +ALL_SCOPES: tuple[str, ...] = (READ, PROPOSE, APPROVE, ADMIN) + +# The useful agent default, and the one where the gate holds: an agent may +# read the KB and file proposals, but cannot decide what the KB believes. +DEFAULT_SCOPES: tuple[str, ...] = (READ, PROPOSE) + + +class ScopeError(ValueError): + """An unknown or malformed scope.""" + + +# Exhaustive over capabilities.METHODS — pinned by a test. +METHOD_SCOPES: dict[str, str] = { + # --- kb:read — analysis that cannot change durable state --------------- + "kb.capabilities": READ, + "kb.status": READ, + "kb.stats": READ, + "kb.activity": READ, + "kb.digest": READ, + "kb.search": READ, + "kb.explain_ranking": READ, + "kb.neighbors": READ, + "kb.experts": READ, + "kb.context": READ, + "kb.synthesize": READ, + "kb.read_page": READ, + "kb.read_claim": READ, + "kb.read_entity": READ, + "kb.read_relation": READ, + "kb.read_evidence": READ, + "kb.read_source": READ, + "kb.diff": READ, + "kb.list_pages": READ, + "kb.list_claims": READ, + "kb.list_entities": READ, + "kb.list_relations": READ, + "kb.list_sources": READ, + "kb.list_pending": READ, + "kb.triage_pending": READ, + "kb.list_sessions": READ, + "kb.session_transcript": READ, + "kb.volunteer_context": READ, + "kb.lint": READ, + "kb.doctor": READ, + "kb.export": READ, + "kb.export_check": READ, + "kb.import_check": READ, + "kb.audit": READ, + "kb.dedup_scan": READ, + "kb.eval_embeddings": READ, + "kb.effectiveness": READ, + "kb.embeddings_stats": READ, + "kb.why": READ, + "kb.trace": READ, + "kb.impact": READ, + "kb.graph_export": READ, + "kb.detect_themes": READ, + "kb.list_skills": READ, + "kb.get_skill": READ, + # --- kb:propose — files work into the review queue --------------------- + "kb.register_source": PROPOSE, + "kb.register_source_from_path": PROPOSE, + "kb.propose_claim": PROPOSE, + "kb.propose_page": PROPOSE, + "kb.propose_entity": PROPOSE, + "kb.propose_relation": PROPOSE, + "kb.propose_delete": PROPOSE, + "kb.propose_theme": PROPOSE, + "kb.cite": PROPOSE, + "kb.source_verify": PROPOSE, + "kb.session_start": PROPOSE, + "kb.session_end": PROPOSE, + # crystallize/summarize_session/compile all file page proposals and + # nothing else — they are propose-path, not approve-path. + "kb.crystallize": PROPOSE, + "kb.summarize_session": PROPOSE, + "kb.compile": PROPOSE, + # --- kb:approve — deciding what the KB believes ------------------------ + "kb.approve": APPROVE, + "kb.reject": APPROVE, + "kb.reject_extracted": APPROVE, + "kb.expire": APPROVE, + "kb.supersede": APPROVE, + "kb.contradict": APPROVE, + "kb.archive": APPROVE, + "kb.confirm": APPROVE, + # --- kb:admin — destructive or index-wide maintenance ------------------ + "kb.clear_claims": ADMIN, + "kb.wipe_dead_refs": ADMIN, + "kb.index_rebuild": ADMIN, + "kb.reindex_embeddings": ADMIN, + "kb.provenance_rebuild": ADMIN, +} + + +def parse_scopes(raw: str | Iterable[str] | None) -> tuple[str, ...]: + """Normalise a scope spec. ``None``/empty means unscoped (all scopes). + + Accepts a comma-separated string or any iterable. Raises + :class:`ScopeError` on an unknown scope rather than dropping it — a typo + in ``--scopes`` must not silently produce a credential with fewer (or + more) powers than the operator asked for. + """ + if raw is None: + return () + items = raw.split(",") if isinstance(raw, str) else list(raw) + out: list[str] = [] + for item in items: + scope = str(item).strip() + if not scope: + continue + if scope not in ALL_SCOPES: + raise ScopeError( + f"unknown scope {scope!r}; valid scopes are {', '.join(ALL_SCOPES)}" + ) + if scope not in out: + out.append(scope) + return tuple(out) + + +def scope_for_method(method: str) -> str | None: + """The scope a method needs, or ``None`` when it is not classified.""" + return METHOD_SCOPES.get(method) + + +def permits(granted: Iterable[str] | None, method: str) -> bool: + """Whether a credential holding ``granted`` may call ``method``. + + Unscoped (empty/None) grants everything — the back-compat rule. A + classified method needs its scope; an *unclassified* one is denied for a + scoped caller, so forgetting the table entry costs a scoped agent a method + rather than silently handing it out. + """ + scopes = tuple(granted or ()) + if not scopes: + return True + required = scope_for_method(method) + if required is None: + return False + return required in scopes + + +def methods_for(granted: Iterable[str] | None) -> list[str]: + """Every method a credential holding ``granted`` may call, sorted. + + Backs the ``kb.capabilities`` scope report so an agent can discover what + it may do up front instead of failing method by method. + """ + return sorted(m for m in METHOD_SCOPES if permits(granted, m)) + + +def describe(granted: Iterable[str] | None) -> dict[str, object]: + """The ``kb.capabilities`` scope block for the calling credential.""" + scopes = tuple(granted or ()) + return { + "available": list(ALL_SCOPES), + "default": list(DEFAULT_SCOPES), + "granted": list(scopes), + "unscoped": not scopes, + "allowed_methods": methods_for(scopes), + } diff --git a/src/vouch/trust.py b/src/vouch/trust.py index 188fcfc8..215af92a 100644 --- a/src/vouch/trust.py +++ b/src/vouch/trust.py @@ -82,13 +82,26 @@ class VouchTrust: remote: bool caller_kind: CallerKind auth_subject: str | None + # Empty means unscoped, i.e. every scope — the back-compat rule that keeps + # tokens issued before #608 working exactly as they did. + scopes: tuple[str, ...] = () def as_meta_block(self) -> dict[str, Any]: - return { + block: dict[str, Any] = { "remote": self.remote, "caller_kind": self.caller_kind, "auth_subject": self.auth_subject, } + # Only surfaced when the credential is actually scoped, so unscoped + # callers see the same block they saw before. + if self.scopes: + block["scopes"] = list(self.scopes) + return block + + def permits(self, method: str) -> bool: + from .scopes import permits as _permits + + return _permits(self.scopes, method) # Presets — one per transport entry point. @@ -170,13 +183,42 @@ def authorized_bearer_token( return token if gate(auth_subject_for_token(token)) else None -def with_auth_subject(trust: VouchTrust, token: str | None) -> VouchTrust: +def with_auth_subject( + trust: VouchTrust, token: str | None, *, scopes: tuple[str, ...] = () +) -> VouchTrust: if token is None: return trust return VouchTrust( remote=trust.remote, caller_kind=trust.caller_kind, auth_subject=auth_subject_for_token(token), + scopes=scopes, + ) + + +class ScopeDenied(PermissionError): + """The active credential's scopes do not cover this method.""" + + +def require_scope(method: str) -> None: + """Raise :class:`ScopeDenied` when the active credential may not call it. + + One check, called from both dispatch points, so MCP / JSONL / HTTP inherit + scoping from the same place rather than three near-identical guards. + """ + from .scopes import scope_for_method + + trust = current() + if trust.permits(method): + return + required = scope_for_method(method) + detail = ( + f"requires {required}" if required + else "is not classified under any scope" + ) + raise ScopeDenied( + f"{method} {detail}; this credential holds " + f"{', '.join(trust.scopes) or 'no scopes'}" ) @@ -196,11 +238,19 @@ def finish_kb_result(result: Any) -> Any: _F = TypeVar("_F", bound=Callable[..., Any]) +def method_name_for_tool(tool_name: str) -> str: + """``kb_read_page`` -> ``kb.read_page`` — the MCP naming convention.""" + return tool_name.replace("_", ".", 1) + + def wrap_tool_fn(fn: _F) -> _F: - """Wrap a sync or async MCP tool so dict results carry ``_meta.vouch_trust``.""" + """Wrap a sync or async MCP tool: scope check in, trust metadata out.""" + method = method_name_for_tool(fn.__name__) + if inspect.iscoroutinefunction(fn): async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + require_scope(method) return finish_kb_result(await fn(*args, **kwargs)) async_wrapper.__name__ = fn.__name__ @@ -208,6 +258,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any: return async_wrapper # type: ignore[return-value] def sync_wrapper(*args: Any, **kwargs: Any) -> Any: + require_scope(method) return finish_kb_result(fn(*args, **kwargs)) sync_wrapper.__name__ = fn.__name__ diff --git a/tests/test_agents.py b/tests/test_agents.py index ba1d9b8d..f6a713d6 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -12,6 +12,7 @@ from vouch import agents, audit, trust from vouch.agents import AgentError, AgentStatus from vouch.cli import cli +from vouch.scopes import PROPOSE, READ from vouch.storage import KBStore TOKEN = "s3cret-token-example" @@ -53,11 +54,11 @@ def test_register_names_a_subject_without_storing_the_token( def test_registry_round_trips(store: KBStore, subject: str) -> None: agents.register( store, subject=subject, name="ci-bot", actor="human", - scopes=("read", "propose"), note="the CI proposer", + scopes=(READ, PROPOSE), note="the CI proposer", ) loaded = agents.load_registry(store) assert len(loaded) == 1 - assert loaded[0].scopes == ("read", "propose") + assert loaded[0].scopes == (READ, PROPOSE) assert loaded[0].note == "the CI proposer" @@ -150,11 +151,11 @@ def test_every_transition_is_audited( def test_registration_is_audited(store: KBStore, subject: str) -> None: agents.register( - store, subject=subject, name="ci-bot", actor="human", scopes=("propose",) + store, subject=subject, name="ci-bot", actor="human", scopes=(PROPOSE,) ) ev = next(e for e in audit.read_events(store.kb_dir) if e.event == "agent.register") assert ev.data["name"] == "ci-bot" - assert ev.data["scopes"] == ["propose"] + assert ev.data["scopes"] == [PROPOSE] # --- the authentication gate --------------------------------------------- @@ -349,14 +350,14 @@ def test_cli_register_list_show_roundtrip(store: KBStore, subject: str) -> None: assert "no registered agents" in empty.output reg = runner.invoke(cli, [ - "agents", "register", "ci-bot", "--subject", subject, "--scope", "propose", + "agents", "register", "ci-bot", "--subject", subject, "--scope", PROPOSE, ]) assert reg.exit_code == 0, reg.output listed = runner.invoke(cli, ["agents", "list"]) assert "ci-bot" in listed.output assert "active" in listed.output - assert "propose" in listed.output + assert PROPOSE in listed.output shown = runner.invoke(cli, ["agents", "show", "ci-bot"]) assert shown.exit_code == 0, shown.output diff --git a/tests/test_scopes.py b/tests/test_scopes.py new file mode 100644 index 00000000..1385983b --- /dev/null +++ b/tests/test_scopes.py @@ -0,0 +1,416 @@ +"""Scoped credentials — issue #608.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from vouch import agents, jsonl_server, scopes, trust +from vouch.capabilities import METHODS, capabilities +from vouch.scopes import ADMIN, APPROVE, PROPOSE, READ, ScopeError +from vouch.storage import KBStore + + +@pytest.fixture +def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> KBStore: + kb = KBStore.init(tmp_path) + monkeypatch.chdir(kb.root) + return kb + + +def _trust(*granted: str) -> trust.VouchTrust: + return trust.VouchTrust( + remote=True, caller_kind="jsonl_http", auth_subject="abc", + scopes=tuple(granted), + ) + + +def _call(method: str, params: dict | None = None) -> dict: + return jsonl_server.handle_request( + {"id": "1", "method": method, "params": params or {}} + ) + + +def _code(method: str, params: dict | None = None) -> str: + res = _call(method, params) + return "ok" if res["ok"] else res["error"]["code"] + + +# --- the classification table -------------------------------------------- + + +def test_every_method_is_classified() -> None: + """The guard the issue asks for. + + Without this a newly-added kb.* method silently lands in no scope — + unreachable for every scoped caller — or, if the default were permissive, + in all of them. + """ + unclassified = [m for m in METHODS if m not in scopes.METHOD_SCOPES] + assert not unclassified, f"methods with no scope: {unclassified}" + + +def test_the_table_has_no_entries_for_methods_that_do_not_exist() -> None: + stale = [m for m in scopes.METHOD_SCOPES if m not in METHODS] + assert not stale, f"scope entries for unknown methods: {stale}" + + +def test_every_scope_is_a_known_scope() -> None: + assert set(scopes.METHOD_SCOPES.values()) <= set(scopes.ALL_SCOPES) + + +@pytest.mark.parametrize( + ("method", "expected"), + [ + ("kb.search", READ), + ("kb.context", READ), + ("kb.audit", READ), + ("kb.propose_claim", PROPOSE), + ("kb.register_source", PROPOSE), + ("kb.compile", PROPOSE), + ("kb.approve", APPROVE), + ("kb.supersede", APPROVE), + ("kb.archive", APPROVE), + ("kb.clear_claims", ADMIN), + ("kb.index_rebuild", ADMIN), + ], +) +def test_representative_methods_land_in_the_right_bucket( + method: str, expected: str +) -> None: + assert scopes.scope_for_method(method) == expected + + +def test_approve_is_not_reachable_from_the_default_scopes() -> None: + """Withholding kb:approve by default *is* the review gate.""" + assert scopes.DEFAULT_SCOPES == (READ, PROPOSE) + assert scopes.permits(scopes.DEFAULT_SCOPES, "kb.approve") is False + assert scopes.permits(scopes.DEFAULT_SCOPES, "kb.propose_claim") is True + + +# --- permits -------------------------------------------------------------- + + +def test_unscoped_permits_everything(store: KBStore) -> None: + """The back-compat rule: a pre-#608 token keeps every power it had.""" + for method in METHODS: + assert scopes.permits((), method) is True + assert scopes.permits(None, method) is True + + +def test_an_unclassified_method_is_denied_to_a_scoped_caller() -> None: + """Fails closed: forgetting a table entry costs a method, not the gate.""" + assert scopes.permits((READ,), "kb.brand_new_method") is False + assert scopes.permits((), "kb.brand_new_method") is True + + +def test_methods_for_lists_only_what_is_allowed() -> None: + read_only = scopes.methods_for((READ,)) + assert "kb.search" in read_only + assert "kb.approve" not in read_only + assert "kb.propose_claim" not in read_only + assert len(scopes.methods_for(())) == len(METHODS) + + +def test_describe_reports_the_caller_contract() -> None: + block = scopes.describe((READ,)) + assert block["available"] == list(scopes.ALL_SCOPES) + assert block["granted"] == [READ] + assert block["unscoped"] is False + assert "kb.approve" not in block["allowed_methods"] + + unscoped = scopes.describe(()) + assert unscoped["unscoped"] is True + + +# --- parse_scopes --------------------------------------------------------- + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (None, ()), + ("", ()), + ("kb:read", (READ,)), + ("kb:read,kb:propose", (READ, PROPOSE)), + (" kb:read , kb:propose ", (READ, PROPOSE)), + ("kb:read,kb:read", (READ,)), + (["kb:read", "kb:approve"], (READ, APPROVE)), + ], +) +def test_parse_scopes_normalises(raw: object, expected: tuple[str, ...]) -> None: + assert scopes.parse_scopes(raw) == expected # type: ignore[arg-type] + + +def test_parse_scopes_rejects_an_unknown_scope() -> None: + """A typo must not silently mint a credential with the wrong powers.""" + with pytest.raises(ScopeError, match="unknown scope"): + scopes.parse_scopes("kb:read,kb:destroy") + + +# --- enforcement at the jsonl / http dispatch ---------------------------- + + +def test_read_only_credential_is_denied_the_write_paths(store: KBStore) -> None: + with trust.trust_context(_trust(READ)): + assert _code("kb.status") == "ok" + assert _code("kb.propose_claim", {"text": "x", "evidence": []}) == ( + "permission_denied" + ) + assert _code("kb.approve", {"proposal_id": "x", "approved_by": "y"}) == ( + "permission_denied" + ) + assert _code("kb.clear_claims") == "permission_denied" + + +def test_default_credential_may_propose_but_not_approve(store: KBStore) -> None: + with trust.trust_context(_trust(*scopes.DEFAULT_SCOPES)): + # reaches the handler (its own validation error), i.e. scope allowed it + assert _code("kb.propose_claim", {"text": "x", "evidence": []}) != ( + "permission_denied" + ) + assert _code("kb.approve", {"proposal_id": "x", "approved_by": "y"}) == ( + "permission_denied" + ) + + +def test_unscoped_credential_reaches_every_handler(store: KBStore) -> None: + with trust.trust_context(_trust()): + assert _code("kb.status") == "ok" + assert _code("kb.clear_claims") == "ok" + assert _code("kb.approve", {"proposal_id": "x", "approved_by": "y"}) != ( + "permission_denied" + ) + + +def test_denial_names_the_scope_it_needed(store: KBStore) -> None: + with trust.trust_context(_trust(READ)): + res = _call("kb.approve", {"proposal_id": "x", "approved_by": "y"}) + assert res["error"]["message"] == ( + "kb.approve requires kb:approve; this credential holds kb:read" + ) + + +def test_denial_happens_before_the_handler_runs( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """A refused call must not have side effects on its way to being refused.""" + called: list[str] = [] + real = jsonl_server.HANDLERS["kb.clear_claims"] + + def spy(params: dict) -> dict: + called.append("ran") + return real(params) + + monkeypatch.setitem(jsonl_server.HANDLERS, "kb.clear_claims", spy) + with trust.trust_context(_trust(READ)): + assert _code("kb.clear_claims") == "permission_denied" + assert called == [] + + +def test_unknown_method_still_reports_method_not_found(store: KBStore) -> None: + """Scope checking must not turn a typo into a permission error.""" + with trust.trust_context(_trust(READ)): + assert _code("kb.no_such_method") == "method_not_found" + + +# --- enforcement on the mcp surface -------------------------------------- + + +def test_tool_name_maps_to_its_method() -> None: + assert trust.method_name_for_tool("kb_search") == "kb.search" + assert trust.method_name_for_tool("kb_read_page") == "kb.read_page" + assert trust.method_name_for_tool("kb_explain_ranking") == "kb.explain_ranking" + + +def test_wrapped_mcp_tool_enforces_scope(store: KBStore) -> None: + def kb_approve(**kwargs: object) -> dict: + return {"approved": True} + + wrapped = trust.wrap_tool_fn(kb_approve) + with trust.trust_context(_trust(READ)), pytest.raises( + trust.ScopeDenied, match="requires kb:approve" + ): + wrapped() + with trust.trust_context(_trust(APPROVE)): + assert wrapped()["approved"] is True + with trust.trust_context(_trust()): + assert wrapped()["approved"] is True + + +def test_wrapped_async_mcp_tool_enforces_scope(store: KBStore) -> None: + """The async branch needs the same guard as the sync one.""" + import asyncio + + async def kb_approve(**kwargs: object) -> dict: + return {"approved": True} + + wrapped = trust.wrap_tool_fn(kb_approve) + with trust.trust_context(_trust(READ)), pytest.raises(trust.ScopeDenied): + asyncio.run(wrapped()) + with trust.trust_context(_trust(APPROVE)): + assert asyncio.run(wrapped())["approved"] is True + + +# --- the trust block + capabilities -------------------------------------- + + +def test_scopes_appear_in_the_trust_block_only_when_scoped() -> None: + assert "scopes" not in _trust().as_meta_block() + assert _trust(READ).as_meta_block()["scopes"] == [READ] + + +def test_capabilities_reports_the_callers_effective_scopes(store: KBStore) -> None: + with trust.trust_context(_trust(READ)): + caps = capabilities().model_dump(mode="json") + assert caps["scopes"]["granted"] == [READ] + assert caps["scopes"]["unscoped"] is False + assert "kb.approve" not in caps["scopes"]["allowed_methods"] + + +def test_capabilities_for_an_unscoped_caller_lists_everything( + store: KBStore +) -> None: + with trust.trust_context(_trust()): + caps = capabilities().model_dump(mode="json") + assert caps["scopes"]["unscoped"] is True + assert len(caps["scopes"]["allowed_methods"]) == len(METHODS) + + +# --- the registry supplies the scopes ------------------------------------ + + +def test_registered_scopes_are_resolved_for_a_subject(store: KBStore) -> None: + agents.register( + store, subject="abc123", name="ci-bot", actor="human", + scopes=(READ,), + ) + assert agents.scopes_for_subject(store, "abc123") == (READ,) + + +def test_an_unregistered_subject_is_unscoped(store: KBStore) -> None: + """Back-compat again: an unknown token keeps every power.""" + assert agents.scopes_for_subject(store, "never-registered") == () + + +def test_registering_an_unknown_scope_is_refused(store: KBStore) -> None: + with pytest.raises(agents.AgentError, match="unknown scope"): + agents.register( + store, subject="abc123", name="ci-bot", actor="human", + scopes=("kb:destroy",), + ) + + +def test_subject_scopes_is_unscoped_without_a_kb( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.chdir(tmp_path) + assert agents.subject_scopes("whatever") == () + + +def test_subject_scopes_reads_the_discovered_registry(store: KBStore) -> None: + agents.register( + store, subject="abc123", name="ci-bot", actor="human", scopes=(READ,), + ) + assert agents.subject_scopes("abc123") == (READ,) + + +def test_cli_registers_a_read_only_agent(store: KBStore) -> None: + from click.testing import CliRunner + + from vouch.cli import cli + + res = CliRunner().invoke(cli, [ + "agents", "register", "reader", "--subject", "abc123", "--scope", READ, + ]) + assert res.exit_code == 0, res.output + assert agents.scopes_for_subject(store, "abc123") == (READ,) + + +def test_cli_rejects_an_unknown_scope_cleanly(store: KBStore) -> None: + from click.testing import CliRunner + + from vouch.cli import cli + + res = CliRunner().invoke(cli, [ + "agents", "register", "reader", "--subject", "abc123", + "--scope", "kb:destroy", + ]) + assert res.exit_code != 0 + assert "unknown scope" in res.output + assert "Traceback" not in res.output + + +# --- the mcp-over-http middleware carries scopes through ----------------- + + +def test_mcp_http_middleware_applies_registry_scopes(store: KBStore) -> None: + """The ASGI wrapper must resolve scopes, not just the subject. + + This is the path a scoped MCP-over-HTTP agent actually authenticates on; + without it a read-only credential would arrive unscoped and hold + everything. + """ + import asyncio + + from vouch.http_server import _McpTrustASGI + + token = "mcp-token-example" + subject = trust.auth_subject_for_token(token) + agents.register( + store, subject=subject, name="mcp-reader", actor="human", scopes=(READ,), + ) + + seen: list[trust.VouchTrust] = [] + + async def inner(scope: dict, receive: object, send: object) -> None: + seen.append(trust.current()) + + app = _McpTrustASGI(inner, accepted=(token,)) + asyncio.run(app( + {"type": "http", + "headers": [(b"authorization", f"Bearer {token}".encode())]}, + None, None, + )) + + assert len(seen) == 1 + assert seen[0].auth_subject == subject + assert seen[0].scopes == (READ,) + assert seen[0].permits("kb.search") is True + assert seen[0].permits("kb.approve") is False + + +def test_mcp_http_middleware_passes_non_http_scopes_through( + store: KBStore +) -> None: + import asyncio + + from vouch.http_server import _McpTrustASGI + + calls: list[str] = [] + + async def inner(scope: dict, receive: object, send: object) -> None: + calls.append(scope["type"]) + + app = _McpTrustASGI(inner, accepted=("t",)) + asyncio.run(app({"type": "lifespan"}, None, None)) + assert calls == ["lifespan"] + + +def test_mcp_http_middleware_leaves_an_unauthenticated_call_unscoped( + store: KBStore +) -> None: + import asyncio + + from vouch.http_server import _McpTrustASGI + + seen: list[trust.VouchTrust] = [] + + async def inner(scope: dict, receive: object, send: object) -> None: + seen.append(trust.current()) + + app = _McpTrustASGI(inner, accepted=("t",)) + asyncio.run(app({"type": "http", "headers": []}, None, None)) + assert seen[0].auth_subject is None + assert seen[0].scopes == () From 2574eee1d8a93c365e0830fc948f1944023b9b37 Mon Sep 17 00:00:00 2001 From: minion1227 Date: Thu, 30 Jul 2026 15:18:13 -0700 Subject: [PATCH 2/3] chore(schemas): regenerate after capabilities gained a scopes field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit the schema drift check compares schemas/ against what scripts/gen_schemas.py emits from models.py. adding the per-credential scopes block to Capabilities changed that output, so the committed schema went stale. this is the sixth registration site a new field can miss — CLAUDE.md's four, plus hot_memory's coverage map, plus this. worth a line in the contributor notes separately. --- schemas/capabilities.schema.json | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/schemas/capabilities.schema.json b/schemas/capabilities.schema.json index 0832cad1..c08186b2 100644 --- a/schemas/capabilities.schema.json +++ b/schemas/capabilities.schema.json @@ -65,6 +65,11 @@ "title": "Review Gated", "type": "boolean" }, + "scopes": { + "additionalProperties": true, + "title": "Scopes", + "type": "object" + }, "scoping": { "additionalProperties": true, "title": "Scoping", From 4f500428f0b7a554cb6a054038d8976d668d8d66 Mon Sep 17 00:00:00 2001 From: minion1227 Date: Fri, 31 Jul 2026 09:56:04 -0700 Subject: [PATCH 3/3] fix(trust): classify the four methods that landed while this sat open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `METHOD_SCOPES` was exhaustive over `capabilities.METHODS` when this branch was written. four methods have merged to `test` since — kb.capture_correction (#679), and kb.list_goals / kb.propose_goal / kb.set_goal_status (#676) — leaving the table stale by four and all three matrix jobs red on `test_every_method_is_classified`. that failure is the guard working. by this branch's own rule an unclassified method is denied to a scoped caller, so merging as-is would have silently locked every scoped credential out of goal writes and correction capture. the right failure direction, but still a regression, and invisible from the diff. the classifications: - kb.list_goals -> kb:read. a listing that cannot change durable state, beside kb.list_sessions. - kb.propose_goal -> kb:propose. files a PENDING goal a human approves. - kb.capture_correction -> kb:propose. routes exclusively through propose_quoted_claim and has no import of approve. - kb.set_goal_status -> kb:approve, not kb:propose. this is the one that departs from the suggestion on the closing comment, and deliberately: it is a lifecycle op living in lifecycle.py beside supersede/archive/confirm, it mutates an already approved goal in place, and server.py documents it as "the only write path for goal status" — status moves never go through a second proposal. filing it under kb:propose would let a propose-only credential change durable state with no review, which is the exact boundary this module exists to hold. no other change: the scope machinery, the two safety rules and the tests are as reviewed. Co-authored-by: Cursor --- src/vouch/scopes.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/vouch/scopes.py b/src/vouch/scopes.py index 78a98c60..64c6150f 100644 --- a/src/vouch/scopes.py +++ b/src/vouch/scopes.py @@ -88,6 +88,7 @@ class ScopeError(ValueError): "kb.list_pending": READ, "kb.triage_pending": READ, "kb.list_sessions": READ, + "kb.list_goals": READ, "kb.session_transcript": READ, "kb.volunteer_context": READ, "kb.lint": READ, @@ -116,6 +117,11 @@ class ScopeError(ValueError): "kb.propose_relation": PROPOSE, "kb.propose_delete": PROPOSE, "kb.propose_theme": PROPOSE, + "kb.propose_goal": PROPOSE, + # capture_correction routes exclusively through propose_quoted_claim and + # has no import of approve — it files a PENDING claim like any other + # proposal, so it belongs here and not with the deciding verbs. + "kb.capture_correction": PROPOSE, "kb.cite": PROPOSE, "kb.source_verify": PROPOSE, "kb.session_start": PROPOSE, @@ -134,6 +140,13 @@ class ScopeError(ValueError): "kb.contradict": APPROVE, "kb.archive": APPROVE, "kb.confirm": APPROVE, + # set_goal_status is a lifecycle op, not a proposal: it mutates an already + # approved goal in place (life.set_goal_status, beside supersede/archive/ + # confirm) and is documented as "the only write path for goal status" — + # status moves never go through a second proposal. Filing it under PROPOSE + # would let a propose-only credential change durable state with no review, + # which is the boundary this module exists to hold. + "kb.set_goal_status": APPROVE, # --- kb:admin — destructive or index-wide maintenance ------------------ "kb.clear_claims": ADMIN, "kb.wipe_dead_refs": ADMIN,