from __future__ import annotations import hashlib import json import re import subprocess import tempfile import unittest from pathlib import Path from scripts.create_serviceification_candidate_receipt import producer_closure from scripts.validate_serviceification_bom import ( canonical_json_bytes, canonical_json_sha256, frontend_source_tuple, load_bom, ) from scripts import verify_serviceification_candidate_for_signing_v2 as signing ROOT = Path(__file__).resolve().parents[1] SOURCE_SHA = "a" * 40 class DevReleaseSigningTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.receipt_path = Path(self.temp.name) / "receipt.json" desired = load_bom(ROOT / "release/serviceification-d3-bom.json") closure_files, self.closure_digest = producer_closure(ROOT) self.source_tuple = frontend_source_tuple(desired["frontend"]) self.receipt = { "schema_version": "nexuscrm.serviceification-candidate-receipt/v1", "repository": "pingcap-inc/nexuscrm-pingcap-pack", "publisher_commit": SOURCE_SHA, "caller_workflow_sha": SOURCE_SHA, "producer_workflow_sha": SOURCE_SHA, "run_id": "123", "run_attempt": "1", "desired_bom_sha256": canonical_json_sha256(desired), "producer_closure_sha256": self.closure_digest, "producer_closure_files": closure_files, "subject_role": "frontend", "oci_repository": "ghcr.io/pingcap-inc/nexuscrm-frontend-assets", "digest": "sha256:" + "b" * 64, "binding_sha256": canonical_json_sha256(self.source_tuple), "source_tuple": self.source_tuple, } def tearDown(self) -> None: self.temp.cleanup() def write_receipt(self, source_ref: str) -> None: self.receipt["caller_workflow_ref"] = f"{signing.CALLER_WORKFLOW}@{source_ref}" self.receipt["producer_workflow_ref"] = f"{signing.PRODUCER_WORKFLOW}@{source_ref}" self.receipt_path.write_bytes(canonical_json_bytes(self.receipt)) def verify(self, source_ref: str) -> dict[str, str]: return signing.verify( self.receipt_path, ROOT, "frontend", SOURCE_SHA, source_ref, "123", "1", [self.closure_digest], ) def test_checked_in_signing_policy_matches_the_current_producer(self) -> None: policy = json.loads((ROOT / "release/serviceification-image-signing-policy.json").read_text()) self.assertIn( self.closure_digest, policy["approved_producer_closures"], "Producer changes must update the reviewed signing policy in the same change", ) def test_any_branch_ref_can_publish_the_exact_commit(self) -> None: for source_ref in ( "refs/heads/feature/formula-fix", "refs/heads/release/2026.08", "refs/heads/hotfix/urgent", ): with self.subTest(source_ref=source_ref): self.write_receipt(source_ref) result = self.verify(source_ref) self.assertEqual(self.receipt["digest"], result["subject_digest"]) def test_workflow_ref_must_match_the_selected_branch(self) -> None: self.write_receipt("refs/heads/feature/formula-fix") with self.assertRaisesRegex(signing.ValidationFailure, "caller_workflow_ref"): self.verify("refs/heads/release/2026.08") def test_source_ref_can_be_derived_from_the_receipt_for_existing_callers(self) -> None: self.write_receipt("refs/heads/release/2026.08") result = self.verify("") self.assertEqual(self.receipt["digest"], result["subject_digest"]) def test_trusted_signer_separates_branch_source_from_reviewed_authority(self) -> None: workflow = ( ROOT / ".github/workflows/trusted-publish-serviceification-test-stack.yml" ).read_text(encoding="utf-8") self.assertNotIn('.head_branch == "main"', workflow) self.assertNotIn('[[ "${producer_sha}" == "${PUBLISHER_SHA}" ]]', workflow) self.assertIn('echo "authority_sha=${authority_sha}"', workflow) self.assertNotIn("candidate_policy", workflow) self.assertNotIn("candidate_public_key", workflow) self.assertIn( 'fetch release/serviceification-image-signing-policy.json ' '"${AUTHORITY_SHA}" "${policy}"', workflow, ) def test_active_dev_publication_uses_reviewed_main_signer(self) -> None: producer = ( ROOT / ".github/workflows/produce-serviceification-test-stack-candidate.yml" ).read_text(encoding="utf-8") publisher = ( ROOT / ".github/workflows/publish-serviceification-test-stack.yml" ).read_text(encoding="utf-8") policy = json.loads( (ROOT / "release/serviceification-image-signing-policy.json").read_text( encoding="utf-8" ) ) verifier = (ROOT / "scripts/verify_serviceification_oci_publication.sh").read_text( encoding="utf-8" ) for workflow in (producer, publisher): self.assertNotIn('[[ "${GITHUB_REF}" == refs/heads/main ]]', workflow) self.assertNotIn('.head_branch == "main"', workflow) self.assertNotIn("producer_ref", publisher) self.assertNotIn("producer_sha", publisher.split("jobs:")[0]) self.assertNotIn("stable_exists", publisher) self.assertNotIn("verify-existing:", publisher) self.assertNotIn("stable tag appeared before commit", publisher) self.assertIn("needs: [sign, verify-new]", publisher) self.assertNotIn(" route:", publisher) trusted_ref = policy["trusted_workflow_ref"] self.assertRegex( trusted_ref, r"^pingcap-inc/nexuscrm-pingcap-pack/\.github/workflows/" r"trusted-publish-serviceification-test-stack\.yml@refs/heads/main$", ) self.assertIn(f"uses: {trusted_ref.replace('@refs/heads/main', '@main')}", publisher) self.assertEqual("nexuscrm.l2-oci-owner-signing-policy/v3", policy["schema_version"]) self.assertNotIn("source_ref", policy) self.assertNotIn("workflow_ref", policy) self.assertEqual( "pingcap-inc/nexuscrm-pingcap-pack/.github/workflows/" "publish-serviceification-test-stack.yml", policy["workflow_path"], ) self.assertIn("nexuscrm.l2-oci-owner-signing-policy/v1)", verifier) self.assertIn("nexuscrm.l2-oci-owner-signing-policy/v2|nexuscrm.l2-oci-owner-signing-policy/v3)", verifier) self.assertIn("nexuscrm.l0-engine-image-signing-policy/v2)", verifier) self.assertIn( "L0 Engine signature does not bind exactly one branch ref", verifier ) self.assertIn("-a 'io.nexuscrm.source.ref=refs/heads/main'", verifier) self.assertIn( "sha256:348ad9206d3cc94e68dbd5bbfb422f5e57f75a90c550ffd7d7c4684942fbd8bf", verifier, ) self.assertIn("L2 owner public key is not the reviewed signing key", verifier) def test_test2_candidate_owns_l0_branch_release_admission(self) -> None: producer = ( ROOT / ".github/workflows/produce-serviceification-test-stack-candidate.yml" ).read_text(encoding="utf-8") gate = ( ROOT / ".github/workflows/serviceification-bom-gate.yml" ).read_text(encoding="utf-8") self.assertIn("Keep legacy v1 Engine releases main-only", producer) self.assertIn("nexuscrm.l0-engine-image-signing-policy/v1)", producer) self.assertIn("git -C exact/runtime merge-base --is-ancestor HEAD", producer) self.assertIn("nexuscrm.l0-engine-image-signing-policy/v2) ;;", producer) self.assertNotIn("verify_serviceification_oci_publication.sh", gate) def test_policy_update_reuses_the_original_candidate_receipt(self) -> None: self.write_receipt("refs/heads/release/original") raw = self.receipt_path.read_bytes() # A new publisher/authority revision approves the old build. Neither # the receipt nor the old producer policy must be rewritten. accepted = signing.verify(self.receipt_path, ROOT, "frontend", SOURCE_SHA, "refs/heads/release/original", "123", "1", ["sha256:" + "f" * 64, self.closure_digest]) self.assertEqual(self.receipt["digest"], accepted["subject_digest"]) self.assertEqual(raw, self.receipt_path.read_bytes()) with self.assertRaisesRegex(signing.ValidationFailure, "reviewed B1 closure"): signing.verify(self.receipt_path, ROOT, "frontend", SOURCE_SHA, "refs/heads/release/original", "123", "1", ["sha256:" + "f" * 64]) def test_receipt_identity_and_build_bytes_still_fail_closed(self) -> None: self.write_receipt("refs/heads/release/original") for field, value in (("publisher_commit", "b" * 40), ("run_id", "999"), ("digest", "not-a-digest")): with self.subTest(field=field): old = self.receipt[field] self.receipt[field] = value self.write_receipt("refs/heads/release/original") with self.assertRaises(signing.ValidationFailure): self.verify("refs/heads/release/original") self.receipt[field] = old file = next(iter(self.receipt["producer_closure_files"])) self.receipt["producer_closure_files"][file] = "sha256:" + "0" * 64 self.write_receipt("refs/heads/release/original") with self.assertRaisesRegex(signing.ValidationFailure, "producer closure bytes"): self.verify("refs/heads/release/original") if __name__ == "__main__": unittest.main()