diff --git a/CHANGELOG.md b/CHANGELOG.md index 00adb4d0..984de821 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **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. + + `catalog_id`, `sequence`, and `previous_catalog_hash` were format checked and then unused, so a record for a different catalog verified against this one. `catalog_id` is now always bound, and `expected_sequence` and `expected_previous_catalog_hash` join `expected_previous_record_hash` as optional checkpoints, optional because they must come from an external pin or transparency receipt. Distinctness now rejects repeats as each approval is read rather than counting distinct values afterwards, which had admitted alice, bob, alice at a threshold of two. String and integer fields are validated before use, so a non-hashable `role` raises `CatalogApprovalError` instead of escaping as `TypeError`, boolean timestamps are rejected as the shipped schema already rejects them, and the signature is checked against the base64url alphabet and for a 64 byte decode, with the decode moved out of the `try` block that was reporting its errors as "signature is invalid". + + The distinctness rules did not close the property on their own, because nothing bound approvals to distinct keys. A threshold of three under a policy that does not require distinct principals was satisfied by one key signing the identical approval three times, and a threshold of two under a policy that does require distinct roles was satisfied by one key signing twice as two different roles, since a `TrustedReviewer` carrying no role lets the record assert whichever role it likes. A reviewer key now counts once per record whatever the policy says: a repeated signature is one approval presented N times, not N approvals. The check runs after the principal and role rules so those keep reporting the more specific cause. + + The four tests that shipped with #519 passed whether or not any of these checks existed. The suite is now 31 tests, each guard verified by deleting it and confirming a test fails. One is xfailed on purpose: the schema sets a minimum of zero on the approval timestamps and the verifier does not, which is left failing until the schema-wiring decision lands. + - **The catalog-approval signing input was not the JCS it claims to be (#517).** `canonical_json` is documented as RFC 8785 compatible and serialized with `ensure_ascii=True`, which is the one thing JCS does not do: it emits an ASCII escape where the standard emits UTF-8. Any record carrying a non-ASCII `principal_id`, `issuer`, `role`, `catalog_id`, or `policy_id` was signed over different bytes than a conforming producer signs, so a record produced anywhere but here failed with an invalid-signature error that points nowhere near the encoding. `sort_keys` was the second divergence, ordering members by code point where JCS orders by UTF-16 code unit. The two disagree for any key outside the BMP, since a surrogate pair leads with `0xD800` and sorts below a BMP character above `0xE000`. Members are now ordered on their UTF-16BE bytes and the output is UTF-8. Values JCS cannot pin down are refused rather than serialized into a signing input that two implementations would read differently: floating point numbers, integers beyond `2**53 - 1`, non-string keys, and unpaired surrogates, each as `CatalogApprovalError` rather than as an escaping `UnicodeEncodeError` or `TypeError`. Approval records carry none of those, so refusing them only closes a door. diff --git a/docs/spec/catalog-approval-provenance.md b/docs/spec/catalog-approval-provenance.md index e9bada70..964eb0c0 100644 --- a/docs/spec/catalog-approval-provenance.md +++ b/docs/spec/catalog-approval-provenance.md @@ -11,9 +11,33 @@ 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. +bootstrap trust. + +The same rule applies to the policy. A record states which policy it followed, +but the verifier is given the expected policy hash and catalog identifier from +its own configuration, and rejects any record citing a different policy. The +record's `policy_hash` must also cover its own policy body, computed over the +policy object with the `policy_hash` field removed. Without this pinning a +single reviewer key could issue a record declaring a threshold of one, so the +M-of-N property depends on the policy being verifier-supplied rather than +record-supplied. + +The verifier rejects unknown fields, revoked or expired keys, invalid +signatures, repeated principals or roles when the policy requires distinctness, +and records whose `new_catalog_hash` differs from the runtime catalog hash. A +reviewer key counts once regardless of policy: a repeated signature is one +approval presented N times, not N approvals, and without that rule a threshold +of N is satisfiable by a single key whenever the policy does not demand +distinct principals, or demands distinct roles while the key's trusted entry +pins no role. +The key identifier is the reviewer key's identity for both of those rules. A +verifier that registers one key under two identifiers has neither: the record +can present it twice toward one threshold, and revoking one identifier leaves +the other usable. `trusted_reviewers` must therefore map distinct identifiers to +distinct keys. +`catalog_id` is always checked. `sequence`, `previous_record_hash`, and +`previous_catalog_hash` are checked when the caller supplies the corresponding +checkpoint, which it must obtain externally as described below. The signing input is RFC 8785 (JCS): UTF-8 output, object members ordered by their UTF-16 code units, and no escaping beyond what ECMAScript `JSON.stringify` diff --git a/src/cmcp_runtime/catalog/approval.py b/src/cmcp_runtime/catalog/approval.py index 0e399f2e..7c214923 100644 --- a/src/cmcp_runtime/catalog/approval.py +++ b/src/cmcp_runtime/catalog/approval.py @@ -20,6 +20,8 @@ # 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.""" @@ -83,17 +85,29 @@ def digest_json(value: Any) -> str: return "sha256:" + hashlib.sha256(canonical_json(value)).hexdigest() +def compute_policy_hash(policy: dict[str, Any]) -> str: + """Digest of the approval policy body, excluding the policy_hash field itself. + + Producers and verifiers must agree on this definition, otherwise the policy a + record claims to follow cannot be pinned to the policy the verifier trusts. + """ + return digest_json({k: v for k, v in policy.items() if k != "policy_hash"}) + + 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") + 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") try: - return base64.urlsafe_b64decode(value + "=" * ((4 - len(value) % 4) % 4)) + raw = base64.urlsafe_b64decode(value + "=" * ((4 - len(value) % 4) % 4)) except (ValueError, TypeError) as exc: raise CatalogApprovalError("signature is not valid base64url") from exc + if len(raw) != 64: + raise CatalogApprovalError("signature must decode to 64 bytes") + return raw def _approval_input(record: dict[str, Any], approval: dict[str, Any]) -> bytes: @@ -119,16 +133,39 @@ 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 + + +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 verify_catalog_change( record: dict[str, Any], trusted_reviewers: dict[str, TrustedReviewer], *, runtime_catalog_hash: str, + expected_policy_hash: str, + expected_catalog_id: str, now: int | None = None, revoked_key_ids: frozenset[str] = frozenset(), + expected_sequence: int | None = None, expected_previous_record_hash: str | None = None, + expected_previous_catalog_hash: str | None = None, ) -> dict[str, Any]: - """Verify policy, chain, reviewer signatures, freshness, and runtime binding.""" + """Verify policy, chain, reviewer signatures, freshness, and runtime binding. + + The approval policy is never taken on the record's word: `expected_policy_hash` + and `expected_catalog_id` come from verifier-side configuration, so a record + 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. + """ if not isinstance(record, dict) or record.get("profile") != PROFILE: raise CatalogApprovalError("unknown or missing catalog approval profile") required = { @@ -138,32 +175,48 @@ def verify_catalog_change( } 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: + 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: + 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") 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 expected_previous_catalog_hash is not None and record["previous_catalog_hash"] != expected_previous_catalog_hash: + raise CatalogApprovalMismatch("previous_catalog_hash does not match the expected checkpoint") 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_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 not isinstance(threshold, int) or isinstance(threshold, bool) or threshold < 1: + 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: 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: @@ -174,21 +227,24 @@ 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") - if not isinstance(approval["approved_at"], int) or not isinstance(approval["expires_at"], int) or approval["expires_at"] <= approval["approved_at"]: + approved_at = _require_int(approval["approved_at"], "approval.approved_at") + if _require_int(approval["expires_at"], "approval.expires_at") <= 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") + signature = _decode(approval["signature"]) try: - reviewer.key.verify(_decode(approval["signature"]), _approval_input(record, approval)) - except (InvalidSignature, ValueError) as exc: + reviewer.key.verify(signature, _approval_input(record, approval)) + except InvalidSignature as exc: raise CatalogApprovalMismatch("approval signature is invalid") from exc + if policy["distinct_principals"] and approval["principal_id"] in principals: + raise CatalogApprovalMismatch("approval set repeats a principal under a distinct-principal policy") + if policy["distinct_roles"] and approval["role"] in roles: + raise CatalogApprovalMismatch("approval set repeats a role under a distinct-role policy") + if key_id in keys_used: + raise CatalogApprovalMismatch("approval set reuses a reviewer key") 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") + keys_used.add(key_id) 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 index 3dffbf3c..b5161f17 100644 --- a/tests/unit/test_catalog_approval.py +++ b/tests/unit/test_catalog_approval.py @@ -1,7 +1,10 @@ from __future__ import annotations import copy +import json +import pathlib +import jsonschema import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey @@ -10,22 +13,27 @@ CatalogApprovalError, CatalogApprovalMismatch, TrustedReviewer, + compute_policy_hash, digest_json, sign_approval, verify_catalog_change, ) +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()) -def _record() -> tuple[dict, Ed25519PrivateKey, Ed25519PrivateKey]: + +def _record(threshold: int = 2, distinct: bool = True) -> tuple[dict, Ed25519PrivateKey, Ed25519PrivateKey]: first, second = Ed25519PrivateKey.generate(), Ed25519PrivateKey.generate() - policy = {"policy_id": "catalog-policy-v1", "threshold": 2, "distinct_principals": True, "distinct_roles": True} + policy = {"policy_id": "catalog-policy-v1", "threshold": threshold, "distinct_principals": distinct, "distinct_roles": distinct} record = { - "profile": PROFILE, "catalog_id": "gateway-prod", "sequence": 2, + "profile": PROFILE, "catalog_id": CATALOG_ID, "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)}, + "approval_policy": {**policy, "policy_hash": compute_policy_hash(policy)}, "automated_checks_digest": digest_json({"ci": "passed", "security": "passed"}), "approvals": [], } @@ -36,43 +44,373 @@ def _record() -> tuple[dict, Ed25519PrivateKey, Ed25519PrivateKey]: return record, first, second +def _trusted(first: Ed25519PrivateKey, second: Ed25519PrivateKey) -> dict[str, TrustedReviewer]: + return { + "k1": TrustedReviewer("alice", "idp", first.public_key(), "security"), + "k2": TrustedReviewer("bob", "idp", second.public_key(), "owner"), + } + + +def _verify(record: dict, trusted: dict[str, TrustedReviewer], **overrides: object) -> dict: + kwargs: dict = { + "runtime_catalog_hash": record["new_catalog_hash"], + "expected_policy_hash": record["approval_policy"]["policy_hash"], + "expected_catalog_id": CATALOG_ID, + "now": 150, + } + kwargs.update(overrides) + return verify_catalog_change(record, trusted, **kwargs) # type: ignore[arg-type] + + 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, - ) + result = _verify(record, _trusted(first, second)) 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())} + trusted = _trusted(first, second) with pytest.raises(CatalogApprovalMismatch, match="runtime"): - verify_catalog_change(record, trusted, runtime_catalog_hash="sha256:" + "4" * 64, now=150) + _verify(record, trusted, runtime_catalog_hash="sha256:" + "4" * 64) 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) + _verify(record, trusted, expected_previous_record_hash="sha256:" + "9" * 64) + + +def test_catalog_identity_and_chain_fields_are_bound() -> None: + """previous_catalog_hash, sequence, and catalog_id must be pinnable, not decorative.""" + record, first, second = _record() + trusted = _trusted(first, second) + with pytest.raises(CatalogApprovalMismatch, match="expected catalog"): + _verify(record, trusted, expected_catalog_id="gateway-dev") + with pytest.raises(CatalogApprovalMismatch, match="sequence number"): + _verify(record, trusted, expected_sequence=7) + with pytest.raises(CatalogApprovalMismatch, match="previous_catalog_hash"): + _verify(record, trusted, expected_previous_catalog_hash="sha256:" + "a" * 64) + assert _verify( + record, trusted, + expected_sequence=2, + expected_previous_record_hash=record["previous_record_hash"], + expected_previous_catalog_hash=record["previous_catalog_hash"], + )["verified"] + + +def test_record_cannot_declare_its_own_policy() -> None: + """A single trusted key must not be able to downgrade the threshold to one.""" + record, first, _ = _record(threshold=1, distinct=False) + record["approvals"] = [record["approvals"][0]] + trusted = {"k1": TrustedReviewer("alice", "idp", first.public_key(), "security")} + two_of_two = {"policy_id": "catalog-policy-v1", "threshold": 2, "distinct_principals": True, "distinct_roles": True} + with pytest.raises(CatalogApprovalMismatch, match="does not trust"): + _verify(record, trusted, expected_policy_hash=compute_policy_hash(two_of_two)) + + +def test_policy_hash_must_cover_the_policy_body() -> None: + record, first, second = _record() + trusted = _trusted(first, second) + forged = copy.deepcopy(record) + forged["approval_policy"]["threshold"] = 1 + with pytest.raises(CatalogApprovalError, match="does not cover"): + _verify(forged, trusted, expected_policy_hash=forged["approval_policy"]["policy_hash"]) 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())} + trusted = _trusted(first, second) 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) + _verify(tampered, trusted, runtime_catalog_hash=tampered["new_catalog_hash"]) 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) + _verify(record, trusted, revoked_key_ids=frozenset({"k1"})) + with pytest.raises(CatalogApprovalMismatch, match="not currently valid"): + _verify(record, trusted, now=10_000) + with pytest.raises(CatalogApprovalMismatch, match="not trusted"): + _verify(record, {"k1": trusted["k1"]}) + + +def test_duplicate_principal_rejected_even_with_surplus_approvals() -> None: + """Distinctness must reject repeats, not merely count distinct values.""" + record, first, second = _record(threshold=2, distinct=True) + trusted = _trusted(first, second) + surplus = copy.deepcopy(record) + trusted["k3"] = TrustedReviewer("alice", "idp", first.public_key(), "security") + surplus["approvals"].append( + sign_approval(surplus, {"principal_id": "alice", "issuer": "idp", "key_id": "k3", "role": "security", "approved_at": 100, "expires_at": 200}, first) + ) + with pytest.raises(CatalogApprovalMismatch, match="repeats a principal"): + _verify(surplus, trusted) 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) + _verify(record, {}) + + +def test_malformed_field_types_fail_closed() -> None: + """Bad field types must surface as CatalogApprovalError, never TypeError.""" + record, first, _ = _record(threshold=1, distinct=False) + trusted = {"k1": TrustedReviewer("alice", "idp", first.public_key())} + unhashable = copy.deepcopy(record) + 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"): + _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"): + _verify(boolean_times, trusted, now=0) + + +def test_signature_encoding_is_validated() -> None: + record, first, second = _record() + trusted = _trusted(first, second) + for bad in ("not base64!!", "c2hvcnQ", "A"): + broken = copy.deepcopy(record) + broken["approvals"][0]["signature"] = bad + with pytest.raises(CatalogApprovalError, match="base64url|64 bytes"): + _verify(broken, trusted) + + +def _resign(record: dict, keys: dict[str, Ed25519PrivateKey]) -> dict: + """Re-sign every approval after the record body changed.""" + record["approvals"] = [ + sign_approval(record, {k: v for k, v in approval.items() if k != "signature"}, keys[approval["key_id"]]) + for approval in record["approvals"] + ] + return record + + +def test_threshold_shortfall_is_rejected() -> None: + """One approval must not satisfy a 2-of-N policy.""" + record, first, second = _record(threshold=2) + short = copy.deepcopy(record) + short["approvals"] = [short["approvals"][0]] + with pytest.raises(CatalogApprovalMismatch, match="threshold is not satisfied"): + _verify(short, _trusted(first, second)) + + +def test_repeated_role_rejected_under_distinct_role_policy() -> None: + """Distinct principals sharing one role must not satisfy a distinct-role policy.""" + record, first, second = _record(threshold=2, distinct=True) + trusted = _trusted(first, second) + trusted["k2"] = TrustedReviewer("bob", "idp", second.public_key(), "security") + shared = copy.deepcopy(record) + shared["approvals"][1]["role"] = "security" + _resign(shared, {"k1": first, "k2": second}) + with pytest.raises(CatalogApprovalMismatch, match="repeats a role"): + _verify(shared, trusted) + + +def test_approval_identity_must_match_the_trusted_key() -> None: + """principal_id, issuer, and role are claims about the key, not free text.""" + record, first, second = _record() + wrong_principal = _trusted(first, second) | {"k1": TrustedReviewer("carol", "idp", first.public_key(), "security")} + with pytest.raises(CatalogApprovalMismatch, match="principal or issuer"): + _verify(record, wrong_principal) + wrong_issuer = _trusted(first, second) | {"k1": TrustedReviewer("alice", "other-idp", first.public_key(), "security")} + with pytest.raises(CatalogApprovalMismatch, match="principal or issuer"): + _verify(record, wrong_issuer) + wrong_role = _trusted(first, second) | {"k1": TrustedReviewer("alice", "idp", first.public_key(), "owner")} + with pytest.raises(CatalogApprovalMismatch, match="role does not match"): + _verify(record, wrong_role) + + +def test_validity_interval_boundaries() -> None: + """approved_at is inclusive, expires_at is exclusive, and the interval must be ordered.""" + record, first, second = _record() + trusted = _trusted(first, second) + assert _verify(record, trusted, now=100)["verified"] + with pytest.raises(CatalogApprovalMismatch, match="not currently valid"): + _verify(record, trusted, now=99) + with pytest.raises(CatalogApprovalMismatch, match="not currently valid"): + _verify(record, trusted, now=200) + inverted = copy.deepcopy(record) + for approval in inverted["approvals"]: + approval["approved_at"], approval["expires_at"] = 200, 100 + _resign(inverted, {"k1": first, "k2": second}) + with pytest.raises(CatalogApprovalError, match="validity interval is invalid"): + _verify(inverted, trusted, now=150) + + +def test_genesis_record_is_representable() -> None: + """The first record in a chain has no predecessor, so previous_record_hash is all zeroes. + + The convention is asserted here rather than in the schema, which still demands a + previous_record_hash without defining what a sequence 1 record puts there. + """ + record, first, second = _record() + genesis = copy.deepcopy(record) + genesis["sequence"] = 1 + genesis["previous_record_hash"] = GENESIS_PREVIOUS_RECORD_HASH + _resign(genesis, {"k1": first, "k2": second}) + jsonschema.validate(genesis, SCHEMA) + assert _verify( + genesis, + _trusted(first, second), + expected_sequence=1, + expected_previous_record_hash=GENESIS_PREVIOUS_RECORD_HASH, + )["verified"] + + +def _unknown_field(record: dict) -> None: + record["unexpected"] = True + + +def _zero_sequence(record: dict) -> None: + record["sequence"] = 0 + + +def _bool_sequence(record: dict) -> None: + record["sequence"] = True + + +def _zero_threshold(record: dict) -> None: + record["approval_policy"]["threshold"] = 0 + + +def _empty_catalog_id(record: dict) -> None: + record["catalog_id"] = "" + + +def _missing_chain_field(record: dict) -> None: + del record["previous_record_hash"] + + +def _missing_policy_field(record: dict) -> None: + del record["approval_policy"]["distinct_roles"] + + +def _no_approvals(record: dict) -> None: + record["approvals"] = [] + + +def _bool_timestamps(record: dict) -> None: + record["approvals"][0]["approved_at"] = False + record["approvals"][0]["expires_at"] = True + + +def _bad_signature_alphabet(record: dict) -> None: + record["approvals"][0]["signature"] = "not base64!!" + + +def _malformed_digest(record: dict) -> None: + record["new_catalog_hash"] = "sha256:" + "z" * 64 + + +@pytest.mark.parametrize( + "mutate", + [ + _unknown_field, + _zero_sequence, + _bool_sequence, + _zero_threshold, + _empty_catalog_id, + _missing_chain_field, + _missing_policy_field, + _no_approvals, + _bool_timestamps, + _bad_signature_alphabet, + _malformed_digest, + ], +) +def test_schema_and_verifier_reject_the_same_records(mutate) -> None: + """Anything the shipped schema rejects the verifier must reject too. + + The verifier reimplements structural validation by hand and does not load the + schema, so the two can only be kept in step by asserting it. + """ + record, first, second = _record() + jsonschema.validate(record, SCHEMA) + assert _verify(record, _trusted(first, second))["verified"] + + mutated = copy.deepcopy(record) + mutate(mutated) + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate(mutated, SCHEMA) + with pytest.raises(CatalogApprovalError): + _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: + record, first, second = _record() + negative = copy.deepcopy(record) + negative["approvals"][0]["approved_at"] = -1 + _resign(negative, {"k1": first, "k2": second}) + with pytest.raises(jsonschema.ValidationError): + jsonschema.validate(negative, SCHEMA) + with pytest.raises(CatalogApprovalError): + _verify(negative, _trusted(first, second)) + + +def test_policy_distinctness_flags_must_be_booleans() -> None: + """A truthy string must not stand in for a policy flag the verifier branches on.""" + record, first, second = _record() + mangled = copy.deepcopy(record) + mangled["approval_policy"]["distinct_roles"] = "yes" + with pytest.raises(CatalogApprovalError, match="distinctness flags"): + _verify(mangled, _trusted(first, second)) + + +def test_unknown_profile_and_malformed_members_fail_closed() -> None: + record, first, second = _record() + trusted = _trusted(first, second) + wrong_profile = copy.deepcopy(record) + wrong_profile["profile"] = "tag:example.com,2026:something-else" + with pytest.raises(CatalogApprovalError, match="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"): + _verify(wrong_shape, trusted) + bad_digest = copy.deepcopy(record) + bad_digest["change_set_digest"] = "sha256:" + "z" * 64 + with pytest.raises(CatalogApprovalError, match="lowercase hexadecimal"): + _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"): + _verify(stray_field, trusted) + + +def test_one_key_cannot_satisfy_a_threshold_by_signing_twice() -> None: + """A repeated signature is one approval pasted N times, not N approvals. + + Both cases verified before this guard existed: a policy that does not demand + distinct principals counted the same approval three times, and a policy that + demands distinct roles counted one key twice when its trusted entry pins no role. + """ + record, first, _ = _record(threshold=3, distinct=False) + repeated = copy.deepcopy(record) + repeated["approvals"] = [copy.deepcopy(repeated["approvals"][0]) for _ in range(3)] + trusted = {"k1": TrustedReviewer("alice", "idp", first.public_key(), "security")} + with pytest.raises(CatalogApprovalMismatch, match="reuses a reviewer key"): + _verify(repeated, trusted) + + +def test_a_roleless_trusted_key_cannot_claim_two_roles() -> None: + """distinct_roles must not be satisfiable by one key asserting two role strings.""" + policy = {"policy_id": "catalog-policy-v1", "threshold": 2, "distinct_principals": False, "distinct_roles": True} + first = Ed25519PrivateKey.generate() + record = { + "profile": PROFILE, "catalog_id": CATALOG_ID, "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": compute_policy_hash(policy)}, + "automated_checks_digest": digest_json({"ci": "passed"}), + "approvals": [], + } + record["approvals"] = [ + sign_approval(record, {"principal_id": "alice", "issuer": "idp", "key_id": "k1", "role": role, "approved_at": 100, "expires_at": 200}, first) + for role in ("security", "owner") + ] + with pytest.raises(CatalogApprovalMismatch, match="reuses a reviewer key"): + _verify(record, {"k1": TrustedReviewer("alice", "idp", first.public_key())}) diff --git a/tests/unit/test_catalog_canonical_json.py b/tests/unit/test_catalog_canonical_json.py index bed566ad..6329690f 100644 --- a/tests/unit/test_catalog_canonical_json.py +++ b/tests/unit/test_catalog_canonical_json.py @@ -10,6 +10,7 @@ CatalogApprovalError, TrustedReviewer, canonical_json, + compute_policy_hash, digest_json, sign_approval, verify_catalog_change, @@ -70,7 +71,7 @@ def test_a_non_ascii_reviewer_identity_signs_and_verifies() -> None: "previous_catalog_hash": "sha256:" + "2" * 64, "new_catalog_hash": "sha256:" + "3" * 64, "change_set_digest": digest_json({"added": ["dossier.lecture"], "removed": []}), - "approval_policy": {**policy, "policy_hash": digest_json(policy)}, + "approval_policy": {**policy, "policy_hash": compute_policy_hash(policy)}, "automated_checks_digest": digest_json({"ci": "réussi"}), "approvals": [], } @@ -81,6 +82,9 @@ def test_a_non_ascii_reviewer_identity_signs_and_verifies() -> None: result = verify_catalog_change( record, {"k1": TrustedReviewer("josé", "idp", key.public_key(), "sécurité")}, - runtime_catalog_hash=record["new_catalog_hash"], now=150, + runtime_catalog_hash=record["new_catalog_hash"], + expected_policy_hash=record["approval_policy"]["policy_hash"], + expected_catalog_id=record["catalog_id"], + now=150, ) assert result["verified"]