"""Independent verifier for CodeMani detached receipt signatures.""" from __future__ import annotations import argparse import base64 import binascii import hashlib import json import math import sys from datetime import datetime, timezone from pathlib import Path from typing import Any from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey TRUST_STORE_SCHEMA = "codemani.receipt_trust_store.v1" SIGNATURE_SCHEMA = "codemani.detached_receipt_signature.v1" SIGNED_PAYLOAD_SCHEMA = "codemani.receipt_signature_payload.v1" DEFAULT_PURPOSE = "codemani.receipt.v1" MAX_RECEIPT_BYTES = 2_000_000 MAX_METADATA_BYTES = 128_000 def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: value: dict[str, Any] = {} for key, item in pairs: if key in value: raise ValueError(f"duplicate JSON key: {key}") value[key] = item return value def _reject_nonfinite(value: str) -> None: raise ValueError(f"non-finite JSON constant: {value}") def _finite_float(value: str) -> float: parsed = float(value) if not math.isfinite(parsed): raise ValueError("non-finite JSON number") return parsed def _load_json(path: Path, *, max_bytes: int, label: str) -> tuple[dict[str, Any], bytes]: if path.is_symlink() or not path.is_file(): raise ValueError(f"{label} must be a regular non-link file") data = path.read_bytes() if not data or len(data) > max_bytes: raise ValueError(f"{label} exceeds the size budget") value = json.loads( data, object_pairs_hook=_reject_duplicate_keys, parse_constant=_reject_nonfinite, parse_float=_finite_float, ) if not isinstance(value, dict): raise ValueError(f"{label} must be a JSON object") return value, data def _canonical_json(value: Any) -> bytes: return json.dumps( value, allow_nan=False, ensure_ascii=True, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def _decode_b64(value: object, *, expected_bytes: int, label: str) -> bytes: if not isinstance(value, str): raise ValueError(f"{label} must be base64 text") try: decoded = base64.b64decode(value, validate=True) except (binascii.Error, ValueError) as exc: raise ValueError(f"{label} is malformed") from exc if len(decoded) != expected_bytes: raise ValueError(f"{label} has the wrong length") return decoded def _parse_utc(value: object, label: str) -> datetime: if not isinstance(value, str) or not value.endswith("Z"): raise ValueError(f"{label} must be an ISO-8601 UTC timestamp") try: parsed = datetime.fromisoformat(value[:-1] + "+00:00") except ValueError as exc: raise ValueError(f"{label} must be an ISO-8601 UTC timestamp") from exc if parsed.utcoffset() != timezone.utc.utcoffset(parsed): raise ValueError(f"{label} must be UTC") return parsed def verify_receipt( receipt_path: Path, signature_path: Path, trust_store_path: Path, *, expected_purpose: str = DEFAULT_PURPOSE, ) -> dict[str, Any]: receipt, receipt_bytes = _load_json( receipt_path, max_bytes=MAX_RECEIPT_BYTES, label="receipt" ) signature, _ = _load_json( signature_path, max_bytes=MAX_METADATA_BYTES, label="receipt signature" ) trust_store, _ = _load_json( trust_store_path, max_bytes=MAX_METADATA_BYTES, label="receipt trust store" ) if not isinstance(receipt.get("schema"), str): raise ValueError("receipt must declare a schema") if signature.get("schema") != SIGNATURE_SCHEMA or signature.get("algorithm") != "ed25519": raise ValueError("unsupported receipt signature schema") if trust_store.get("schema") != TRUST_STORE_SCHEMA: raise ValueError("unsupported receipt trust-store schema") if trust_store.get("purpose") != DEFAULT_PURPOSE: raise ValueError("unsupported receipt trust-store purpose") payload = signature.get("signed_payload") if not isinstance(payload, dict) or payload.get("schema") != SIGNED_PAYLOAD_SCHEMA: raise ValueError("invalid signed receipt payload") key_id = signature.get("key_id") if not isinstance(key_id, str) or payload.get("key_id") != key_id: raise ValueError("receipt signature key binding mismatch") if payload.get("purpose") != expected_purpose: raise ValueError("receipt signature purpose mismatch") artifact_name = payload.get("artifact_name") if not isinstance(artifact_name, str) or not artifact_name: raise ValueError("receipt signature artifact name is missing") if artifact_name != receipt_path.name: # The signature binds the receipt BYTES (digest + byte count below); # the signed artifact name is informational, so a renamed local copy # (for example "llms.txt.receipt (1).json" from a second download) # must not fail verification. Surface the canonical name instead. print( f"note: signed artifact name is {artifact_name!r}; " f"local file is named {receipt_path.name!r}", file=sys.stderr, ) if payload.get("receipt_schema") != receipt.get("schema"): raise ValueError("receipt signature schema binding mismatch") if payload.get("receipt_bytes") != len(receipt_bytes): raise ValueError("receipt signature byte-count mismatch") digest = hashlib.sha256(receipt_bytes).hexdigest() if payload.get("receipt_sha256") != digest: raise ValueError("receipt signature digest mismatch") if payload.get("timestamp_semantics") != "signer_asserted_not_external_timestamp": raise ValueError("unsupported receipt timestamp semantics") rows = trust_store.get("trusted_keys") if not isinstance(rows, list) or not rows: raise ValueError("receipt trust store has no keys") indexed: dict[str, dict[str, Any]] = {} for row in rows: if not isinstance(row, dict) or not isinstance(row.get("key_id"), str): raise ValueError("invalid receipt trust-store key") row_id = row["key_id"] if row_id in indexed: raise ValueError("duplicate receipt trust-store key id") if row.get("algorithm") != "ed25519" or row.get("status") not in { "active", "retired", "revoked", }: raise ValueError("unsupported receipt trust-store key policy") public_bytes = _decode_b64(row.get("public_key_b64"), expected_bytes=32, label="public key") if row.get("public_key_sha256") != hashlib.sha256(public_bytes).hexdigest(): raise ValueError("receipt trust-store public-key fingerprint mismatch") _parse_utc(row.get("not_before_utc"), "not_before_utc") if row.get("not_after_utc") is not None: _parse_utc(row.get("not_after_utc"), "not_after_utc") if row.get("revoked_utc") is not None: _parse_utc(row.get("revoked_utc"), "revoked_utc") indexed[row_id] = row if key_id not in indexed: raise ValueError("unknown receipt signing key") key = indexed[key_id] issued = _parse_utc(payload.get("issued_utc"), "issued_utc") not_before = _parse_utc(key["not_before_utc"], "not_before_utc") not_after = _parse_utc(key["not_after_utc"], "not_after_utc") if key.get("not_after_utc") else None if key["status"] == "revoked" or key.get("revoked_utc") is not None: raise ValueError("receipt signing key is revoked") if key["status"] == "retired" and not_after is None: raise ValueError("retired receipt key must declare not_after_utc") if issued < not_before or (not_after is not None and issued > not_after): raise ValueError("receipt signature falls outside the key validity interval") public_bytes = _decode_b64(key["public_key_b64"], expected_bytes=32, label="public key") signature_bytes = _decode_b64(signature.get("signature_b64"), expected_bytes=64, label="signature") try: Ed25519PublicKey.from_public_bytes(public_bytes).verify( signature_bytes, _canonical_json(payload), ) except InvalidSignature as exc: raise ValueError("receipt signature verification failed") from exc return { "key_id": key_id, "public_key_sha256": key["public_key_sha256"], "purpose": expected_purpose, "receipt_bytes": len(receipt_bytes), "receipt_schema": receipt["schema"], "receipt_sha256": digest, "status": "PASS", "timestamp_semantics": payload["timestamp_semantics"], } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--receipt", required=True) parser.add_argument("--signature", required=True) parser.add_argument("--trusted-keys", required=True) parser.add_argument("--purpose", default=DEFAULT_PURPOSE) args = parser.parse_args() try: result = verify_receipt( Path(args.receipt), Path(args.signature), Path(args.trusted_keys), expected_purpose=args.purpose, ) except Exception as exc: print(json.dumps({"reason": str(exc), "status": "FAIL"}, sort_keys=True)) return 1 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())