#!/usr/bin/env python3 """Create canonical, run-owned receipts for unsigned L2 OCI candidates.""" from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path from typing import Any 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 ( ValidationFailure, canonical_json_bytes, canonical_json_sha256, frontend_source_tuple, 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" ) PRODUCER_CLOSURE_PATHS = ( ".github/workflows/run-serviceification-test-stack-candidate.yml", ".github/workflows/produce-serviceification-test-stack-candidate.yml", "infra/serviceification-frontend.Dockerfile", "infra/serviceification-resolved-bom.Dockerfile", "scripts/create_serviceification_candidate_receipt.py", "scripts/materialize_frontend_app_packs_source.py", "scripts/produce_serviceification_candidate.sh", "scripts/resolve_l0_engine_release_locator.sh", "scripts/serviceification_deployment_tools.py", "scripts/resolve_serviceification_bom.py", "scripts/lib/final_pack_contract.py", "scripts/validate_serviceification_bom.py", "scripts/verify_buildkit_sbom.py", "scripts/verify_serviceification_d1_metadata_authority.py", "scripts/verify_serviceification_l2_artifact.py", "scripts/verify_serviceification_oci_publication.sh", "scripts/verified_l2_signing_policy.py", "deployments/offline/runtime/common.py", ) def file_digest(path: Path) -> str: return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() def producer_closure(root: Path) -> tuple[dict[str, str], str]: files: dict[str, str] = {} for relative in PRODUCER_CLOSURE_PATHS: path = root / relative if not path.is_file(): raise ValidationFailure(f"producer closure file is missing: {relative}") files[relative] = file_digest(path) return files, canonical_json_sha256(files) def common(args: argparse.Namespace, desired: dict[str, Any]) -> dict[str, Any]: errors = validate_bom(desired, require_desired=True) if not COMMIT.fullmatch(args.publisher_commit): errors.append("publisher commit must be exact") caller_prefix = f"{CALLER_WORKFLOW}@" source_ref = ( args.caller_workflow_ref.removeprefix(caller_prefix) if args.caller_workflow_ref.startswith(caller_prefix) else "" ) if not BRANCH_REF.fullmatch(source_ref): errors.append("caller workflow ref must identify an exact repository branch") if args.producer_workflow_ref != f"{PRODUCER_WORKFLOW}@{source_ref}": errors.append("producer workflow ref must identify the same repository branch") if args.caller_workflow_sha != args.publisher_commit: errors.append("caller workflow SHA must equal the publisher commit") if args.producer_workflow_sha != args.publisher_commit: errors.append("producer workflow SHA must equal the publisher commit") if not re.fullmatch(r"[1-9][0-9]*", args.run_id): errors.append("run ID must be positive") if not re.fullmatch(r"[1-9][0-9]*", args.run_attempt): errors.append("run attempt must be positive") if not SHA256.fullmatch(args.digest): errors.append("candidate digest must be canonical") if errors: raise ValidationFailure("; ".join(errors)) closure_files, closure_digest = producer_closure(Path(args.producer_root).resolve()) return { "repository": "pingcap-inc/nexuscrm-pingcap-pack", "publisher_commit": args.publisher_commit, "caller_workflow_ref": args.caller_workflow_ref, "caller_workflow_sha": args.caller_workflow_sha, "producer_workflow_ref": args.producer_workflow_ref, "producer_workflow_sha": args.producer_workflow_sha, "run_id": args.run_id, "run_attempt": args.run_attempt, "desired_bom_sha256": canonical_json_sha256(desired), "producer_closure_sha256": closure_digest, "producer_closure_files": closure_files, } def expected_receipts(desired: dict[str, Any]) -> set[str]: return {f"{role}.json" for role in publication_receipt_roles(desired)} def frontend_receipt(args: argparse.Namespace, desired: dict[str, Any]) -> dict[str, Any]: value = common(args, desired) source_tuple = frontend_source_tuple(desired["frontend"]) binding = canonical_json_sha256(source_tuple) value.update({ "schema_version": "nexuscrm.serviceification-candidate-receipt/v1", "subject_role": "frontend", "oci_repository": desired["frontend"]["oci_repository"], "digest": args.digest, "binding_sha256": binding, "source_tuple": source_tuple, }) return value def resolved_bom_receipt(args: argparse.Namespace, desired: dict[str, Any]) -> dict[str, Any]: value = common(args, desired) resolved_path = Path(args.resolved_bom).resolve() resolved = load_bom(resolved_path) errors = validate_bom(resolved, require_deployable=True) if resolved.get("desired_bom_sha256") != value["desired_bom_sha256"]: errors.append("resolved BOM disagrees with desired BOM") frontend = resolved.get("frontend", {}).get("publication", {}) if not COMMIT.fullmatch(args.frontend_source_commit or ""): errors.append("frontend source commit must be exact") if frontend.get("source_commit") != args.frontend_source_commit: errors.append("resolved frontend disagrees with signed frontend source commit") if frontend.get("digest") != args.frontend_digest: errors.append("resolved frontend disagrees with signed frontend digest") if frontend.get("source_tuple_sha256") != args.frontend_binding: errors.append("resolved frontend disagrees with frontend binding") if not SHA256.fullmatch(args.frontend_digest or "") or not SHA256.fullmatch(args.frontend_binding or ""): errors.append("frontend digest and binding must be canonical") receipts_dir = Path(args.receipts_dir).resolve() expected = expected_receipts(desired) actual = {path.name for path in receipts_dir.glob("*.json")} if actual != expected: errors.append( f"verification receipt set mismatch: got {sorted(actual)}, want {sorted(expected)}" ) evidence_dir = Path(args.evidence_dir).resolve() actual_evidence = {path.name for path in evidence_dir.glob("*.json")} expected_evidence = {name.removesuffix(".json") + ".evidence.json" for name in expected} if actual_evidence != expected_evidence: errors.append( f"verification evidence set mismatch: got {sorted(actual_evidence)}, want {sorted(expected_evidence)}" ) if errors: raise ValidationFailure("; ".join(errors)) value.update({ "schema_version": "nexuscrm.serviceification-candidate-receipt/v1", "subject_role": "resolved_bom", "oci_repository": "ghcr.io/pingcap-inc/nexuscrm-serviceification-bom", "digest": args.digest, "binding_sha256": value["desired_bom_sha256"], "resolved_bom_sha256": file_digest(resolved_path), "frontend_source_commit": args.frontend_source_commit, "frontend_digest": args.frontend_digest, "frontend_binding_sha256": args.frontend_binding, "verification_receipts": { path.name.removesuffix(".json"): file_digest(path) for path in sorted(receipts_dir.glob("*.json")) }, "verification_evidence": { path.name.removesuffix(".evidence.json"): file_digest(path) for path in sorted(evidence_dir.glob("*.json")) }, }) return value def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("phase", choices=("frontend", "resolved-bom")) parser.add_argument("--desired-bom", default="release/serviceification-d3-bom.json") parser.add_argument("--publisher-commit", required=True) parser.add_argument("--producer-root", required=True) parser.add_argument("--caller-workflow-ref", required=True) parser.add_argument("--caller-workflow-sha", required=True) parser.add_argument("--producer-workflow-ref", required=True) parser.add_argument("--producer-workflow-sha", required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--run-attempt", required=True) parser.add_argument("--digest", required=True) parser.add_argument("--output", required=True) parser.add_argument("--resolved-bom", default="") parser.add_argument("--receipts-dir", default="") parser.add_argument("--evidence-dir", default="") parser.add_argument("--frontend-source-commit", default="") parser.add_argument("--frontend-digest", default="") parser.add_argument("--frontend-binding", default="") args = parser.parse_args() try: desired = load_bom(Path(args.desired_bom).resolve()) receipt = frontend_receipt(args, desired) if args.phase == "frontend" else resolved_bom_receipt(args, desired) except (OSError, ValidationFailure) as exc: print(f"cannot create serviceification candidate receipt: {exc}") return 1 output = Path(args.output).resolve() output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(canonical_json_bytes(receipt)) print(f"serviceification {args.phase} candidate receipt: {output}") return 0 if __name__ == "__main__": raise SystemExit(main())