#!/usr/bin/env python3 """Validate an exact-commit dev release receipt against the signing policy.""" from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path SCRIPT_DIRECTORY = Path(__file__).resolve().parent if str(SCRIPT_DIRECTORY) not in sys.path: sys.path.insert(0, str(SCRIPT_DIRECTORY)) from validate_serviceification_bom import ( canonical_json_sha256, load_bom, publication_receipt_roles, validate_bom, ) SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$") COMMIT = re.compile(r"^[0-9a-f]{40}$") BRANCH_REF = re.compile(r"^refs/heads/[^\s]+$") CALLER_WORKFLOW = ( "pingcap-inc/nexuscrm-pingcap-pack/.github/workflows/" "run-serviceification-test-stack-candidate.yml" ) PRODUCER_WORKFLOW = ( "pingcap-inc/nexuscrm-pingcap-pack/.github/workflows/" "produce-serviceification-test-stack-candidate.yml" ) DESIRED_BOM_PATHS = ("release/serviceification-d3-bom.json",) class ValidationFailure(RuntimeError): pass def reject_duplicates(pairs: list[tuple[str, object]]) -> dict[str, object]: value: dict[str, object] = {} for key, item in pairs: if key in value: raise ValidationFailure(f"duplicate candidate-receipt field: {key}") value[key] = item return value def canonical_json_bytes(value: object) -> bytes: return (json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n").encode() def digest_bytes(value: bytes) -> str: return "sha256:" + hashlib.sha256(value).hexdigest() def _require_digest(value: object, label: str) -> str: if not isinstance(value, str) or not SHA256.fullmatch(value): raise ValidationFailure(f"{label} is not canonical") return value def expected_evidence_owners(producer_root: Path, desired_bom_sha256: str) -> frozenset[str]: """Derive signable owner evidence from the exact desired BOM being signed.""" matches: list[dict[str, object]] = [] for relative in DESIRED_BOM_PATHS: bom_path = producer_root / relative if not bom_path.is_file(): continue desired = load_bom(bom_path) errors = validate_bom(desired, require_desired=True) if errors: raise ValidationFailure(f"producer desired BOM is invalid ({relative}): " + "; ".join(errors)) if canonical_json_sha256(desired) == desired_bom_sha256: matches.append(desired) if len(matches) != 1: raise ValidationFailure( "candidate receipt does not bind exactly one authorized producer desired BOM" ) desired = matches[0] return publication_receipt_roles(desired) def verify( receipt_path: Path, producer_root: Path, role: str, producer_sha: str, source_ref: str, run_id: str, run_attempt: str, approved_closures: list[str], ) -> dict[str, str]: if role not in {"frontend", "resolved_bom"}: raise ValidationFailure("subject role is not signable") if not COMMIT.fullmatch(producer_sha): raise ValidationFailure("producer commit is not exact") if not isinstance(approved_closures, list) or not approved_closures: raise ValidationFailure("signing policy has no approved producer closures") for closure_digest in approved_closures: _require_digest(closure_digest, "approved producer closure") raw = receipt_path.read_bytes() try: receipt = json.loads(raw, object_pairs_hook=reject_duplicates) except (json.JSONDecodeError, UnicodeDecodeError) as exc: raise ValidationFailure(f"candidate receipt is not valid JSON: {exc}") from exc if not isinstance(receipt, dict) or raw != canonical_json_bytes(receipt): raise ValidationFailure("candidate receipt is not canonical JSON") common = { "schema_version", "repository", "publisher_commit", "caller_workflow_ref", "caller_workflow_sha", "producer_workflow_ref", "producer_workflow_sha", "run_id", "run_attempt", "desired_bom_sha256", "producer_closure_sha256", "producer_closure_files", "subject_role", "oci_repository", "digest", "binding_sha256", } role_fields = { "frontend": {"source_tuple"}, "resolved_bom": { "resolved_bom_sha256", "frontend_source_commit", "frontend_digest", "frontend_binding_sha256", "verification_receipts", "verification_evidence", }, } if set(receipt) != common | role_fields[role]: raise ValidationFailure("candidate receipt has the wrong field set") if not source_ref: caller_ref = receipt.get("caller_workflow_ref") prefix = f"{CALLER_WORKFLOW}@" if isinstance(caller_ref, str) and caller_ref.startswith(prefix): source_ref = caller_ref.removeprefix(prefix) if not BRANCH_REF.fullmatch(source_ref): raise ValidationFailure("source ref is not an exact branch ref") expected = { "schema_version": "nexuscrm.serviceification-candidate-receipt/v1", "repository": "pingcap-inc/nexuscrm-pingcap-pack", "publisher_commit": producer_sha, "caller_workflow_ref": f"{CALLER_WORKFLOW}@{source_ref}", "caller_workflow_sha": producer_sha, "producer_workflow_ref": f"{PRODUCER_WORKFLOW}@{source_ref}", "producer_workflow_sha": producer_sha, "run_id": run_id, "run_attempt": run_attempt, "subject_role": role, "oci_repository": ( "ghcr.io/pingcap-inc/nexuscrm-frontend-assets" if role == "frontend" else "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom" ), } for field, wanted in expected.items(): if receipt.get(field) != wanted: raise ValidationFailure(f"candidate receipt {field} disagrees with the authorized run") for field in ("desired_bom_sha256", "digest", "binding_sha256"): _require_digest(receipt.get(field), f"candidate receipt {field}") expected_closure = receipt.get("producer_closure_sha256") if expected_closure not in approved_closures: raise ValidationFailure("candidate receipt does not use the reviewed B1 closure") closure = receipt.get("producer_closure_files") if not isinstance(closure, dict) or not closure: raise ValidationFailure("candidate receipt has no producer closure") root = producer_root.resolve() actual: dict[str, str] = {} for relative, claimed in closure.items(): if not isinstance(relative, str): raise ValidationFailure("candidate producer closure has a malformed path") _require_digest(claimed, "candidate producer closure digest") path = (root / relative).resolve() if root not in path.parents or not path.is_file(): raise ValidationFailure(f"candidate producer closure path is invalid: {relative}") actual[relative] = digest_bytes(path.read_bytes()) if actual != closure or digest_bytes(canonical_json_bytes(actual)) != expected_closure: raise ValidationFailure("candidate producer closure bytes disagree with the reviewed closure") if role == "frontend": source_tuple = receipt.get("source_tuple") expected_fields = { "nexuscrm_repository", "nexuscrm_commit", "app_packs_repository", "app_packs_commit", "l2_repository", "l2_commit", } if not isinstance(source_tuple, dict) or set(source_tuple) != expected_fields: raise ValidationFailure("frontend source tuple has the wrong shape") if digest_bytes(canonical_json_bytes(source_tuple)) != receipt["binding_sha256"]: raise ValidationFailure("frontend source tuple disagrees with its binding") else: if receipt["binding_sha256"] != receipt["desired_bom_sha256"]: raise ValidationFailure("resolved-BOM binding disagrees with desired BOM") if not isinstance(receipt.get("frontend_source_commit"), str) or not COMMIT.fullmatch( receipt["frontend_source_commit"] ): raise ValidationFailure("resolved-BOM frontend source commit is not exact") for field in ("resolved_bom_sha256", "frontend_digest", "frontend_binding_sha256"): _require_digest(receipt.get(field), f"resolved-BOM receipt {field}") expected_owners = expected_evidence_owners(root, receipt["desired_bom_sha256"]) for field in ("verification_receipts", "verification_evidence"): value = receipt.get(field) if not isinstance(value, dict) or set(value) != expected_owners: raise ValidationFailure(f"resolved-BOM receipt {field} has the wrong owner set") for item in value.values(): _require_digest(item, f"resolved-BOM receipt {field}") return { "subject_digest": receipt["digest"], "binding_sha256": receipt["binding_sha256"], "candidate_receipt_sha256": digest_bytes(raw), "producer_closure_sha256": expected_closure, "oci_repository": receipt["oci_repository"], } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--receipt", required=True) parser.add_argument("--producer-root", required=True) parser.add_argument("--subject-role", required=True, choices=("frontend", "resolved_bom")) parser.add_argument("--producer-sha", required=True) parser.add_argument("--source-ref", default="") parser.add_argument("--run-id", required=True) parser.add_argument("--run-attempt", required=True) parser.add_argument("--policy", required=True) parser.add_argument("--output", required=True) args = parser.parse_args() try: policy = json.loads(Path(args.policy).read_text(), object_pairs_hook=reject_duplicates) if (policy.get("schema_version") != "nexuscrm.l2-oci-owner-signing-policy/v3" or policy.get("publisher_state") != "active" or policy.get("owner_repository") != "pingcap-inc/nexuscrm-pingcap-pack"): raise ValidationFailure("signing authority policy is not active L2 v3") outputs = verify( Path(args.receipt).resolve(), Path(args.producer_root).resolve(), args.subject_role, args.producer_sha, args.source_ref, args.run_id, args.run_attempt, policy.get("approved_producer_closures"), ) except (OSError, ValueError, ValidationFailure) as exc: print(f"serviceification candidate signing gate: {exc}", file=sys.stderr) return 1 with Path(args.output).open("a", encoding="utf-8") as output: for key, value in outputs.items(): output.write(f"{key}={value}\n") return 0 if __name__ == "__main__": raise SystemExit(main())