feat: add verifiable catalog approval provenance

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.

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}$"}}
+}

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"]}

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)