From c9f298023b583d4f07284d5b29fc20185406b6fe Mon Sep 17 00:00:00 2001 From: Mohammed Zoheb Shaik Date: Sat, 22 Aug 2026 20:39:38 +0400 Subject: [PATCH] fix(catalog): make the schema the structural authority for approval records schemas/catalog-approval.schema.json shipped in the repository and nowhere else. verify_catalog_change reimplemented structural validation by hand, the two drifted, and the file was absent from the wheel force-include, so an installed distribution 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 record is now validated against the schema before any other check, and verification refuses outright when the schema is missing from the installation, following loader.py rather than falling back to hand-written checks. Duplicated checks are deleted rather than left unreachable: field presence and unknown members on the record, the policy and each approval, digest shapes, integer and boolean types, string emptiness, and the signature alphabet. _require_str and _require_int go with them. What stays is what JSON Schema cannot express: * the runtime catalog hash binding and the caller's chain checkpoints * the policy pin, including that policy_hash covers its own body * reviewer identity, role, revocation and key reuse * the validity interval ordering * the signatures The xfail is now a plain assertion, since the case fails closed. previous_record_hash on the first record in a chain is the all-zero digest, sha256: followed by 64 zeros, written into the schema and the spec doc. It validated against the pattern before and meant nothing. Packaging: the schema is force-included beside catalog-entry.schema.json, and 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. loader.py's _CATALOG_ENTRY_SCHEMA_PATH is now CATALOG_ENTRY_SCHEMA_PATH, since the distribution check imports it. Verified by mutation: dropping the schema call fails 7 tests, accepting a missing schema fails 1, dropping the 64-byte signature check fails 1. Coverage of approval.py is 149 statements with none missing. Signed-off-by: Mohammed Zoheb Shaik --- CHANGELOG.md | 8 ++ docs/spec/catalog-approval-provenance.md | 13 ++++ pyproject.toml | 1 + schemas/catalog-approval.schema.json | 2 +- scripts/verify_python_distribution.py | 9 +++ src/cmcp_runtime/catalog/approval.py | 94 +++++++++++++----------- src/cmcp_runtime/catalog/loader.py | 6 +- tests/unit/test_catalog.py | 2 +- tests/unit/test_catalog_approval.py | 63 +++++++++++++--- 9 files changed, 139 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e9938ad0..d327af12 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/spec/catalog-approval-provenance.md b/docs/spec/catalog-approval-provenance.md index 964eb0c0..49709ad8 100644 --- a/docs/spec/catalog-approval-provenance.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 3b31e593..c0912e0c 100644 --- a/pyproject.toml +++ b/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] diff --git a/schemas/catalog-approval.schema.json b/schemas/catalog-approval.schema.json index 75c8054e..5576550e 100644 --- a/schemas/catalog-approval.schema.json +++ b/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"}, diff --git a/scripts/verify_python_distribution.py b/scripts/verify_python_distribution.py index 26c27d46..43277763 100644 --- a/scripts/verify_python_distribution.py +++ b/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") diff --git a/src/cmcp_runtime/catalog/approval.py b/src/cmcp_runtime/catalog/approval.py index 7c214923..1526e2eb 100644 --- a/src/cmcp_runtime/catalog/approval.py +++ b/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") diff --git a/src/cmcp_runtime/catalog/loader.py b/src/cmcp_runtime/catalog/loader.py index 830bf059..520933b9 100644 --- a/src/cmcp_runtime/catalog/loader.py +++ b/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 diff --git a/tests/unit/test_catalog.py b/tests/unit/test_catalog.py index 8fcaa342..b3d2edef 100644 --- a/tests/unit/test_catalog.py +++ b/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])) diff --git a/tests/unit/test_catalog_approval.py b/tests/unit/test_catalog_approval.py index b5161f17..7f3964b7 100644 --- a/tests/unit/test_catalog_approval.py +++ b/tests/unit/test_catalog_approval.py @@ -3,12 +3,16 @@ import copy import json import pathlib +import tomllib import jsonschema import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cmcp_runtime.catalog import approval as approval_module from cmcp_runtime.catalog.approval import ( + CATALOG_APPROVAL_SCHEMA_PATH, + GENESIS_PREVIOUS_RECORD_HASH, PROFILE, CatalogApprovalError, CatalogApprovalMismatch, @@ -20,8 +24,7 @@ ) CATALOG_ID = "gateway-prod" -GENESIS_PREVIOUS_RECORD_HASH = "sha256:" + "0" * 64 -SCHEMA = json.loads((pathlib.Path(__file__).parents[2] / "schemas" / "catalog-approval.schema.json").read_text()) +SCHEMA = json.loads(CATALOG_APPROVAL_SCHEMA_PATH.read_text()) def _record(threshold: int = 2, distinct: bool = True) -> tuple[dict, Ed25519PrivateKey, Ed25519PrivateKey]: @@ -145,7 +148,7 @@ def test_duplicate_principal_rejected_even_with_surplus_approvals() -> None: def test_malformed_record_fails_closed() -> None: record, _, _ = _record() record["unexpected"] = True - with pytest.raises(CatalogApprovalError, match="unknown"): + with pytest.raises(CatalogApprovalError, match="Additional properties"): _verify(record, {}) @@ -157,13 +160,13 @@ def test_malformed_field_types_fail_closed() -> None: unhashable["approvals"] = [ sign_approval(unhashable, {"principal_id": "alice", "issuer": "idp", "key_id": "k1", "role": {"nested": "obj"}, "approved_at": 100, "expires_at": 200}, first) ] - with pytest.raises(CatalogApprovalError, match="approval.role"): + with pytest.raises(CatalogApprovalError, match="approvals/0/role"): _verify(unhashable, trusted) boolean_times = copy.deepcopy(record) boolean_times["approvals"] = [ sign_approval(boolean_times, {"principal_id": "alice", "issuer": "idp", "key_id": "k1", "role": "security", "approved_at": False, "expires_at": True}, first) ] - with pytest.raises(CatalogApprovalError, match="must be an integer"): + with pytest.raises(CatalogApprovalError, match="approvals/0/(approved_at|expires_at)"): _verify(boolean_times, trusted, now=0) @@ -173,7 +176,7 @@ def test_signature_encoding_is_validated() -> None: for bad in ("not base64!!", "c2hvcnQ", "A"): broken = copy.deepcopy(record) broken["approvals"][0]["signature"] = bad - with pytest.raises(CatalogApprovalError, match="base64url|64 bytes"): + with pytest.raises(CatalogApprovalError, match="base64url|64 bytes|approvals/0/signature"): _verify(broken, trusted) @@ -337,8 +340,8 @@ def test_schema_and_verifier_reject_the_same_records(mutate) -> None: _verify(mutated, _trusted(first, second)) -@pytest.mark.xfail(strict=True, reason="the verifier does not enforce the schema's minimum on timestamps; pending the schema-wiring decision in #533") def test_negative_approved_at_is_rejected_like_the_schema() -> None: + """The divergence #531 could only document: the schema sets a minimum, and now the verifier applies it.""" record, first, second = _record() negative = copy.deepcopy(record) negative["approvals"][0]["approved_at"] = -1 @@ -354,7 +357,7 @@ def test_policy_distinctness_flags_must_be_booleans() -> None: record, first, second = _record() mangled = copy.deepcopy(record) mangled["approval_policy"]["distinct_roles"] = "yes" - with pytest.raises(CatalogApprovalError, match="distinctness flags"): + with pytest.raises(CatalogApprovalError, match="approval_policy/distinct_roles"): _verify(mangled, _trusted(first, second)) @@ -363,19 +366,19 @@ def test_unknown_profile_and_malformed_members_fail_closed() -> None: trusted = _trusted(first, second) wrong_profile = copy.deepcopy(record) wrong_profile["profile"] = "tag:example.com,2026:something-else" - with pytest.raises(CatalogApprovalError, match="profile"): + with pytest.raises(CatalogApprovalError, match="schema violation at profile"): _verify(wrong_profile, trusted) wrong_shape = copy.deepcopy(record) wrong_shape["change_set_digest"] = "not-a-digest" - with pytest.raises(CatalogApprovalError, match="must be a sha256 digest"): + with pytest.raises(CatalogApprovalError, match="schema violation at change_set_digest"): _verify(wrong_shape, trusted) bad_digest = copy.deepcopy(record) bad_digest["change_set_digest"] = "sha256:" + "z" * 64 - with pytest.raises(CatalogApprovalError, match="lowercase hexadecimal"): + with pytest.raises(CatalogApprovalError, match="schema violation at change_set_digest"): _verify(bad_digest, trusted) stray_field = copy.deepcopy(record) stray_field["approvals"][0]["note"] = "looks harmless" - with pytest.raises(CatalogApprovalError, match="approval has missing or unknown fields"): + with pytest.raises(CatalogApprovalError, match="Additional properties"): _verify(stray_field, trusted) @@ -414,3 +417,39 @@ def test_a_roleless_trusted_key_cannot_claim_two_roles() -> None: ] with pytest.raises(CatalogApprovalMismatch, match="reuses a reviewer key"): _verify(record, {"k1": TrustedReviewer("alice", "idp", first.public_key())}) + + +def test_verification_refuses_when_the_schema_is_not_installed(monkeypatch: pytest.MonkeyPatch) -> None: + """Follows loader.py: no structural validation available means no verification.""" + record, first, second = _record() + monkeypatch.setattr(approval_module, "CATALOG_APPROVAL_SCHEMA_PATH", pathlib.Path("no-such-schema.json")) + monkeypatch.setattr(approval_module, "_schema_cache", None) + with pytest.raises(CatalogApprovalError, match="missing from the CMCP installation"): + _verify(record, _trusted(first, second)) + + +def test_every_schema_the_runtime_loads_ships_in_the_wheel() -> None: + """force-include is the only thing putting these next to the code; nothing else notices if it goes.""" + pyproject = tomllib.loads((pathlib.Path(__file__).parents[2] / "pyproject.toml").read_text()) + force_include = pyproject["tool"]["hatch"]["build"]["targets"]["wheel"]["force-include"] + for schema in ("catalog-entry.schema.json", "catalog-approval.schema.json"): + assert force_include.get(f"schemas/{schema}") == f"cmcp_runtime/schemas/{schema}" + + +def test_a_malformed_expected_policy_hash_is_rejected() -> None: + """The caller's pin is not schema-validated, so the verifier still checks its shape.""" + record, first, second = _record() + with pytest.raises(CatalogApprovalError, match="must be a sha256 digest"): + _verify(record, _trusted(first, second), expected_policy_hash="not-a-digest") + with pytest.raises(CatalogApprovalError, match="lowercase hexadecimal"): + _verify(record, _trusted(first, second), expected_policy_hash="sha256:" + "Z" * 64) + + +def test_an_unreadable_schema_fails_closed(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: + record, first, second = _record() + broken = tmp_path / "catalog-approval.schema.json" + broken.write_text("{ not json") + monkeypatch.setattr(approval_module, "CATALOG_APPROVAL_SCHEMA_PATH", broken) + monkeypatch.setattr(approval_module, "_schema_cache", None) + with pytest.raises(CatalogApprovalError, match="cannot load catalog approval schema"): + _verify(record, _trusted(first, second))