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", 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 e7720687..3101fb4d 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 @@ -156,6 +158,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 9dba52bd..c9a19cdc 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -3882,7 +3882,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 38a34cb8..cb305ed0 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -1079,12 +1079,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 2118602b..0c10c507 100644 --- a/src/vouch/models.py +++ b/src/vouch/models.py @@ -581,6 +581,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 == ()