From ff64ed0e492dbd24dd64ab1d1d1fc1f23b017cd5 Mon Sep 17 00:00:00 2001 From: Imran Siddique Date: Sun, 16 Aug 2026 19:42:07 -0700 Subject: [PATCH] feat: add verifiable catalog approval provenance Signed-off-by: Imran Siddique --- docs/spec/catalog-approval-provenance.md | 25 ++++ schemas/catalog-approval.schema.json | 45 +++++++ src/cmcp_runtime/catalog/approval.py | 154 +++++++++++++++++++++++ tests/unit/test_catalog_approval.py | 78 ++++++++++++ 4 files changed, 302 insertions(+) create mode 100644 docs/spec/catalog-approval-provenance.md create mode 100644 schemas/catalog-approval.schema.json create mode 100644 src/cmcp_runtime/catalog/approval.py create mode 100644 tests/unit/test_catalog_approval.py diff --git a/docs/spec/catalog-approval-provenance.md b/docs/spec/catalog-approval-provenance.md new file mode 100644 index 00000000..86fe59c5 --- /dev/null +++ b/docs/spec/catalog-approval-provenance.md @@ -0,0 +1,25 @@ +# Verifiable catalog-approval provenance + +The active catalog hash proves that the gateway is using the approved bytes at +runtime. It does not prove how those bytes reached the approved state. The +detached record defined by `schemas/catalog-approval.schema.json` supplies that +missing history without putting mutable approval status into the catalog or +TRACE claim. + +Each record identifies a catalog, sequence, previous record, previous and new +catalog hashes, a change-set digest, an automated-checks digest, and an +approval policy. Every approval is an Ed25519 signature over the record body and +its own principal, issuer, role, validity interval, and key identifier. The +verifier resolves keys from trusted configuration; no record-embedded key can +bootstrap trust. It rejects unknown fields, revoked or expired keys, invalid +signatures, duplicate principals or roles when the policy requires distinctness, +and records whose `new_catalog_hash` differs from the runtime catalog hash. + +The record chain is not a freshness oracle. A verifier must obtain the expected +previous-record checkpoint from an external pin or transparency receipt. A +valid chain presented from an old checkpoint remains an old, valid chain rather +than proof that it is the latest approval. + +This provenance answers who approved which catalog change and which checks were +reported. It does not prove tool safety, prevent reviewer collusion, or replace +runtime catalog measurement and attestation. diff --git a/schemas/catalog-approval.schema.json b/schemas/catalog-approval.schema.json new file mode 100644 index 00000000..75c8054e --- /dev/null +++ b/schemas/catalog-approval.schema.json @@ -0,0 +1,45 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://cmcp.agentrust-io.com/schemas/catalog-approval.schema.json", + "title": "cMCP Detached Catalog Approval v1", + "type": "object", + "additionalProperties": false, + "required": ["profile", "catalog_id", "sequence", "previous_record_hash", "previous_catalog_hash", "new_catalog_hash", "change_set_digest", "approval_policy", "automated_checks_digest", "approvals"], + "properties": { + "profile": {"const": "tag:agentrust-io.com,2026:cmcp-catalog-approval-v1"}, + "catalog_id": {"type": "string", "minLength": 1}, + "sequence": {"type": "integer", "minimum": 1}, + "previous_record_hash": {"$ref": "#/$defs/digest"}, + "previous_catalog_hash": {"$ref": "#/$defs/digest"}, + "new_catalog_hash": {"$ref": "#/$defs/digest"}, + "change_set_digest": {"$ref": "#/$defs/digest"}, + "automated_checks_digest": {"$ref": "#/$defs/digest"}, + "approval_policy": { + "type": "object", "additionalProperties": false, + "required": ["policy_id", "policy_hash", "threshold", "distinct_principals", "distinct_roles"], + "properties": { + "policy_id": {"type": "string", "minLength": 1}, + "policy_hash": {"$ref": "#/$defs/digest"}, + "threshold": {"type": "integer", "minimum": 1}, + "distinct_principals": {"type": "boolean"}, + "distinct_roles": {"type": "boolean"} + } + }, + "approvals": { + "type": "array", "minItems": 1, + "items": {"type": "object", "additionalProperties": false, + "required": ["principal_id", "issuer", "key_id", "role", "approved_at", "expires_at", "signature"], + "properties": { + "principal_id": {"type": "string", "minLength": 1}, + "issuer": {"type": "string", "minLength": 1}, + "key_id": {"type": "string", "minLength": 1}, + "role": {"type": "string", "minLength": 1}, + "approved_at": {"type": "integer", "minimum": 0}, + "expires_at": {"type": "integer", "minimum": 1}, + "signature": {"type": "string", "pattern": "^[A-Za-z0-9_-]+$"} + } + } + } + }, + "$defs": {"digest": {"type": "string", "pattern": "^sha256:[0-9a-f]{64}$"}} +} diff --git a/src/cmcp_runtime/catalog/approval.py b/src/cmcp_runtime/catalog/approval.py new file mode 100644 index 00000000..1351151f --- /dev/null +++ b/src/cmcp_runtime/catalog/approval.py @@ -0,0 +1,154 @@ +"""Detached, signed provenance for approved tool-catalog changes (#517).""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import time +from dataclasses import dataclass +from typing import Any + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +PROFILE = "tag:agentrust-io.com,2026:cmcp-catalog-approval-v1" + + +class CatalogApprovalError(ValueError): + """The detached approval record is malformed or cannot be trusted.""" + + +class CatalogApprovalMismatch(CatalogApprovalError): + """The record does not apply to the catalog or policy being verified.""" + + +@dataclass(frozen=True) +class TrustedReviewer: + principal_id: str + issuer: str + key: Ed25519PublicKey + role: str | None = None + + +def canonical_json(value: Any) -> bytes: + """Return the RFC 8785-compatible JSON form used by cMCP records.""" + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + + +def digest_json(value: Any) -> str: + return "sha256:" + hashlib.sha256(canonical_json(value)).hexdigest() + + +def _b64(value: bytes) -> str: + return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") + + +def _decode(value: Any) -> bytes: + if not isinstance(value, str) or not value: + raise CatalogApprovalError("signature must be a base64url string") + try: + return base64.urlsafe_b64decode(value + "=" * ((4 - len(value) % 4) % 4)) + except (ValueError, TypeError) as exc: + raise CatalogApprovalError("signature is not valid base64url") from exc + + +def _approval_input(record: dict[str, Any], approval: dict[str, Any]) -> bytes: + body = {k: v for k, v in record.items() if k != "approvals"} + unsigned = {k: v for k, v in approval.items() if k != "signature"} + return canonical_json({"record": body, "approval": unsigned}) + + +def sign_approval( + record: dict[str, Any], approval: dict[str, Any], key: Ed25519PrivateKey +) -> dict[str, Any]: + """Return an approval with a signature over the record and approval fields.""" + signed = dict(approval) + signed["signature"] = _b64(key.sign(_approval_input(record, signed))) + return signed + + +def _require_digest(value: Any, field: str) -> str: + if not isinstance(value, str) or len(value) != 71 or not value.startswith("sha256:"): + raise CatalogApprovalError(f"{field} must be a sha256 digest") + if any(c not in "0123456789abcdef" for c in value[7:]): + raise CatalogApprovalError(f"{field} must be lowercase hexadecimal") + return value + + +def verify_catalog_change( + record: dict[str, Any], + trusted_reviewers: dict[str, TrustedReviewer], + *, + runtime_catalog_hash: str, + now: int | None = None, + revoked_key_ids: frozenset[str] = frozenset(), + expected_previous_record_hash: str | None = None, +) -> dict[str, Any]: + """Verify policy, chain, reviewer signatures, freshness, and runtime binding.""" + if not isinstance(record, dict) or record.get("profile") != PROFILE: + raise CatalogApprovalError("unknown or missing catalog approval profile") + required = { + "catalog_id", "sequence", "previous_record_hash", "previous_catalog_hash", + "new_catalog_hash", "change_set_digest", "approval_policy", "automated_checks_digest", + "approvals", + } + if set(record) != {"profile", *required}: + raise CatalogApprovalError("record contains missing or unknown fields") + if not isinstance(record["sequence"], int) or isinstance(record["sequence"], bool) or record["sequence"] < 1: + raise CatalogApprovalError("sequence must be a positive integer") + for field in ("previous_record_hash", "previous_catalog_hash", "new_catalog_hash", "change_set_digest", "automated_checks_digest"): + _require_digest(record[field], field) + if expected_previous_record_hash is not None and record["previous_record_hash"] != expected_previous_record_hash: + raise CatalogApprovalMismatch("record is not the expected next chain element") + if record["new_catalog_hash"] != runtime_catalog_hash: + raise CatalogApprovalMismatch("new_catalog_hash does not match runtime catalog hash") + + policy = record["approval_policy"] + if not isinstance(policy, dict) or set(policy) != {"policy_id", "policy_hash", "threshold", "distinct_principals", "distinct_roles"}: + raise CatalogApprovalError("approval_policy has missing or unknown fields") + _require_digest(policy["policy_hash"], "approval_policy.policy_hash") + threshold = policy["threshold"] + if not isinstance(threshold, int) or isinstance(threshold, bool) or threshold < 1: + raise CatalogApprovalError("approval threshold must be a positive integer") + instant = int(time.time()) if now is None else now + approvals = record["approvals"] + if not isinstance(approvals, list) or len(approvals) < threshold: + raise CatalogApprovalMismatch("approval threshold is not satisfied") + principals: set[str] = set() + roles: set[str] = set() + valid = 0 + for approval in approvals: + if not isinstance(approval, dict) or set(approval) != {"principal_id", "issuer", "key_id", "role", "approved_at", "expires_at", "signature"}: + raise CatalogApprovalError("approval has missing or unknown fields") + key_id = approval["key_id"] + reviewer = trusted_reviewers.get(key_id) + if key_id in revoked_key_ids: + raise CatalogApprovalMismatch(f"reviewer key {key_id!r} is revoked") + if reviewer is None: + raise CatalogApprovalMismatch(f"reviewer key {key_id!r} is not trusted") + if approval["principal_id"] != reviewer.principal_id or approval["issuer"] != reviewer.issuer: + raise CatalogApprovalMismatch("approval principal or issuer does not match trusted key") + if reviewer.role is not None and approval["role"] != reviewer.role: + raise CatalogApprovalMismatch("approval role does not match trusted key") + if not isinstance(approval["approved_at"], int) or not isinstance(approval["expires_at"], int) or approval["expires_at"] <= approval["approved_at"]: + raise CatalogApprovalError("approval validity interval is invalid") + if instant < approval["approved_at"] or instant >= approval["expires_at"]: + raise CatalogApprovalMismatch("approval is not currently valid") + try: + reviewer.key.verify(_decode(approval["signature"]), _approval_input(record, approval)) + except (InvalidSignature, ValueError) as exc: + raise CatalogApprovalMismatch("approval signature is invalid") from exc + valid += 1 + principals.add(approval["principal_id"]) + roles.add(approval["role"]) + if policy["distinct_principals"] and len(principals) < threshold: + raise CatalogApprovalMismatch("approval threshold lacks distinct principals") + if policy["distinct_roles"] and len(roles) < threshold: + raise CatalogApprovalMismatch("approval threshold lacks distinct roles") + if valid < threshold: + raise CatalogApprovalMismatch("approval threshold is not satisfied") + return {"verified": True, "valid_approvals": valid, "new_catalog_hash": record["new_catalog_hash"]} diff --git a/tests/unit/test_catalog_approval.py b/tests/unit/test_catalog_approval.py new file mode 100644 index 00000000..3dffbf3c --- /dev/null +++ b/tests/unit/test_catalog_approval.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import copy + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from cmcp_runtime.catalog.approval import ( + PROFILE, + CatalogApprovalError, + CatalogApprovalMismatch, + TrustedReviewer, + digest_json, + sign_approval, + verify_catalog_change, +) + + +def _record() -> tuple[dict, Ed25519PrivateKey, Ed25519PrivateKey]: + first, second = Ed25519PrivateKey.generate(), Ed25519PrivateKey.generate() + policy = {"policy_id": "catalog-policy-v1", "threshold": 2, "distinct_principals": True, "distinct_roles": True} + record = { + "profile": PROFILE, "catalog_id": "gateway-prod", "sequence": 2, + "previous_record_hash": "sha256:" + "1" * 64, + "previous_catalog_hash": "sha256:" + "2" * 64, + "new_catalog_hash": "sha256:" + "3" * 64, + "change_set_digest": digest_json({"added": ["ehr.read"], "removed": []}), + "approval_policy": {**policy, "policy_hash": digest_json(policy)}, + "automated_checks_digest": digest_json({"ci": "passed", "security": "passed"}), + "approvals": [], + } + record["approvals"] = [ + sign_approval(record, {"principal_id": "alice", "issuer": "idp", "key_id": "k1", "role": "security", "approved_at": 100, "expires_at": 200}, first), + sign_approval(record, {"principal_id": "bob", "issuer": "idp", "key_id": "k2", "role": "owner", "approved_at": 100, "expires_at": 200}, second), + ] + return record, first, second + + +def test_two_distinct_valid_approvals_bind_runtime_catalog() -> None: + record, first, second = _record() + result = verify_catalog_change( + record, + {"k1": TrustedReviewer("alice", "idp", first.public_key(), "security"), "k2": TrustedReviewer("bob", "idp", second.public_key(), "owner")}, + runtime_catalog_hash=record["new_catalog_hash"], now=150, + ) + assert result == {"verified": True, "valid_approvals": 2, "new_catalog_hash": record["new_catalog_hash"]} + + +def test_runtime_hash_and_chain_are_bound() -> None: + record, first, second = _record() + trusted = {"k1": TrustedReviewer("alice", "idp", first.public_key()), "k2": TrustedReviewer("alice", "idp", second.public_key())} + with pytest.raises(CatalogApprovalMismatch, match="runtime"): + verify_catalog_change(record, trusted, runtime_catalog_hash="sha256:" + "4" * 64, now=150) + with pytest.raises(CatalogApprovalMismatch, match="next chain"): + verify_catalog_change(record, trusted, runtime_catalog_hash=record["new_catalog_hash"], expected_previous_record_hash="sha256:" + "9" * 64, now=150) + + +def test_tampering_expiry_revocation_and_duplicate_principal_fail() -> None: + record, first, second = _record() + trusted = {"k1": TrustedReviewer("alice", "idp", first.public_key()), "k2": TrustedReviewer("alice", "idp", second.public_key())} + tampered = copy.deepcopy(record) + tampered["new_catalog_hash"] = "sha256:" + "4" * 64 + with pytest.raises(CatalogApprovalMismatch, match="signature"): + verify_catalog_change(tampered, trusted, runtime_catalog_hash=tampered["new_catalog_hash"], now=150) + with pytest.raises(CatalogApprovalMismatch, match="revoked"): + verify_catalog_change(record, trusted, runtime_catalog_hash=record["new_catalog_hash"], revoked_key_ids=frozenset({"k1"}), now=150) + duplicate = copy.deepcopy(record) + duplicate["approvals"][1]["principal_id"] = "alice" + duplicate["approvals"][1] = sign_approval(record, {k: v for k, v in duplicate["approvals"][1].items() if k != "signature"}, second) + with pytest.raises(CatalogApprovalMismatch, match="distinct principals"): + verify_catalog_change(duplicate, trusted, runtime_catalog_hash=record["new_catalog_hash"], now=150) + + +def test_malformed_record_fails_closed() -> None: + record, _, _ = _record() + record["unexpected"] = True + with pytest.raises(CatalogApprovalError, match="unknown"): + verify_catalog_change(record, {}, runtime_catalog_hash=record["new_catalog_hash"], now=150)