"""Deterministic audit of an illustrative MCP host authority boundary. This fixture does not open a transport, parse JSON-RPC, authenticate a user, perform OAuth, inspect a real MCP server, or prove protocol conformance. It checks a frozen local host policy only. Production hosts must bind consent and authorization to authenticated identities, current policy, exact requests, and real transport/session evidence at execution time. """ from __future__ import annotations from dataclasses import asdict, dataclass from hashlib import sha256 import json from math import isfinite import re from typing import Literal from urllib.parse import urlsplit Transport = Literal["streamable_http"] Primitive = Literal["tool", "resource", "prompt"] Effect = Literal["read", "write"] ApprovalMode = Literal["none", "per_call"] Decision = Literal[ "ALLOW", "DENY_PROTOCOL", "DENY_HOST_BINDING", "DENY_SERVER_BINDING", "DENY_CAPABILITY", "DENY_METHOD", "DENY_AUDIENCE", "DENY_SCOPE", "DENY_CONSENT", "DENY_PAYLOAD", "DENY_TIMEOUT", "DENY_CALL_BUDGET", ] ID_PATTERN = re.compile(r"^[a-z][a-z0-9._:-]{2,95}$") NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]{1,63}$") SCOPE_PATTERN = re.compile(r"^[a-z][a-z0-9:_-]{1,95}$") DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") PROTOCOL_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$") def _exact_text(name: str, value: object, *, maximum: int = 512) -> str: if type(value) is not str or not value.strip(): raise ValueError(f"{name} must be non-empty exact str") if value != value.strip(): raise ValueError(f"{name} must not have surrounding whitespace") if len(value) > maximum or any(ord(character) < 32 for character in value): raise ValueError(f"{name} is outside the bounded text contract") return value def _canonical_id(name: str, value: object) -> str: text = _exact_text(name, value, maximum=96) if not ID_PATTERN.fullmatch(text): raise ValueError(f"{name} must be a canonical identifier") return text def _exact_bool(name: str, value: object) -> bool: if type(value) is not bool: raise ValueError(f"{name} must be exact bool") return value def _exact_int(name: str, value: object, *, minimum: int, maximum: int) -> int: if type(value) is not int: raise ValueError(f"{name} must be exact int") if not minimum <= value <= maximum: raise ValueError(f"{name} must be between {minimum} and {maximum}") return value def _finite_number(name: str, value: object, *, minimum: float, maximum: float) -> float: if type(value) is int: numeric: int | float = value elif type(value) is float: numeric = value if not isfinite(numeric): raise ValueError(f"{name} must be finite") else: raise ValueError(f"{name} must be a non-boolean number") if not minimum <= numeric <= maximum: raise ValueError(f"{name} must be between {minimum} and {maximum}") return float(numeric) def _https_origin(name: str, value: object) -> str: text = _exact_text(name, value, maximum=256) parsed = urlsplit(text) if ( parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment or parsed.path not in {"", "/"} ): raise ValueError(f"{name} must be an HTTPS origin without credentials or path") if parsed.port is not None and not 1 <= parsed.port <= 65535: raise ValueError(f"{name} has an invalid port") return text[:-1] if text.endswith("/") else text def _https_audience(name: str, value: object) -> str: text = _exact_text(name, value, maximum=256) parsed = urlsplit(text) if parsed.scheme != "https" or not parsed.hostname or parsed.fragment: raise ValueError(f"{name} must be an HTTPS resource audience") if parsed.username is not None or parsed.password is not None: raise ValueError(f"{name} must not contain credentials") return text def _stable_digest(payload: object) -> str: encoded = json.dumps( payload, allow_nan=False, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") return sha256(encoded).hexdigest() @dataclass(frozen=True) class CapabilityPolicy: name: str primitive: Primitive method: str effect: Effect audience: str required_scopes: tuple[str, ...] approval_mode: ApprovalMode max_payload_bytes: int max_timeout_seconds: float def __post_init__(self) -> None: _exact_text("capability.name", self.name, maximum=64) if not NAME_PATTERN.fullmatch(self.name): raise ValueError("capability.name must be canonical snake_case") if self.primitive not in {"tool", "resource", "prompt"}: raise ValueError("capability.primitive is unsupported") _exact_text("capability.primitive", self.primitive, maximum=16) _exact_text("capability.method", self.method, maximum=64) expected_prefix = {"tool": "tools/", "resource": "resources/", "prompt": "prompts/"}[ self.primitive ] if not self.method.startswith(expected_prefix): raise ValueError("capability.method does not match primitive") if self.effect not in {"read", "write"}: raise ValueError("capability.effect is unsupported") _exact_text("capability.effect", self.effect, maximum=8) normalized_audience = _https_audience("capability.audience", self.audience) object.__setattr__(self, "audience", normalized_audience) try: scopes = tuple(self.required_scopes) except TypeError as error: raise ValueError("required_scopes must be iterable") from error object.__setattr__(self, "required_scopes", scopes) if not scopes: raise ValueError("required_scopes must not be empty") for scope in scopes: _exact_text("required_scope", scope, maximum=96) if not SCOPE_PATTERN.fullmatch(scope): raise ValueError("required_scope must be canonical") if len(scopes) != len(set(scopes)): raise ValueError("required_scopes must be unique") if tuple(sorted(scopes)) != scopes: raise ValueError("required_scopes must be sorted") if self.approval_mode not in {"none", "per_call"}: raise ValueError("approval_mode is unsupported") _exact_text("approval_mode", self.approval_mode, maximum=16) if self.effect == "write" and self.approval_mode != "per_call": raise ValueError("write capability requires per-call approval") _exact_int("max_payload_bytes", self.max_payload_bytes, minimum=1, maximum=10_000_000) timeout = _finite_number( "max_timeout_seconds", self.max_timeout_seconds, minimum=0.001, maximum=3600, ) object.__setattr__(self, "max_timeout_seconds", timeout) @dataclass(frozen=True) class McpHostContract: version: str protocol_version: str host_id: str client_id: str server_id: str server_origin: str transport: Transport capabilities: tuple[CapabilityPolicy, ...] max_calls_per_run: int def __post_init__(self) -> None: _canonical_id("contract.version", self.version) _exact_text("contract.protocol_version", self.protocol_version, maximum=10) if not PROTOCOL_PATTERN.fullmatch(self.protocol_version): raise ValueError("protocol_version must be a dated MCP version") _canonical_id("contract.host_id", self.host_id) _canonical_id("contract.client_id", self.client_id) _canonical_id("contract.server_id", self.server_id) origin = _https_origin("contract.server_origin", self.server_origin) object.__setattr__(self, "server_origin", origin) if self.transport != "streamable_http": raise ValueError("transport is unsupported") _exact_text("contract.transport", self.transport, maximum=32) try: capabilities = tuple(self.capabilities) except TypeError as error: raise ValueError("capabilities must be iterable") from error object.__setattr__(self, "capabilities", capabilities) if not capabilities or any(type(item) is not CapabilityPolicy for item in capabilities): raise ValueError("capabilities must contain exact CapabilityPolicy records") names = [item.name for item in capabilities] if len(names) != len(set(names)): raise ValueError("capability names must be unique") _exact_int("max_calls_per_run", self.max_calls_per_run, minimum=1, maximum=100_000) @property def content_id(self) -> str: return "mcp-host-contract@sha256:" + _stable_digest(asdict(self)) def approval_binding( *, request_id: str, server_id: str, capability_name: str, audience: str, payload_digest: str, ) -> str: for name, value in ( ("request_id", request_id), ("server_id", server_id), ("capability_name", capability_name), ): _canonical_id(name, value) normalized_audience = _https_audience("audience", audience) _exact_text("payload_digest", payload_digest, maximum=64) if not DIGEST_PATTERN.fullmatch(payload_digest): raise ValueError("payload_digest must be lowercase SHA-256") return _stable_digest( { "audience": normalized_audience, "capability_name": capability_name, "payload_digest": payload_digest, "request_id": request_id, "server_id": server_id, } ) @dataclass(frozen=True) class McpCallProposal: protocol_version: str host_id: str client_id: str server_id: str server_origin: str request_id: str method: str capability_name: str audience: str delegated_scopes: tuple[str, ...] payload_manifest: str payload_digest: str payload_bytes: int timeout_seconds: float user_approved: bool approval_binding: str | None call_index: int def __post_init__(self) -> None: _exact_text("proposal.protocol_version", self.protocol_version, maximum=10) if not PROTOCOL_PATTERN.fullmatch(self.protocol_version): raise ValueError("proposal.protocol_version must be dated") _canonical_id("proposal.host_id", self.host_id) _canonical_id("proposal.client_id", self.client_id) _canonical_id("proposal.server_id", self.server_id) origin = _https_origin("proposal.server_origin", self.server_origin) object.__setattr__(self, "server_origin", origin) _canonical_id("proposal.request_id", self.request_id) _exact_text("proposal.method", self.method, maximum=64) _exact_text("proposal.capability_name", self.capability_name, maximum=64) if not NAME_PATTERN.fullmatch(self.capability_name): raise ValueError("proposal.capability_name must be canonical snake_case") audience = _https_audience("proposal.audience", self.audience) object.__setattr__(self, "audience", audience) try: scopes = tuple(self.delegated_scopes) except TypeError as error: raise ValueError("delegated_scopes must be iterable") from error object.__setattr__(self, "delegated_scopes", scopes) for scope in scopes: _exact_text("delegated_scope", scope, maximum=96) if not SCOPE_PATTERN.fullmatch(scope): raise ValueError("delegated_scope must be canonical") if len(scopes) != len(set(scopes)): raise ValueError("delegated_scopes must be unique") if tuple(sorted(scopes)) != scopes: raise ValueError("delegated_scopes must be sorted") manifest = _exact_text( "proposal.payload_manifest", self.payload_manifest, maximum=10_000_000, ) try: parsed_manifest = json.loads(manifest) canonical_manifest = json.dumps( parsed_manifest, allow_nan=False, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ) except (ValueError, TypeError) as error: raise ValueError("proposal.payload_manifest must be finite canonical JSON") from error if type(parsed_manifest) is not dict or canonical_manifest != manifest: raise ValueError("proposal.payload_manifest must be one canonical JSON object") _exact_text("proposal.payload_digest", self.payload_digest, maximum=64) if not DIGEST_PATTERN.fullmatch(self.payload_digest): raise ValueError("proposal.payload_digest must be lowercase SHA-256") _exact_int("proposal.payload_bytes", self.payload_bytes, minimum=0, maximum=100_000_000) encoded_manifest = manifest.encode("utf-8") if sha256(encoded_manifest).hexdigest() != self.payload_digest: raise ValueError("proposal.payload_digest does not match payload_manifest") if len(encoded_manifest) != self.payload_bytes: raise ValueError("proposal.payload_bytes does not match payload_manifest") timeout = _finite_number( "proposal.timeout_seconds", self.timeout_seconds, minimum=0.001, maximum=86_400, ) object.__setattr__(self, "timeout_seconds", timeout) _exact_bool("proposal.user_approved", self.user_approved) if self.approval_binding is not None: _exact_text("proposal.approval_binding", self.approval_binding, maximum=64) if not DIGEST_PATTERN.fullmatch(self.approval_binding): raise ValueError("proposal.approval_binding must be lowercase SHA-256") _exact_int("proposal.call_index", self.call_index, minimum=1, maximum=1_000_000) @dataclass(frozen=True) class AuditResult: contract: McpHostContract proposal: McpCallProposal capability: CapabilityPolicy | None decision: Decision def _rebuild_capability(value: object) -> CapabilityPolicy: if type(value) is not CapabilityPolicy: raise ValueError("capabilities must contain exact CapabilityPolicy records") return CapabilityPolicy( value.name, value.primitive, value.method, value.effect, value.audience, value.required_scopes, value.approval_mode, value.max_payload_bytes, value.max_timeout_seconds, ) def _rebuild_contract(value: object) -> McpHostContract: if type(value) is not McpHostContract: raise ValueError("contract must be exact McpHostContract") try: capabilities = tuple(_rebuild_capability(item) for item in value.capabilities) return McpHostContract( value.version, value.protocol_version, value.host_id, value.client_id, value.server_id, value.server_origin, value.transport, capabilities, value.max_calls_per_run, ) except (AttributeError, TypeError) as error: raise ValueError("contract failed reconstruction") from error def _rebuild_proposal(value: object) -> McpCallProposal: if type(value) is not McpCallProposal: raise ValueError("proposal must be exact McpCallProposal") try: return McpCallProposal( value.protocol_version, value.host_id, value.client_id, value.server_id, value.server_origin, value.request_id, value.method, value.capability_name, value.audience, value.delegated_scopes, value.payload_manifest, value.payload_digest, value.payload_bytes, value.timeout_seconds, value.user_approved, value.approval_binding, value.call_index, ) except (AttributeError, TypeError) as error: raise ValueError("proposal failed reconstruction") from error def audit(contract: object, proposal: object) -> AuditResult: """Evaluate one host proposal without connecting to or trusting a server.""" checked_contract = _rebuild_contract(contract) checked_proposal = _rebuild_proposal(proposal) capabilities = {item.name: item for item in checked_contract.capabilities} capability = capabilities.get(checked_proposal.capability_name) if checked_proposal.protocol_version != checked_contract.protocol_version: decision: Decision = "DENY_PROTOCOL" elif ( checked_proposal.host_id != checked_contract.host_id or checked_proposal.client_id != checked_contract.client_id ): decision = "DENY_HOST_BINDING" elif ( checked_proposal.server_id != checked_contract.server_id or checked_proposal.server_origin != checked_contract.server_origin ): decision = "DENY_SERVER_BINDING" elif capability is None: decision = "DENY_CAPABILITY" elif checked_proposal.method != capability.method: decision = "DENY_METHOD" elif checked_proposal.audience != capability.audience: decision = "DENY_AUDIENCE" elif checked_proposal.delegated_scopes != capability.required_scopes: decision = "DENY_SCOPE" elif capability.approval_mode == "per_call" and ( not checked_proposal.user_approved or checked_proposal.approval_binding != approval_binding( request_id=checked_proposal.request_id, server_id=checked_proposal.server_id, capability_name=checked_proposal.capability_name, audience=checked_proposal.audience, payload_digest=checked_proposal.payload_digest, ) ): decision = "DENY_CONSENT" elif checked_proposal.payload_bytes > capability.max_payload_bytes: decision = "DENY_PAYLOAD" elif checked_proposal.timeout_seconds > capability.max_timeout_seconds: decision = "DENY_TIMEOUT" elif checked_proposal.call_index > checked_contract.max_calls_per_run: decision = "DENY_CALL_BUDGET" else: decision = "ALLOW" return AuditResult(checked_contract, checked_proposal, capability, decision) ILLUSTRATIVE_CAPABILITY = CapabilityPolicy( name="create_draft_event", primitive="tool", method="tools/call", effect="write", audience="https://calendar.example.invalid", required_scopes=("calendar:drafts:write",), approval_mode="per_call", max_payload_bytes=4096, max_timeout_seconds=5.0, ) ILLUSTRATIVE_CONTRACT = McpHostContract( version="mcp_host_contract_v1", protocol_version="2026-07-28", host_id="academy_host", client_id="academy_calendar_client", server_id="calendar_mcp_server", server_origin="https://mcp.example.invalid", transport="streamable_http", capabilities=(ILLUSTRATIVE_CAPABILITY,), max_calls_per_run=2, ) ILLUSTRATIVE_PAYLOAD_MANIFEST = ( '{"title":"Illustrative review","write_mode":"draft"}' ) ILLUSTRATIVE_PAYLOAD_DIGEST = sha256( ILLUSTRATIVE_PAYLOAD_MANIFEST.encode("utf-8") ).hexdigest() ILLUSTRATIVE_PROPOSAL = McpCallProposal( protocol_version="2026-07-28", host_id="academy_host", client_id="academy_calendar_client", server_id="calendar_mcp_server", server_origin="https://mcp.example.invalid", request_id="request_0042", method="tools/call", capability_name="create_draft_event", audience="https://calendar.example.invalid", delegated_scopes=("calendar:drafts:write",), payload_manifest=ILLUSTRATIVE_PAYLOAD_MANIFEST, payload_digest=ILLUSTRATIVE_PAYLOAD_DIGEST, payload_bytes=len(ILLUSTRATIVE_PAYLOAD_MANIFEST.encode("utf-8")), timeout_seconds=3.0, user_approved=True, approval_binding=approval_binding( request_id="request_0042", server_id="calendar_mcp_server", capability_name="create_draft_event", audience="https://calendar.example.invalid", payload_digest=ILLUSTRATIVE_PAYLOAD_DIGEST, ), call_index=1, ) def format_example() -> str: result = audit(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_PROPOSAL) capability = result.capability if capability is None: # Defensive; the illustrative proposal is fixed. raise AssertionError("illustrative capability is missing") return "\n".join( ( "example=illustrative_only", f"contract_id={result.contract.content_id}", f"protocol={result.contract.protocol_version}", f"host={result.contract.host_id}", f"client={result.contract.client_id}", f"server={result.contract.server_id}", f"capability={capability.name}", f"transport={result.contract.transport}", "delegated_scopes=" + ",".join(result.proposal.delegated_scopes), "consent=HOST_BOUND_TO_CALL", f"decision={result.decision}", "claim=LOCAL_HOST_POLICY_CHECK_ONLY", "certification=NOT_MCP_OR_OAUTH_CONFORMANCE", ) ) def main() -> None: print(format_example()) if __name__ == "__main__": main()