fix(catalog): make the schema the structural authority for approval records (#533)

CHANGELOG.md

@@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **The catalog-approval schema shipped in the repository and nowhere else, so nothing validated against it (#533, follow-up to #531).** `schemas/catalog-approval.schema.json` was added by #519 and never loaded: `verify_catalog_change` reimplemented structural validation by hand, the two drifted, and the schema was absent from `[tool.hatch.build.targets.wheel.force-include]` so an installed wheel did not carry it at all. #531 could only document one divergence, a negative `approved_at` that the schema rejects and the verifier accepted, as a strict xfail.
+
+  The schema is now the structural authority. `verify_catalog_change` loads it and validates the record before any other check, and refuses to verify when it is missing from the installation, exactly as `loader.py` refuses to load a catalog without `catalog-entry.schema.json`. The hand-written checks that duplicated it are deleted rather than left unreachable: field presence and unknown members, digest shapes, integer and boolean types, string emptiness, and the signature alphabet. What stays in code is what JSON Schema cannot express, the runtime hash binding, the policy pin, reviewer identity and key rules, the validity interval ordering, and the signatures. The xfail is gone because the case now fails closed.
+
+  The schema is force-included in the wheel alongside the catalog entry schema, and `scripts/verify_python_distribution.py` checks both resolve to a file inside the installed distribution, so a wheel that drops one fails the release smoke test rather than a verifier at runtime. A unit test asserts the force-include mapping, since that line is the only thing putting these files next to the code.
+
+  `previous_record_hash` on the first record in a chain is the all-zero digest, `sha256:` followed by 64 zeros, stated in both the schema and the spec doc. It validated against the digest pattern before and meant nothing, which is the worst combination: representable and unspecified.
+
 - **A single trusted reviewer key could approve any catalog change, under any policy (#517, follow-up to #519).** `verify_catalog_change` read `threshold`, `distinct_principals`, and `distinct_roles` out of the record it was verifying, and checked `policy_hash` for digest shape only: never recomputed over the policy body, never compared to anything. A record declaring `threshold: 1` with arbitrary bytes in `policy_hash` verified on one signature, so the M-of-N property the module advertises was unenforced. This is the failure the spec doc already forbids for keys, "no record-embedded key can bootstrap trust", applied to the policy instead. M is only meaningful when it comes from verifier-side configuration.
 
   `expected_policy_hash` and `expected_catalog_id` are now required arguments. A record's `policy_hash` must cover its own policy body, which `compute_policy_hash` defines so that producers and verifiers agree on it, and must equal the policy the verifier was configured with. Neither can be defaulted without falling back to the record's own claim, so the signature change is deliberate; nothing calls this module yet.

docs/spec/catalog-approval-provenance.md

@@ -48,6 +48,19 @@ differently: floating point numbers, integers outside the exact range of an IEEE
 754 double, non-string object keys, and unpaired surrogates are rejected as
 malformed. Approval records carry none of them.
 
+The schema is the structural authority. `verify_catalog_change` loads
+`schemas/catalog-approval.schema.json` and validates the record against it before
+any other check, and refuses to verify at all when the schema is missing from the
+installation, as the catalog loader does for `catalog-entry.schema.json`. The
+checks the verifier keeps in code are the ones JSON Schema cannot express: the
+runtime hash binding, the policy pin, reviewer identity and key rules, and the
+signatures.
+
+The first record in a chain has no predecessor, and the schema cannot express
+absence for a required digest. That record sets `previous_record_hash` to the
+all-zero digest, `sha256:` followed by 64 zeros, so no predecessor is
+distinguishable from a real chain link rather than left to producer convention.
+
 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

pyproject.toml

@@ -81,6 +81,7 @@ packages = ["src/cmcp_runtime", "src/cmcp_verify"]
 
 [tool.hatch.build.targets.wheel.force-include]
 "schemas/catalog-entry.schema.json" = "cmcp_runtime/schemas/catalog-entry.schema.json"
+"schemas/catalog-approval.schema.json" = "cmcp_runtime/schemas/catalog-approval.schema.json"
 
 
 [tool.pytest.ini_options]

schemas/catalog-approval.schema.json

@@ -9,7 +9,7 @@
     "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_record_hash": {"$ref": "#/$defs/digest", "description": "Digest of the preceding record. The first record in a chain has no predecessor and uses the all-zero digest, \"sha256:\" followed by 64 zeros, so that no predecessor is distinguishable from a real chain link."},
     "previous_catalog_hash": {"$ref": "#/$defs/digest"},
     "new_catalog_hash": {"$ref": "#/$defs/digest"},
     "change_set_digest": {"$ref": "#/$defs/digest"},

scripts/verify_python_distribution.py

@@ -9,6 +9,8 @@
 
 import cmcp_runtime
 import cmcp_verify
+from cmcp_runtime.catalog.approval import CATALOG_APPROVAL_SCHEMA_PATH
+from cmcp_runtime.catalog.loader import CATALOG_ENTRY_SCHEMA_PATH
 from cmcp_runtime.config import Config
 
 
@@ -36,6 +38,13 @@ def main() -> None:
                 f"smoke test imported checkout source {module_path}, not the distribution"
             )
 
+    for schema_path in (CATALOG_ENTRY_SCHEMA_PATH, CATALOG_APPROVAL_SCHEMA_PATH):
+        resolved = schema_path.resolve()
+        if not resolved.is_file():
+            raise SystemExit(f"schema {resolved} is missing from the distribution")
+        if resolved.is_relative_to(forbidden_root):
+            raise SystemExit(f"schema resolved to checkout source {resolved}, not the distribution")
+
     config = Config()
     if config.max_response_size_bytes <= 0:
         raise SystemExit("installed Config produced an invalid response-size bound")

src/cmcp_runtime/catalog/approval.py

@@ -7,8 +7,10 @@
 import json
 import time
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
+import jsonschema
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric.ed25519 import (
     Ed25519PrivateKey,
@@ -17,11 +19,21 @@
 
 PROFILE = "tag:agentrust-io.com,2026:cmcp-catalog-approval-v1"
 
+_PACKAGED_SCHEMA_PATH = Path(__file__).parent.parent / "schemas" / "catalog-approval.schema.json"
+_SOURCE_SCHEMA_PATH = (
+    Path(__file__).parent.parent.parent.parent / "schemas" / "catalog-approval.schema.json"
+)
+CATALOG_APPROVAL_SCHEMA_PATH = (
+    _PACKAGED_SCHEMA_PATH if _PACKAGED_SCHEMA_PATH.exists() else _SOURCE_SCHEMA_PATH
+)
+
+# The first record in a chain has no predecessor. The schema cannot express "absent"
+# for a required digest, so the convention is the all-zero one.
+GENESIS_PREVIOUS_RECORD_HASH = "sha256:" + "0" * 64
+
 # RFC 8785 numbers are IEEE 754 doubles, so integers stay exact only to 2**53 - 1.
 _MAX_EXACT_INT = 2**53 - 1
 
-_B64URL = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_")
-
 
 class CatalogApprovalError(ValueError):
     """The detached approval record is malformed or cannot be trusted."""
@@ -98,9 +110,8 @@ 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 or any(c not in _B64URL for c in value):
-        raise CatalogApprovalError("signature must be an unpadded base64url string")
+def _decode(value: str) -> bytes:
+    """Decode a signature the schema has already constrained to unpadded base64url."""
     try:
         raw = base64.urlsafe_b64decode(value + "=" * ((4 - len(value) % 4) % 4))
     except (ValueError, TypeError) as exc:
@@ -133,16 +144,36 @@ def _require_digest(value: Any, field: str) -> str:
     return value
 
 
-def _require_str(value: Any, field: str) -> str:
-    if not isinstance(value, str) or not value:
-        raise CatalogApprovalError(f"{field} must be a non-empty string")
-    return value
+_schema_cache: dict[str, Any] | None = None
 
 
-def _require_int(value: Any, field: str) -> int:
-    if not isinstance(value, int) or isinstance(value, bool):
-        raise CatalogApprovalError(f"{field} must be an integer")
-    return value
+def _approval_schema() -> dict[str, Any]:
+    """Return the record schema, refusing to verify without it.
+
+    `loader.py` refuses to load a catalog when its schema is missing from the
+    installation rather than validating structure by hand, and an approval record
+    carries more weight than a catalog entry, not less.
+    """
+    global _schema_cache
+    if _schema_cache is None:
+        if not CATALOG_APPROVAL_SCHEMA_PATH.is_file():
+            raise CatalogApprovalError(
+                "catalog approval schema is missing from the CMCP installation; "
+                "refusing to verify a record without structural validation"
+            )
+        try:
+            _schema_cache = dict(json.loads(CATALOG_APPROVAL_SCHEMA_PATH.read_text()))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise CatalogApprovalError(f"cannot load catalog approval schema: {exc}") from exc
+    return _schema_cache
+
+
+def _validate_against_schema(record: Any) -> None:
+    try:
+        jsonschema.validate(record, _approval_schema())
+    except jsonschema.ValidationError as exc:
+        where = "/".join(str(part) for part in exc.absolute_path) or "record"
+        raise CatalogApprovalError(f"schema violation at {where}: {exc.message}") from exc
 
 
 def verify_catalog_change(
@@ -165,21 +196,13 @@ def verify_catalog_change(
     cannot declare its own threshold or distinctness rules. The chain checkpoints
     stay optional because they must come from an external pin or transparency
     receipt, which the record itself cannot supply.
+
+    Structure is the schema's to decide. The checks below cover only what JSON
+    Schema cannot express: the runtime hash binding, the policy pin, reviewer
+    identity and key rules, and the signatures.
     """
-    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 _require_int(record["sequence"], "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 _require_str(record["catalog_id"], "catalog_id") != expected_catalog_id:
+    _validate_against_schema(record)
+    if record["catalog_id"] != expected_catalog_id:
         raise CatalogApprovalMismatch("record does not apply to the expected catalog")
     if expected_sequence is not None and record["sequence"] != expected_sequence:
         raise CatalogApprovalMismatch("record is not the expected sequence number")
@@ -191,32 +214,20 @@ def verify_catalog_change(
         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_str(policy["policy_id"], "approval_policy.policy_id")
-    _require_digest(policy["policy_hash"], "approval_policy.policy_hash")
-    if not isinstance(policy["distinct_principals"], bool) or not isinstance(policy["distinct_roles"], bool):
-        raise CatalogApprovalError("approval_policy distinctness flags must be booleans")
     threshold = policy["threshold"]
-    if _require_int(threshold, "approval threshold") < 1:
-        raise CatalogApprovalError("approval threshold must be a positive integer")
     if compute_policy_hash(policy) != policy["policy_hash"]:
         raise CatalogApprovalError("approval_policy.policy_hash does not cover the policy body")
     if policy["policy_hash"] != _require_digest(expected_policy_hash, "expected_policy_hash"):
         raise CatalogApprovalMismatch("record cites a policy the verifier does not trust")
     instant = int(time.time()) if now is None else now
     approvals = record["approvals"]
-    if not isinstance(approvals, list) or len(approvals) < threshold:
+    if len(approvals) < threshold:
         raise CatalogApprovalMismatch("approval threshold is not satisfied")
     principals: set[str] = set()
     roles: set[str] = set()
     keys_used: 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")
-        for field in ("principal_id", "issuer", "key_id", "role"):
-            _require_str(approval[field], f"approval.{field}")
         key_id = approval["key_id"]
         reviewer = trusted_reviewers.get(key_id)
         if key_id in revoked_key_ids:
@@ -227,8 +238,7 @@ def verify_catalog_change(
             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")
-        approved_at = _require_int(approval["approved_at"], "approval.approved_at")
-        if _require_int(approval["expires_at"], "approval.expires_at") <= approved_at:
+        if 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")

src/cmcp_runtime/catalog/loader.py

@@ -26,7 +26,7 @@
 _SOURCE_ENTRY_SCHEMA_PATH = (
     Path(__file__).parent.parent.parent.parent / "schemas" / "catalog-entry.schema.json"
 )
-_CATALOG_ENTRY_SCHEMA_PATH = (
+CATALOG_ENTRY_SCHEMA_PATH = (
     _PACKAGED_ENTRY_SCHEMA_PATH
     if _PACKAGED_ENTRY_SCHEMA_PATH.exists()
     else _SOURCE_ENTRY_SCHEMA_PATH
@@ -197,13 +197,13 @@ def _catalog_hash(raw_entries: list[dict[str, Any]]) -> str:
 
 
 def _load_entry_schema() -> dict[str, Any]:
-    if not _CATALOG_ENTRY_SCHEMA_PATH.is_file():
+    if not CATALOG_ENTRY_SCHEMA_PATH.is_file():
         raise ConfigError(
             "Catalog entry schema is missing from the CMCP installation; "
             "refusing to load a catalog without structural validation"
         )
     try:
-        return dict(json.loads(_CATALOG_ENTRY_SCHEMA_PATH.read_text()))
+        return dict(json.loads(CATALOG_ENTRY_SCHEMA_PATH.read_text()))
     except (OSError, json.JSONDecodeError) as exc:
         raise ConfigError(f"Cannot load catalog entry schema: {exc}") from exc
 

tests/unit/test_catalog.py

@@ -68,7 +68,7 @@ def _write(entries: list) -> str:
 
 def test_missing_catalog_schema_fails_closed(catalog_file, tmp_path, monkeypatch):
     missing_schema = tmp_path / "missing-catalog-entry.schema.json"
-    monkeypatch.setattr(catalog_loader, "_CATALOG_ENTRY_SCHEMA_PATH", missing_schema)
+    monkeypatch.setattr(catalog_loader, "CATALOG_ENTRY_SCHEMA_PATH", missing_schema)
 
     with pytest.raises(ConfigError, match="schema is missing"):
         load_catalog(catalog_file([ENTRY_1]))