diff --git a/src/cmcp_runtime/catalog/approval.py b/src/cmcp_runtime/catalog/approval.py index ea0c71e4..8b1234c6 100644 --- a/src/cmcp_runtime/catalog/approval.py +++ b/src/cmcp_runtime/catalog/approval.py @@ -5,6 +5,7 @@ import base64 import hashlib import json +import math from dataclasses import dataclass from pathlib import Path from typing import Any @@ -68,7 +69,10 @@ def _utf16_order(key: Any) -> bytes: def _canonical_members(value: Any) -> Any: """Rebuild containers in canonical order, refusing what JCS cannot pin down.""" if isinstance(value, dict): - return {k: _canonical_members(v) for k, v in sorted(value.items(), key=lambda kv: _utf16_order(kv[0]))} + return { + k: _canonical_members(v) + for k, v in sorted(value.items(), key=lambda kv: _utf16_order(kv[0])) + } if isinstance(value, list): return [_canonical_members(item) for item in value] if isinstance(value, float): @@ -85,7 +89,9 @@ def canonical_json(value: Any) -> bytes: `ensure_ascii` would put ASCII escapes where an interoperating producer puts UTF-8 bytes, giving two different signing inputs for one record. """ - text = json.dumps(_canonical_members(value), ensure_ascii=False, allow_nan=False, separators=(",", ":")) + text = json.dumps( + _canonical_members(value), ensure_ascii=False, allow_nan=False, separators=(",", ":") + ) try: return text.encode("utf-8") except UnicodeEncodeError as exc: @@ -111,12 +117,14 @@ def _b64(value: bytes) -> str: def _decode(value: str) -> bytes: """Decode a signature the schema has already constrained to unpadded base64url.""" + if not isinstance(value, str) or len(value) % 4 == 1: + raise CatalogApprovalError("signature is not valid base64url") try: 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") + if len(raw) != 64 or _b64(raw) != value: + raise CatalogApprovalError("signature must be canonical 64-byte base64url") return raw @@ -160,9 +168,35 @@ def _approval_schema() -> dict[str, Any]: "catalog approval schema is missing from the CMCP installation; " "refusing to verify a record without structural validation" ) + + def _no_dup_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + out: dict[str, Any] = {} + for k, v in pairs: + if k in out: + raise ValueError(f"duplicate JSON object key: {k!r}") + out[k] = v + return out + + def _no_nonfinite(token: str) -> Any: + raise ValueError(f"non-finite JSON constant: {token}") + + def _finite_float(token: str) -> float: + val = float(token) + if not math.isfinite(val): + raise ValueError(f"non-finite JSON number: {token}") + return val + try: - _schema_cache = dict(json.loads(CATALOG_APPROVAL_SCHEMA_PATH.read_text())) - except (OSError, json.JSONDecodeError) as exc: + loaded = json.loads( + CATALOG_APPROVAL_SCHEMA_PATH.read_text(encoding="utf-8"), + object_pairs_hook=_no_dup_pairs, + parse_constant=_no_nonfinite, + parse_float=_finite_float, + ) + if not isinstance(loaded, dict): + raise ValueError("catalog approval schema must be a JSON object") + _schema_cache = dict(loaded) + except (OSError, json.JSONDecodeError, ValueError, TypeError) as exc: raise CatalogApprovalError(f"cannot load catalog approval schema: {exc}") from exc return _schema_cache @@ -212,10 +246,18 @@ def verify_catalog_change( 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: + 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 ( + 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") @@ -241,7 +283,10 @@ def verify_catalog_change( 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: + 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") @@ -256,13 +301,21 @@ def verify_catalog_change( 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") + 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") + 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"]) keys_used.add(key_id) - return {"verified": True, "valid_approvals": valid, "new_catalog_hash": record["new_catalog_hash"]} + return { + "verified": True, + "valid_approvals": valid, + "new_catalog_hash": record["new_catalog_hash"], + } diff --git a/src/cmcp_runtime/policy/bundle.py b/src/cmcp_runtime/policy/bundle.py index 1a87af3e..999561e3 100644 --- a/src/cmcp_runtime/policy/bundle.py +++ b/src/cmcp_runtime/policy/bundle.py @@ -9,6 +9,8 @@ import io import json import logging +import math +import re import threading import time from collections.abc import Iterable, Mapping @@ -72,13 +74,75 @@ class PolicyBundle: signing_key_id: str | None = None +_B64URL_RE = re.compile(r"^[A-Za-z0-9_-]+$") + + +def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, value in pairs: + if key in out: + raise ValueError(f"duplicate JSON object key: {key!r}") + out[key] = value + return out + + +class _InvalidRevocationJson: + """Sentinel marking a revocation statement containing duplicate keys or non-finite values.""" + + +_INVALID_REVOCATION_JSON = _InvalidRevocationJson() + + +def _revocation_object_pairs(pairs: list[tuple[str, Any]]) -> Any: + out: dict[str, Any] = {} + for key, value in pairs: + if key in out or value is _INVALID_REVOCATION_JSON: + return _INVALID_REVOCATION_JSON + pending = [value] + while pending: + item = pending.pop() + if item is _INVALID_REVOCATION_JSON: + return _INVALID_REVOCATION_JSON + if isinstance(item, list): + pending.extend(item) + out[key] = value + return out + + +def _revocation_nonfinite_constant(_value: str) -> Any: + return _INVALID_REVOCATION_JSON + + +def _revocation_parse_float(token: str) -> Any: + val = float(token) + if not math.isfinite(val): + return _INVALID_REVOCATION_JSON + return val + + +def _no_nonfinite_constant(value: str) -> Any: + raise ValueError(f"non-finite JSON constant is not allowed: {value}") + + +def _finite_float(token: str) -> float: + val = float(token) + if not math.isfinite(val): + raise ValueError(f"non-finite JSON number is not allowed: {token}") + return val + + def _sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _b64url_decode(value: str) -> bytes: + if not isinstance(value, str) or not _B64URL_RE.fullmatch(value) or len(value) % 4 == 1: + raise ValueError("invalid canonical base64url") padding = 4 - (len(value) % 4) - return base64.urlsafe_b64decode(value + ("=" * padding if padding != 4 else "")) + raw = base64.urlsafe_b64decode(value + ("=" * padding if padding != 4 else "")) + if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value: + raise ValueError("non-canonical base64url encoding") + return raw def signing_pre_image(bundle_hash: str) -> bytes: @@ -142,9 +206,7 @@ class PolicySigningKeys: def __init__(self, current: bytes, successor: bytes | None = None) -> None: if successor is not None and successor == current: - raise ConfigError( - "the successor policy signing key must differ from the current one" - ) + raise ConfigError("the successor policy signing key must differ from the current one") self._current: bytes | None = current self._successor = successor self._revoked: list[bytes] = [] @@ -231,8 +293,13 @@ def apply_file(self, bundle_path: str) -> list[str]: if not path.is_file(): return [] try: - statements = json.loads(path.read_text()) - except (json.JSONDecodeError, OSError) as exc: + statements = json.loads( + path.read_text(encoding="utf-8"), + object_pairs_hook=_revocation_object_pairs, + parse_constant=_revocation_nonfinite_constant, + parse_float=_revocation_parse_float, + ) + except (json.JSONDecodeError, ValueError, OSError) as exc: logger.warning("POLICY_KEY_REVOCATION_INVALID: cannot read %s: %s", path, exc) return [] if not isinstance(statements, list): @@ -246,9 +313,7 @@ def apply_file(self, bundle_path: str) -> list[str]: try: revoked = self.apply(statement) except PolicyKeyRevocationInvalid as exc: - logger.warning( - "POLICY_KEY_REVOCATION_INVALID: %s (%s)", exc, exc.detail or "" - ) + logger.warning("POLICY_KEY_REVOCATION_INVALID: %s (%s)", exc, exc.detail or "") continue if revoked is not None: applied.append(revoked) @@ -262,9 +327,18 @@ def parse_bundle_version(version: str) -> tuple[int, ...]: orderable. A version that cannot be parsed is refused rather than treated as equal-or-newer, because "unparseable" must not be a way past the check. """ - parts = version.split(".") + if ( + not isinstance(version, str) + or not version + or not all(p.isascii() and p.isdecimal() for p in version.split(".")) + ): + raise ConfigError( + f"policy manifest version {version!r} is not a dotted sequence of integers; " + "a runtime policy change is authorised by an increasing version, so the " + "version must be orderable" + ) try: - return tuple(int(p) for p in parts) + return tuple(int(p) for p in version.split(".")) except ValueError as exc: raise ConfigError( f"policy manifest version {version!r} is not a dotted sequence of integers; " @@ -339,8 +413,7 @@ def _canonical_bundle_hash( }) """ policy_hashes = { - name: _sha256_hex(content.encode()) - for name, content in sorted(policy_files.items()) + name: _sha256_hex(content.encode()) for name, content in sorted(policy_files.items()) } hashed_manifest = {k: v for k, v in manifest.items() if k not in _UNHASHED_MANIFEST_KEYS} # docs/spec/cedar-policy.md section 1 defines canonical_json as RFC 8785 @@ -440,9 +513,16 @@ def load_policy_bundle( if not manifest_path.exists(): raise ConfigError(f"Policy bundle missing manifest.json in {bundle_path}") try: - raw_manifest: dict[str, Any] = json.loads(manifest_path.read_text()) - except (json.JSONDecodeError, OSError) as exc: + raw_manifest: dict[str, Any] = json.loads( + manifest_path.read_text(encoding="utf-8"), + object_pairs_hook=_no_duplicate_keys, + parse_constant=_no_nonfinite_constant, + parse_float=_finite_float, + ) + except (json.JSONDecodeError, ValueError, OSError) as exc: raise ConfigError(f"Cannot load manifest.json: {exc}") from exc + if not isinstance(raw_manifest, dict): + raise ConfigError("manifest.json must be a JSON object") required = {"version", "authored_at", "author_identity", "commit_sha"} missing = required - raw_manifest.keys() @@ -503,9 +583,7 @@ def load_policy_bundle( if signing_key is not None: # POLICY-004. After the hash, because the signature is over the hash: a # signature can only mean anything once we know what was hashed. - verify_bundle_signature( - raw_manifest, f"sha256:{computed}", signing_key, revoked_keys - ) + verify_bundle_signature(raw_manifest, f"sha256:{computed}", signing_key, revoked_keys) # Refuse a version we cannot order, even on first load. Discovering at the # first reload that the running bundle's version was never comparable is # worse than refusing to start with it. @@ -614,8 +692,7 @@ def _check_not_a_downgrade(self, new_bundle: PolicyBundle) -> None: raise PolicySignatureInvalid( "policy bundle version did not increase; refusing a possible downgrade", detail=( - f"running={self._bundle.manifest.version} " - f"offered={new_bundle.manifest.version}" + f"running={self._bundle.manifest.version} offered={new_bundle.manifest.version}" ), ) diff --git a/tests/unit/test_policy_bundle.py b/tests/unit/test_policy_bundle.py index 019ca1c2..20e171d1 100644 --- a/tests/unit/test_policy_bundle.py +++ b/tests/unit/test_policy_bundle.py @@ -345,10 +345,9 @@ def test_load_bundle_warns_that_agent_os_version_is_legacy(bundle_dir, caplog): # spec computed different hashes for the same bundle. # --------------------------------------------------------------------------- + def _bundle_hash(manifest): - return bundle_module._canonical_bundle_hash( - manifest, {"policy.cedar": CEDAR_POLICY}, SCHEMA - ) + return bundle_module._canonical_bundle_hash(manifest, {"policy.cedar": CEDAR_POLICY}, SCHEMA) def test_bundle_hash_matches_rfc8785_for_non_ascii_author_identity(): @@ -366,9 +365,7 @@ def test_bundle_hash_matches_rfc8785_for_non_ascii_author_identity(): for k, v in manifest.items() if k not in bundle_module._UNHASHED_MANIFEST_KEYS }, - "policy_files": { - "policy.cedar": hashlib.sha256(CEDAR_POLICY.encode()).hexdigest() - }, + "policy_files": {"policy.cedar": hashlib.sha256(CEDAR_POLICY.encode()).hexdigest()}, "schema_hash": hashlib.sha256(SCHEMA.encode()).hexdigest(), } ) @@ -383,9 +380,7 @@ def test_bundle_hash_diverges_from_the_ascii_escaping_form(): legacy = json.dumps( { "manifest": { - k: v - for k, v in manifest.items() - if k not in bundle_module._UNHASHED_MANIFEST_KEYS + k: v for k, v in manifest.items() if k not in bundle_module._UNHASHED_MANIFEST_KEYS }, "policy_files": {"policy.cedar": bundle_module._sha256_hex(CEDAR_POLICY.encode())}, "schema_hash": bundle_module._sha256_hex(SCHEMA.encode()), @@ -403,9 +398,7 @@ def test_ascii_only_bundles_keep_the_hash_they_already_had(): legacy = json.dumps( { "manifest": { - k: v - for k, v in MANIFEST.items() - if k not in bundle_module._UNHASHED_MANIFEST_KEYS + k: v for k, v in MANIFEST.items() if k not in bundle_module._UNHASHED_MANIFEST_KEYS }, "policy_files": {"policy.cedar": bundle_module._sha256_hex(CEDAR_POLICY.encode())}, "schema_hash": bundle_module._sha256_hex(SCHEMA.encode()), @@ -431,7 +424,9 @@ def test_a_read_of_the_bundle_hashes_to_what_the_loader_computes(bundle_dir): ) (bundle_dir / "README.md").write_text("measured, but not part of the bundle hash") files = read_policy_files(str(bundle_dir)) - assert bundle_module.bundle_hash_of_files(files) == load_policy_bundle(str(bundle_dir)).bundle_hash + assert ( + bundle_module.bundle_hash_of_files(files) == load_policy_bundle(str(bundle_dir)).bundle_hash + ) def test_files_that_are_not_a_loadable_bundle_have_no_hash(bundle_dir): @@ -443,3 +438,153 @@ def test_files_that_are_not_a_loadable_bundle_have_no_hash(bundle_dir): assert bundle_module.bundle_hash_of_files({**files, "manifest.json": b"{not json"}) is None no_policy = {k: v for k, v in files.items() if not k.endswith(".cedar")} assert bundle_module.bundle_hash_of_files(no_policy) is None + + +@pytest.mark.parametrize("bad_ver", ["", "1.-2.0", "1.+2.0", "1.2_0.0", " 1.0.0 ", "1.\u0662.0"]) +def test_parse_bundle_version_rejects_non_decimal_segments(bad_ver: str): + from cmcp_runtime.errors import ConfigError + from cmcp_runtime.policy.bundle import parse_bundle_version + + with pytest.raises(ConfigError): + parse_bundle_version(bad_ver) + + +def test_bundle_and_approval_reject_non_canonical_base64url_and_duplicate_keys(bundle_dir): + from cmcp_runtime.catalog.approval import CatalogApprovalError, _decode + from cmcp_runtime.errors import ConfigError + from cmcp_runtime.policy.bundle import _b64url_decode + + # 64 zero bytes has canonical base64url 'A'*86; 'A'*85 + 'B' has non-zero unused low bits + with pytest.raises(ValueError): + _b64url_decode("A" * 85 + "B") + with pytest.raises(CatalogApprovalError): + _decode("A" * 85 + "B") + + # Duplicate keys or overflow floats in manifest.json + raw = json.dumps(MANIFEST) + (bundle_dir / "manifest.json").write_text(raw[:-1] + ', "version": "9.9.9"}', encoding="utf-8") + with pytest.raises(ConfigError, match="duplicate JSON object key"): + load_policy_bundle(str(bundle_dir)) + (bundle_dir / "manifest.json").write_text(raw[:-1] + ', "extra": 1e999}', encoding="utf-8") + with pytest.raises(ConfigError, match="non-finite JSON number"): + load_policy_bundle(str(bundle_dir)) + + +def test_approval_schema_rejects_exponent_overflow_and_accepts_finite_float(tmp_path, monkeypatch): + import cmcp_runtime.catalog.approval as approval_mod + from cmcp_runtime.catalog.approval import CatalogApprovalError, _approval_schema + + orig_schema_text = approval_mod.CATALOG_APPROVAL_SCHEMA_PATH.read_text(encoding="utf-8") + for bad_num in ("1e999", "-1e999", '{"nested": 1e999}'): + bad_path = tmp_path / "bad_schema.json" + bad_path.write_text( + orig_schema_text.rstrip()[:-1] + f', "x_ignored": {bad_num}}}', + encoding="utf-8", + ) + monkeypatch.setattr(approval_mod, "CATALOG_APPROVAL_SCHEMA_PATH", bad_path) + monkeypatch.setattr(approval_mod, "_schema_cache", None) + with pytest.raises(CatalogApprovalError, match="non-finite JSON number"): + _approval_schema() + + finite_path = tmp_path / "finite_schema.json" + finite_path.write_text( + orig_schema_text.rstrip()[:-1] + ', "x_ignored": 1.5}', + encoding="utf-8", + ) + monkeypatch.setattr(approval_mod, "CATALOG_APPROVAL_SCHEMA_PATH", finite_path) + monkeypatch.setattr(approval_mod, "_schema_cache", None) + assert _approval_schema()["x_ignored"] == 1.5 + monkeypatch.setattr(approval_mod, "_schema_cache", None) + + +@pytest.mark.parametrize( + "extra", + ( + '[[{"x":1,"x":2}]]', + "[[NaN]]", + "[[Infinity]]", + "[[[[-Infinity]]]]", + "1e999", + "-1e999", + "[[1e999]]", + '{"nested": -1e999}', + ), +) +def test_revocation_file_rejects_duplicate_key_statement_without_blocking_valid_revocation( + bundle_dir, + extra, +): + import base64 + + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + + from cmcp_runtime.errors import PolicySigningKeyRevoked + from cmcp_runtime.policy.bundle import ( + REVOCATION_FILE, + PolicySigningKeys, + policy_key_id, + revocation_pre_image, + ) + + cur_priv = Ed25519PrivateKey.generate() + cur_pub = cur_priv.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw + ) + suc_priv = Ed25519PrivateKey.generate() + suc_pub = suc_priv.public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw + ) + cur_id = policy_key_id(cur_pub) + valid_sig = ( + base64.urlsafe_b64encode(suc_priv.sign(revocation_pre_image(cur_id))) + .rstrip(b"=") + .decode("ascii") + ) + + # First statement is a duplicate-key forgery targeting successor; fails closed + dup_only = ( + f'[{{"revoked_key_id": "decoy", "revoked_key_id": "{cur_id}", "signature": "{valid_sig}"}}]' + ) + (bundle_dir / REVOCATION_FILE).write_text(dup_only, encoding="utf-8") + keys1 = PolicySigningKeys(cur_pub, suc_pub) + assert keys1.apply_file(str(bundle_dir)) == [] + assert keys1.current == cur_pub + + # Even ignored extension data must be strict JSON at every array depth. + nested = f'[{{"revoked_key_id":"{cur_id}","signature":"{valid_sig}","extra":{extra}}}]' + (bundle_dir / REVOCATION_FILE).write_text(nested, encoding="utf-8") + nested_keys = PolicySigningKeys(cur_pub, suc_pub) + assert nested_keys.apply_file(str(bundle_dir)) == [] + assert nested_keys.current == cur_pub + # An invalid statement must not suppress the later valid statement (including one with a finite float extension). + valid = json.dumps({"revoked_key_id": cur_id, "signature": valid_sig, "extra": 1.5}) + (bundle_dir / REVOCATION_FILE).write_text(nested[:-1] + "," + valid + "]", encoding="utf-8") + assert nested_keys.apply_file(str(bundle_dir)) == [cur_id] + assert nested_keys.current == suc_pub + + # A malformed duplicate-key or NaN statement alongside a valid signed revocation + # must NOT block the valid revocation from revoking cur_pub. + mixed = ( + f'[{{"revoked_key_id": "a", "revoked_key_id": "b", "signature": "{valid_sig}"}}, ' + f'{{"bad": NaN}}, ' + f'{{"revoked_key_id": "{cur_id}", "signature": "{valid_sig}"}}]' + ) + (bundle_dir / REVOCATION_FILE).write_text(mixed, encoding="utf-8") + keys2 = PolicySigningKeys(cur_pub, suc_pub) + assert keys2.apply_file(str(bundle_dir)) == [cur_id] + assert keys2.current == suc_pub + + # Now a bundle signed by the revoked cur_priv is refused as PolicySigningKeyRevoked + unsigned = load_policy_bundle(str(bundle_dir)) + manifest = dict(MANIFEST) + manifest["signature"] = ( + base64.urlsafe_b64encode( + cur_priv.sign(bundle_module.signing_pre_image(unsigned.bundle_hash)) + ) + .rstrip(b"=") + .decode("ascii") + ) + (bundle_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + with pytest.raises(PolicySigningKeyRevoked): + load_policy_bundle(str(bundle_dir), None, keys2.current, keys2.revoked)