diff --git a/README.md b/README.md index 257089e..ff3be59 100644 --- a/README.md +++ b/README.md @@ -1,10 +1,18 @@ # AgentPass -**Runtime control, durable assurance, and observability for AI agent actions.** +**Open interoperability and conformance for action-bound AI agent authorization.** -AgentPass is an action-control boundary and evidence layer that sits outside -the agent loop. It checks tool calls before execution and preserves durable -records of what was proposed, authorized, executed, observed, and assessed. +AgentPass is an open reference implementation, action-control boundary, and +conformance project that sits outside the agent loop. It checks tool calls +before execution and preserves durable records of what was proposed, +authorized, executed, observed, and assessed. + +Its standards role is deliberately narrow: compose existing identity, +delegation, workload, policy, and signing standards into independently +verifiable evidence for a specific agent action. AgentPass does not define a +general agent identity or replace IAM, OAuth, MCP authorization, AuthZEN, +SPIFFE, WIMSE, OPA, Cedar, or provider business authorization. See +[Interoperability Positioning](docs/interoperability-positioning.md). ```text Agent proposes tool call -> AgentPass checks policy + state -> allow / deny / challenge @@ -244,6 +252,10 @@ contracts, receipts, or standards alignment: - Receipt profiles: [`docs/receipt-profiles.md`](docs/receipt-profiles.md) - Agentic identity standards crosswalk and executable MCP vector: [`docs/agentic-identity-standards-crosswalk.md`](docs/agentic-identity-standards-crosswalk.md) +- MCP authorization conformance suite v2: + [`fixtures/mcp-authorization-conformance-v2/`](fixtures/mcp-authorization-conformance-v2/) +- Interoperability positioning and contribution boundary: + [`docs/interoperability-positioning.md`](docs/interoperability-positioning.md) - Community RFCs: [`docs/proposals/`](docs/proposals/) - Intent observability execution plan: [Intent Observability & Assurance](https://github.com/users/dinpd/projects/2) @@ -378,16 +390,19 @@ AgentPass currently includes: ## Roadmap Focus -Near-term work has three public tracks: +Near-term work has four public tracks: -1. Production-grade intent observability and assurance: canonical evidence +1. Action-bound authorization interoperability: executable MCP conformance + cases, provider-verifier behavior, negative fixtures, implementation + profiles, and demonstrations with independent implementations. +2. Production-grade intent observability and assurance: canonical evidence contracts, action-boundary lifecycle hardening, OpenTelemetry causal tracing, correlation, measurement profiles, independent control assessment, complete population accounting, immutable assessment revisions, privacy, migration, and conformance. -2. Provider trust enforcement with production receipts and contract drift +3. Provider trust enforcement with production receipts and contract drift detection. -3. Framework and workflow wrappers selected from adopter demand. +4. Framework and workflow wrappers selected from adopter demand. The [Action Gate Roadmap](docs/action-gate-roadmap.md) tracks adopter-facing product demonstrations. The public diff --git a/docs/agentic-identity-standards-crosswalk.md b/docs/agentic-identity-standards-crosswalk.md index f3e746a..a088819 100644 --- a/docs/agentic-identity-standards-crosswalk.md +++ b/docs/agentic-identity-standards-crosswalk.md @@ -117,6 +117,12 @@ particular, the AgentPass evidence model should be compared with both active individual IETF drafts that already address authorization evidence and high-risk action receipts. +The project's positioning and testable success criteria are defined in +[Interoperability Positioning](interoperability-positioning.md). The +[MCP Authorization Conformance Suite v2](../fixtures/mcp-authorization-conformance-v2/) +turns this boundary into executable positive, trust-failure, lifecycle, and +closure cases. + The desired implementation seam is: ```text diff --git a/docs/interoperability-positioning.md b/docs/interoperability-positioning.md new file mode 100644 index 0000000..eb052bb --- /dev/null +++ b/docs/interoperability-positioning.md @@ -0,0 +1,71 @@ +# AgentPass Interoperability Positioning + +AgentPass is an open reference implementation and conformance project for +action-bound authorization of AI agent operations. + +Its focus is the provider trust boundary: + +> Can a provider independently verify that this exact action, with these +> arguments, was authorized for this principal, agent, runtime, delegation, +> policy, approval, audience, and time window—and handle it without unsafe +> replay? + +## What AgentPass Composes + +AgentPass reuses existing work rather than defining a general agent identity: + +- OpenID Connect and OAuth establish principals, clients, delegation, + resources, and audiences. +- SPIFFE and WIMSE address executing workload identity and workload-to-workload + security. +- AuthZEN separates policy enforcement from policy decision and is developing + approval and MCP authorization mappings. +- MCP authorization secures access to MCP servers. +- JOSE/JWKS, COSE, and HTTP Message Signatures provide established signing and + request-integrity building blocks. +- Shared Signals can communicate lifecycle, risk, and revocation changes. +- Provenance and receipt formats can describe what happened after dispatch. + +AgentPass experiments at the seam between those layers: canonical action +binding, signed authorization evidence, independent provider verification, +single-use or retry-safe consumption, and linked execution closure. + +## Contribution Boundary + +AgentPass should contribute, in this order: + +1. field mappings and explicit assumptions; +2. positive and negative interoperability vectors; +3. reference verifiers and provider middleware; +4. conformance tests and implementation reports; +5. narrowly scoped profiles where existing standards leave choices open; and +6. proposed new vocabulary only when multiple implementations demonstrate a + gap that cannot be addressed by profiling existing standards. + +AgentPass should not present itself as an agent identity provider, universal +agent registry, new DID method, OAuth replacement, or complete authorization +standard. + +## Evidence of Progress + +Documentation alone is not the success criterion. The useful milestones are: + +- independent implementations canonicalize the same action identically; +- providers accept the same valid evidence and reject the same invalid cases; +- replay, retry, revocation, and partial execution have explicit outcomes; +- execution closure remains linked to the authorized action; and +- implementation feedback is contributed to the relevant standards groups. + +The strongest milestone is two independent providers passing the same public +suite without AgentPass-specific coordination at runtime. + +## Current Public Artifacts + +- [Agentic Identity Standards Crosswalk](agentic-identity-standards-crosswalk.md) +- [MCP Authorization Interoperability Vector v1](../fixtures/mcp-authorization-interoperability-v1/) +- [MCP Authorization Conformance Suite v2](../fixtures/mcp-authorization-conformance-v2/) +- [Provider receipt profiles](receipt-profiles.md) + +These artifacts are experimental unless explicitly identified as an adopted +external standard. Passing an AgentPass suite is not certification by IETF, +OpenID, W3C, SPIFFE, NIST, ISO, or the MCP project. diff --git a/docs/standards-alignment.md b/docs/standards-alignment.md index a575701..19347aa 100644 --- a/docs/standards-alignment.md +++ b/docs/standards-alignment.md @@ -1,7 +1,10 @@ # Standards Alignment -AgentPass should align with distributed identity standards without becoming a -replacement for them. +AgentPass aligns with distributed identity and authorization standards as an +open reference implementation and conformance project for action-bound +authorization. It does not create a general agent identity or replace those +standards. The detailed project boundary is in +[Interoperability Positioning](interoperability-positioning.md). For the current field-level mapping across AuthZEN, OAuth, WIMSE, SPIFFE, MCP, authorization-evidence proposals, and execution provenance, see the @@ -9,6 +12,10 @@ authorization-evidence proposals, and execution provenance, see the Its companion [MCP authorization interoperability vector](../fixtures/mcp-authorization-interoperability-v1/) provides deterministic positive, mutation, audience, and replay cases. +The broader +[MCP Authorization Conformance Suite v2](../fixtures/mcp-authorization-conformance-v2/) +adds JWS/JWKS trust, delegation, runtime, policy, approval, revocation, retry, +and partial-execution cases. The core distinction is: diff --git a/fixtures/mcp-authorization-conformance-v2/README.md b/fixtures/mcp-authorization-conformance-v2/README.md new file mode 100644 index 0000000..f57da6b --- /dev/null +++ b/fixtures/mcp-authorization-conformance-v2/README.md @@ -0,0 +1,37 @@ +# MCP Authorization Conformance Suite v2 + +This directory contains an experimental, non-normative conformance suite for +action-bound authorization of a state-changing MCP `tools/call`. + +Run it from the repository root: + +```bash +python scripts/run_mcp_authorization_conformance_v2.py \ + fixtures/mcp-authorization-conformance-v2/vector.json +``` + +The runner exits with status `0` only when every case matches its expected +portable failure codes and execution-closure behavior. It covers: + +- RS256 JWS verification against an in-memory JWKS; +- trusted issuer, provider audience, algorithm, and key selection; +- exact MCP action and provider-owned field bindings; +- runtime identity, delegation attenuation, policy, approval, and risk state; +- unknown signing-key rejection; +- idempotent replay of a prior completed result; and +- partial-then-complete execution closure correlation. + +The runner generates fresh RSA keys in memory. It never writes or prints a +private key, JWS, or operational credential. All fixture identifiers use +reserved `.test` domains or visibly synthetic values. + +## Interpretation + +Passing this suite means an implementation agrees with these AgentPass +experimental cases. It is not certification against an IETF, OpenID, W3C, or +MCP standard and does not make the AgentPass evidence fields standardized. + +The partial-execution case demonstrates correlation semantics only. A real +provider must define whether a partial operation is retryable, compensatable, +or terminal and must enforce that decision atomically with its own idempotency +state. diff --git a/fixtures/mcp-authorization-conformance-v2/vector.json b/fixtures/mcp-authorization-conformance-v2/vector.json new file mode 100644 index 0000000..047557b --- /dev/null +++ b/fixtures/mcp-authorization-conformance-v2/vector.json @@ -0,0 +1,241 @@ +{ + "schema_version": "agentpass.mcp-authorization-conformance-vector.v2", + "suite_id": "mcp-authorization-conformance-v2", + "status": "experimental-non-normative", + "protocol_version": "2025-11-25", + "now": "2026-08-04T00:50:00Z", + "mcp_call": { + "jsonrpc": "2.0", + "id": "call-v2-2048", + "method": "tools/call", + "params": { + "name": "provider.crm.update_customer", + "arguments": { + "customer_id": "cus_example_2048", + "email_opt_in": false + } + } + }, + "authorization_evidence": { + "schema_version": "agentpass.action-authorization-evidence.v2", + "profile": "https://profiles.agentpass.example.test/mcp-action-authorization/v2", + "issuer": "https://authz.example.test", + "decision_id": "decision-example-2048", + "outcome": "ALLOW", + "tenant_id": "tenant-example", + "principal_id": "user:support-representative-example", + "user_id": "support-representative-example", + "agent_id": "support-agent", + "agent_app_id": "https://agents.example.test/support-agent/2.0.0", + "runtime_id": "spiffe://example.test/ns/agents/sa/support-agent-v2", + "delegation_id": "delegation-example-2048", + "delegation_attenuation": "downscoped", + "delegation_chain": [ + { + "actor": "user:support-representative-example", + "subject": "https://agents.example.test/support-agent/2.0.0", + "scope": "customer-support" + }, + { + "actor": "https://agents.example.test/support-agent/2.0.0", + "subject": "spiffe://example.test/ns/agents/sa/support-agent-v2", + "scope": "customer-email-preferences-write" + } + ], + "task_id": "support-case-resolution", + "job_id": "support_case_resolution_v2", + "case_id": "case-example-2048", + "customer_id": "cus_example_2048", + "audience": "https://provider.example.test/mcp", + "tool": "provider.crm.update_customer", + "action": "write", + "resource": "provider/customer/cus_example_2048", + "action_digest": "computed-by-runner", + "canonicalization": "agentid_canonical_json_v1", + "policy_id": "support-customer-update", + "policy_hash": "sha256:9b8c86ed2600634e78b32d67e594be07d02561725bc18f4fba70bde103d57c1e", + "approval_id": "approval-example-2048", + "approval_evidence_ref": "urn:example:approval:2048", + "approval_expires_at": "2026-08-04T00:52:00Z", + "jit_grant_id": "grant-example-2048", + "risk_state_ref": "urn:example:risk:low:2048", + "idempotency_key": "customer-update-example-2048", + "issued_at": "2026-08-04T00:49:00Z", + "expires_at": "2026-08-04T00:55:00Z" + }, + "provider": { + "id": "provider-crm-example", + "trusted_issuer": "https://authz.example.test", + "audience": "https://provider.example.test/mcp", + "key_id": "agentpass-conformance-v2", + "closure_id_prefix": "closure-example-2048", + "attempt_times": [ + "2026-08-04T00:50:01Z", + "2026-08-04T00:50:02Z" + ], + "expected_bindings": { + "tenant_id": "tenant-example", + "principal_id": "user:support-representative-example", + "user_id": "support-representative-example", + "agent_id": "support-agent", + "agent_app_id": "https://agents.example.test/support-agent/2.0.0", + "runtime_id": "spiffe://example.test/ns/agents/sa/support-agent-v2", + "delegation_id": "delegation-example-2048", + "delegation_attenuation": "downscoped", + "delegation_chain": [ + { + "actor": "user:support-representative-example", + "subject": "https://agents.example.test/support-agent/2.0.0", + "scope": "customer-support" + }, + { + "actor": "https://agents.example.test/support-agent/2.0.0", + "subject": "spiffe://example.test/ns/agents/sa/support-agent-v2", + "scope": "customer-email-preferences-write" + } + ], + "task_id": "support-case-resolution", + "job_id": "support_case_resolution_v2", + "case_id": "case-example-2048", + "customer_id": "cus_example_2048", + "tool": "provider.crm.update_customer", + "action": "write", + "resource": "provider/customer/cus_example_2048", + "policy_id": "support-customer-update", + "policy_hash": "sha256:9b8c86ed2600634e78b32d67e594be07d02561725bc18f4fba70bde103d57c1e", + "approval_id": "approval-example-2048", + "approval_evidence_ref": "urn:example:approval:2048", + "jit_grant_id": "grant-example-2048", + "risk_state_ref": "urn:example:risk:low:2048" + }, + "partial_result": { + "customer_id": "cus_example_2048", + "stage": "validated-not-committed" + }, + "result": { + "customer_id": "cus_example_2048", + "updated": true + } + }, + "cases": [ + { + "id": "valid_execution", + "expect": { + "ok": true, + "closure_statuses": ["executed"], + "dispositions": ["executed"], + "completed_mutations": 1 + } + }, + { + "id": "retry_returns_prior_closure", + "execution_plan": ["execute", "retry"], + "expect": { + "ok": true, + "closure_statuses": ["executed", "executed"], + "dispositions": ["executed", "replayed"], + "completed_mutations": 1 + } + }, + { + "id": "partial_then_completed_retry", + "execution_plan": ["partial", "execute"], + "expect": { + "ok": true, + "closure_statuses": ["partial", "executed"], + "dispositions": ["partial", "executed"], + "completed_mutations": 1, + "partial_attempts": 1 + } + }, + { + "id": "mutated_action_rejected", + "call_overrides": [ + { + "path": "params.arguments.email_opt_in", + "value": true + } + ], + "expect": { + "ok": false, + "codes": ["out_of_scope"] + } + }, + { + "id": "runtime_identity_mismatch_rejected", + "receipt_overrides": [ + { + "path": "runtime_id", + "value": "spiffe://attacker.example.test/ns/agents/sa/other-runtime" + } + ], + "expect": { + "ok": false, + "codes": ["out_of_scope"] + } + }, + { + "id": "delegation_escalation_rejected", + "receipt_overrides": [ + { + "path": "delegation_attenuation", + "value": "expanded" + } + ], + "expect": { + "ok": false, + "codes": ["out_of_scope"] + } + }, + { + "id": "expired_approval_rejected", + "receipt_overrides": [ + { + "path": "approval_expires_at", + "value": "2026-08-04T00:49:59Z" + } + ], + "expect": { + "ok": false, + "codes": ["expired"] + } + }, + { + "id": "revoked_risk_state_rejected", + "revoked_risk_state_refs": ["urn:example:risk:low:2048"], + "expect": { + "ok": false, + "codes": ["revoked"] + } + }, + { + "id": "policy_drift_rejected", + "receipt_overrides": [ + { + "path": "policy_hash", + "value": "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" + } + ], + "expect": { + "ok": false, + "codes": ["out_of_scope"] + } + }, + { + "id": "wrong_audience_rejected", + "provider_audience": "https://other-provider.example.test/mcp", + "expect": { + "ok": false, + "codes": ["wrong_audience"] + } + }, + { + "id": "unknown_signing_key_rejected", + "signing_key": "untrusted", + "expect": { + "ok": false, + "codes": ["unknown_key"] + } + } + ] +} diff --git a/scripts/run_mcp_authorization_conformance_v2.py b/scripts/run_mcp_authorization_conformance_v2.py new file mode 100644 index 0000000..5b4c64b --- /dev/null +++ b/scripts/run_mcp_authorization_conformance_v2.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python3 +"""Run the experimental AgentPass MCP authorization conformance suite v2.""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from jwt.algorithms import RSAAlgorithm + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from agentid.provider import ( + provider_receipt_failure_codes, + sign_provider_receipt_jws, + verify_provider_receipt, +) + + +def canonical_json(value: Any) -> str: + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def sha256_json(value: Any) -> str: + return "sha256:" + hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +def canonical_action( + call: dict[str, Any], *, audience: str, protocol_version: str +) -> dict[str, Any]: + params = call.get("params") if isinstance(call.get("params"), dict) else {} + arguments = params.get("arguments") if isinstance(params.get("arguments"), dict) else {} + return { + "protocol": "mcp", + "protocol_version": protocol_version, + "audience": audience, + "method": call.get("method"), + "tool": params.get("name"), + "arguments": { + key: value + for key, value in arguments.items() + if key != "_agentid_receipt" + }, + } + + +def action_digest(call: dict[str, Any], *, audience: str, protocol_version: str) -> str: + return sha256_json( + canonical_action(call, audience=audience, protocol_version=protocol_version) + ) + + +def _set_path(value: dict[str, Any], path: str, replacement: Any) -> None: + parts = path.split(".") + current: dict[str, Any] = value + for part in parts[:-1]: + child = current.get(part) + if not isinstance(child, dict): + raise ValueError(f"override path does not resolve to an object: {path}") + current = child + current[parts[-1]] = replacement + + +def _rsa_signing_material(key_id: str) -> tuple[str, dict[str, Any]]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + public_jwk = json.loads(RSAAlgorithm.to_jwk(private_key.public_key())) + public_jwk.update({"alg": "RS256", "kid": key_id, "use": "sig"}) + return private_pem, {"keys": [public_jwk]} + + +def _parse_timestamp(value: Any) -> datetime | None: + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def _append_once(findings: list[str], finding: str) -> None: + if finding not in findings: + findings.append(finding) + + +def _jws_trust_findings(findings: list[str]) -> list[str]: + return [ + finding + for finding in findings + if finding.startswith("receipt JWS") or finding.startswith("receipt JWKS") + ] + + +def _verify_case( + vector: dict[str, Any], + case: dict[str, Any], + call: dict[str, Any], + receipt: dict[str, Any], + signed_receipt: dict[str, Any], + trusted_jwks: dict[str, Any], +) -> list[str]: + provider = vector["provider"] + expected = provider["expected_bindings"] + expected_audience = case.get("provider_audience", provider["audience"]) + now = _parse_timestamp(vector["now"]) + if now is None: + raise ValueError("vector now must be an ISO-8601 timestamp") + + verification = verify_provider_receipt( + signed_receipt, + jwks=trusted_jwks, + expected_issuer=provider["trusted_issuer"], + expected_audience=expected_audience, + allowed_algs=["RS256"], + require_signed=True, + expected_tenant=expected["tenant_id"], + expected_agent=expected["agent_id"], + expected_tool=expected["tool"], + expected_action=expected["action"], + expected_resource=expected["resource"], + expected_job=expected["job_id"], + expected_case=expected["case_id"], + expected_customer=expected["customer_id"], + expected_approval=expected["approval_id"], + expected_jit_grant=expected["jit_grant_id"], + now=now, + ) + trust_findings = _jws_trust_findings(verification.findings) + if trust_findings: + return trust_findings + + findings = list(verification.findings) + actual_digest = action_digest( + call, + audience=receipt["audience"], + protocol_version=vector["protocol_version"], + ) + if receipt.get("action_digest") != actual_digest: + _append_once(findings, "receipt action_digest mismatch") + + for field, expected_value in expected.items(): + if receipt.get(field) != expected_value: + _append_once(findings, f"receipt {field} mismatch") + + approval_expires_at = _parse_timestamp(receipt.get("approval_expires_at")) + if approval_expires_at is None: + _append_once(findings, "receipt approval_expires_at is invalid") + elif approval_expires_at <= now: + _append_once(findings, "receipt approval is expired") + + revoked_risk_refs = case.get("revoked_risk_state_refs", []) + if receipt.get("risk_state_ref") in revoked_risk_refs: + _append_once(findings, "receipt risk state is revoked") + + return findings + + +def _execution_closure( + vector: dict[str, Any], + receipt: dict[str, Any], + *, + attempt: int, + status: str, +) -> dict[str, Any]: + provider = vector["provider"] + result = provider["partial_result"] if status == "partial" else provider["result"] + closure = { + "schema_version": "agentpass.execution-closure.v2", + "closure_id": f"{provider['closure_id_prefix']}-{attempt}", + "decision_id": receipt["decision_id"], + "authorization_evidence_digest": sha256_json(receipt), + "action_digest": receipt["action_digest"], + "provider_id": provider["id"], + "status": status, + "executed_at": provider["attempt_times"][attempt - 1], + "idempotency_key": receipt["idempotency_key"], + "result_digest": sha256_json(result), + } + closure["link_valid"] = ( + closure["decision_id"] == receipt["decision_id"] + and closure["action_digest"] == receipt["action_digest"] + ) + return closure + + +def _execute_case( + vector: dict[str, Any], case: dict[str, Any], receipt: dict[str, Any] +) -> dict[str, Any]: + attempts: list[dict[str, Any]] = [] + completed_closure: dict[str, Any] | None = None + completed_mutations = 0 + partial_attempts = 0 + + for attempt_number, instruction in enumerate( + case.get("execution_plan", ["execute"]), start=1 + ): + if instruction == "retry": + if completed_closure is None: + raise ValueError("retry requires a prior completed closure") + attempts.append( + { + "attempt": attempt_number, + "disposition": "replayed", + "closure": copy.deepcopy(completed_closure), + } + ) + continue + + if instruction == "partial": + partial_attempts += 1 + status = "partial" + disposition = "partial" + elif instruction == "execute": + completed_mutations += 1 + status = "executed" + disposition = "executed" + else: + raise ValueError(f"unsupported execution instruction: {instruction}") + + closure = _execution_closure( + vector, + receipt, + attempt=attempt_number, + status=status, + ) + attempts.append( + { + "attempt": attempt_number, + "disposition": disposition, + "closure": closure, + } + ) + if status == "executed": + completed_closure = closure + + return { + "attempts": attempts, + "completed_mutations": completed_mutations, + "partial_attempts": partial_attempts, + } + + +def _observed_result(result: dict[str, Any]) -> dict[str, Any]: + attempts = result.get("attempts", []) + return { + "ok": result["ok"], + "codes": result["codes"], + "closure_statuses": [attempt["closure"]["status"] for attempt in attempts], + "dispositions": [attempt["disposition"] for attempt in attempts], + "completed_mutations": result["completed_mutations"], + "partial_attempts": result["partial_attempts"], + } + + +def _expected_result(case: dict[str, Any]) -> dict[str, Any]: + expected = case["expect"] + return { + "ok": expected["ok"], + "codes": expected.get("codes", []), + "closure_statuses": expected.get("closure_statuses", []), + "dispositions": expected.get("dispositions", []), + "completed_mutations": expected.get("completed_mutations", 0), + "partial_attempts": expected.get("partial_attempts", 0), + } + + +def run_suite(vector: dict[str, Any]) -> dict[str, Any]: + provider = vector["provider"] + trusted_private_key, trusted_jwks = _rsa_signing_material(provider["key_id"]) + untrusted_private_key, _ = _rsa_signing_material("untrusted-test-key") + + receipt_template = copy.deepcopy(vector["authorization_evidence"]) + receipt_template["action_digest"] = action_digest( + vector["mcp_call"], + audience=receipt_template["audience"], + protocol_version=vector["protocol_version"], + ) + + results: list[dict[str, Any]] = [] + for case in vector["cases"]: + call = copy.deepcopy(vector["mcp_call"]) + for override in case.get("call_overrides", []): + _set_path(call, override["path"], override["value"]) + + receipt = copy.deepcopy(receipt_template) + for override in case.get("receipt_overrides", []): + _set_path(receipt, override["path"], override["value"]) + + signer = ( + untrusted_private_key + if case.get("signing_key") == "untrusted" + else trusted_private_key + ) + key_id = ( + "untrusted-test-key" + if case.get("signing_key") == "untrusted" + else provider["key_id"] + ) + signed_receipt = sign_provider_receipt_jws( + receipt, + signer, + issuer=receipt["issuer"], + subject=receipt["agent_id"], + audience=receipt["audience"], + key_id=key_id, + algorithm="RS256", + ) + + findings = _verify_case( + vector, + case, + call, + receipt, + signed_receipt, + trusted_jwks, + ) + execution = ( + _execute_case(vector, case, receipt) + if not findings + else {"attempts": [], "completed_mutations": 0, "partial_attempts": 0} + ) + result = { + "id": case["id"], + "ok": not findings, + "codes": provider_receipt_failure_codes(findings), + "findings": findings, + **execution, + } + result["matches_expected"] = _observed_result(result) == _expected_result(case) + results.append(result) + + return { + "schema_version": "agentpass.mcp-authorization-conformance-result.v2", + "suite_id": vector["suite_id"], + "signing_profile": "RS256/JWKS", + "trusted_key_id": provider["key_id"], + "all_cases_match": all(result["matches_expected"] for result in results), + "cases": results, + } + + +def load_vector(path: str | Path) -> dict[str, Any]: + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("vector", type=Path, help="Path to the v2 suite vector JSON") + args = parser.parse_args() + result = run_suite(load_vector(args.vector)) + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["all_cases_match"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_mcp_authorization_conformance_v2.py b/tests/test_mcp_authorization_conformance_v2.py new file mode 100644 index 0000000..edbf254 --- /dev/null +++ b/tests/test_mcp_authorization_conformance_v2.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent.parent +VECTOR = ROOT / "fixtures" / "mcp-authorization-conformance-v2" / "vector.json" +RUNNER = ROOT / "scripts" / "run_mcp_authorization_conformance_v2.py" + + +def run_suite() -> dict: + completed = subprocess.run( + [sys.executable, str(RUNNER), str(VECTOR)], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + return json.loads(completed.stdout) + + +def test_mcp_authorization_conformance_v2_is_deterministic(): + first = run_suite() + second = run_suite() + + assert first == second + assert first["signing_profile"] == "RS256/JWKS" + assert first["trusted_key_id"] == "agentpass-conformance-v2" + assert first["all_cases_match"] is True + + +def test_mcp_authorization_conformance_v2_covers_required_fail_closed_cases(): + result = run_suite() + cases = {case["id"]: case for case in result["cases"]} + + expected_codes = { + "mutated_action_rejected": ["out_of_scope"], + "runtime_identity_mismatch_rejected": ["out_of_scope"], + "delegation_escalation_rejected": ["out_of_scope"], + "expired_approval_rejected": ["expired"], + "revoked_risk_state_rejected": ["revoked"], + "policy_drift_rejected": ["out_of_scope"], + "wrong_audience_rejected": ["wrong_audience"], + "unknown_signing_key_rejected": ["unknown_key"], + } + + for case_id, codes in expected_codes.items(): + assert cases[case_id]["ok"] is False + assert cases[case_id]["codes"] == codes + assert cases[case_id]["attempts"] == [] + assert cases[case_id]["completed_mutations"] == 0 + + +def test_mcp_authorization_conformance_v2_models_retry_and_partial_closure(): + result = run_suite() + cases = {case["id"]: case for case in result["cases"]} + + valid = cases["valid_execution"] + assert valid["ok"] is True + assert valid["attempts"][0]["closure"]["status"] == "executed" + assert valid["attempts"][0]["closure"]["link_valid"] is True + + retry = cases["retry_returns_prior_closure"] + assert retry["completed_mutations"] == 1 + assert [attempt["disposition"] for attempt in retry["attempts"]] == [ + "executed", + "replayed", + ] + assert retry["attempts"][0]["closure"] == retry["attempts"][1]["closure"] + + partial = cases["partial_then_completed_retry"] + assert partial["completed_mutations"] == 1 + assert partial["partial_attempts"] == 1 + assert [attempt["closure"]["status"] for attempt in partial["attempts"]] == [ + "partial", + "executed", + ] + assert all(attempt["closure"]["link_valid"] for attempt in partial["attempts"]) + assert len({attempt["closure"]["decision_id"] for attempt in partial["attempts"]}) == 1 + assert len({attempt["closure"]["action_digest"] for attempt in partial["attempts"]}) == 1