diff --git a/CHANGELOG.md b/CHANGELOG.md index 2833e9c..c1c005c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,11 @@ based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). content-addressed record refuses construction whenever any gate is unmet and never counts unsigned, untrusted, expired, or duplicate reviewer material. Its schema stays internal to release governance rather than entering the - frozen public compatibility contract. + frozen public compatibility contract. A `tools/seal_release_quorum.py` driver + seals a certificate from a release-evidence manifest — running the + representative campaign live and loading the campaign, reviewer, and + trust-store evidence — or verifies a sealed certificate, without adding any + frozen `rigor` command. - A digest-bound proposed-1.0 compatibility contract classifies and ratchets the stable package-level Python imports, every installed CLI command with its option and positional spellings, and 61 explicitly inventoried schema diff --git a/tests/test_seal_release_quorum.py b/tests/test_seal_release_quorum.py new file mode 100644 index 0000000..5c225c2 --- /dev/null +++ b/tests/test_seal_release_quorum.py @@ -0,0 +1,243 @@ +# SPDX-License-Identifier: Apache-2.0 +# Apache License 2.0; see LICENSE. +# © Concepts 1996–2026 Miroslav Šotek. All rights reserved. +# © Code 2020–2026 Miroslav Šotek. All rights reserved. +# ORCID: 0009-0009-3560-0851 +# Contact: www.anulum.li | protoscience@anulum.li +# RigorFoundry — release-quorum seal driver tests +"""Prove the driver seals only fully evidenced candidates and verifies seals.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from repository_audit_git_repository import GitRepository +from signing_fixtures import sign_message, trust_store + +from rigor_foundry.campaign_evidence import ToolchainIdentity +from rigor_foundry.campaign_models import AuditCampaign +from rigor_foundry.models import canonical_digest +from rigor_foundry.review_attestation import ReviewerAttestation +from rigor_foundry.scanner import scan_repository +from rigor_foundry.trust import ED25519_ALGORITHM, REVIEW_ATTESTATION_SIGNATURE_DOMAIN +from tools import seal_release_quorum as driver + +POLICY_PATH = Path("rigor-foundry-policy.json") +FROZEN_AT = "2026-07-21T03:00:00Z" +INSTANT = "2026-07-25T00:00:00Z" +REVIEWED_AT = "2026-07-20T10:00:00Z" +EXPIRES_AT = "2026-08-20T10:00:00Z" +COMMIT = "f2038c686ed15e175ea9e86bc87e6b91182277f0" +TREE = "7817012b57c45103e6f8e0debdd2591fde28fef4" + + +def _prepare(path: Path, repository_id: str) -> Path: + """Create one real historical repository and return its root.""" + repository = GitRepository.create(path) + repository.write_text(f"src/{repository_id}.py", "VALUE = 1\n") + repository.write_policy() + repository.commit(f"test: create {repository_id}") + return repository.root + + +def _write_campaign(path: Path) -> Path: + """Write one real attestation campaign JSON and return its path.""" + root = _prepare(path / "campaign-source", "campaign") + campaign = AuditCampaign.build( + scan_repository(root, POLICY_PATH), + campaign_id="release-attestation", + project="rigor-foundry", + policy_path=POLICY_PATH.as_posix(), + toolchain=ToolchainIdentity.current(), + created_by="release-operator", + created_at=FROZEN_AT, + purpose="diagnostic", + expected_runs=2, + required_model_witnesses=2, + ) + target = path / "campaign.json" + target.write_text(json.dumps(campaign.to_dict()), encoding="utf-8") + return target + + +def _write_attestation(path: Path, key_id: str, assessment_digest: str) -> Path: + """Write one genuinely signed reviewer attestation JSON and return its path.""" + payload_digest = ReviewerAttestation.payload_digest( + reviewer_id=key_id, + algorithm=ED25519_ALGORITHM, + key_id=key_id, + assessment_body_digest=assessment_digest, + decision="pass", + reviewed_at=REVIEWED_AT, + expires_at=EXPIRES_AT, + ) + attestation = ReviewerAttestation.build( + reviewer_id=key_id, + key_id=key_id, + assessment_body_digest=assessment_digest, + decision="pass", + reviewed_at=REVIEWED_AT, + expires_at=EXPIRES_AT, + signature_hex=sign_message(key_id, REVIEW_ATTESTATION_SIGNATURE_DOMAIN, payload_digest), + ) + target = path / f"{key_id}.json" + target.write_text(json.dumps(attestation.to_dict()), encoding="utf-8") + return target + + +def _manifest(tmp_path: Path) -> Path: + """Assemble a complete, self-consistent release-evidence manifest.""" + app = _prepare(tmp_path / "sources" / "app", "app") + library = _prepare(tmp_path / "sources" / "library", "library") + assessment_digest = canonical_digest({"candidate": COMMIT, "tree": TREE}) + trust = trust_store("reviewer-a", "reviewer-b") + trust_path = tmp_path / "trust.json" + trust_path.write_text(json.dumps(trust.to_dict()), encoding="utf-8") + manifest = { + "candidate": {"commit": COMMIT, "tree": TREE, "version": "1.0.0"}, + "replay": { + "campaign_id": "release-replay", + "frozen_at": FROZEN_AT, + "repositories": [ + { + "repository_id": "app", + "repository_root": str(app), + "policy_path": "rigor-foundry-policy.json", + }, + { + "repository_id": "library", + "repository_root": str(library), + "policy_path": "rigor-foundry-policy.json", + }, + ], + "edges": [ + { + "from_repository": "app", + "to_repository": "library", + "relationship": "depends-on", + "rationale": "the app imports the library", + } + ], + }, + "attestation_campaign": str(_write_campaign(tmp_path)), + "observed_runs": 2, + "observed_model_witnesses": 2, + "reviewers": { + "attestations": [ + str(_write_attestation(tmp_path, "reviewer-a", assessment_digest)), + str(_write_attestation(tmp_path, "reviewer-b", assessment_digest)), + ], + "assessment_digest": assessment_digest, + "required": 2, + "trust_store": str(trust_path), + "instant": INSTANT, + }, + "artefacts": { + "wheel_sha256": canonical_digest({"a": "wheel"}), + "sdist_sha256": canonical_digest({"a": "sdist"}), + "sbom_sha256": canonical_digest({"a": "sbom"}), + }, + "surface": { + "signing_bundle_digests": [ + canonical_digest({"b": "wheel"}), + canonical_digest({"b": "sdist"}), + ], + "provenance_digest": canonical_digest({"p": "slsa"}), + "pages_digest": canonical_digest({"p": "docs"}), + }, + } + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + return manifest_path + + +def test_seal_produces_a_valid_certificate(tmp_path: Path) -> None: + """A complete manifest seals a certificate that round-trips through from_dict.""" + certificate = driver.seal(_manifest(tmp_path)) + certificate.validate() + assert certificate.candidate.commit == COMMIT + assert certificate.verified_reviewer_keys == ("reviewer-a", "reviewer-b") + assert certificate.execution_resolution == "succeeded" + + +def test_seal_command_writes_and_verify_command_reads(tmp_path: Path) -> None: + """The CLI seals to a file and verifies the same file end to end.""" + manifest_path = _manifest(tmp_path) + output = tmp_path / "certificate.json" + assert driver.main(["seal", "--manifest", str(manifest_path), "--output", str(output)]) == 0 + assert output.exists() + assert driver.main(["verify", "--certificate", str(output)]) == 0 + + +def test_verify_rejects_a_tampered_certificate(tmp_path: Path) -> None: + """A certificate whose sealed digest was altered fails CLI verification.""" + manifest_path = _manifest(tmp_path) + output = tmp_path / "certificate.json" + driver.main(["seal", "--manifest", str(manifest_path), "--output", str(output)]) + payload = json.loads(output.read_text(encoding="utf-8")) + payload["certificate_digest"] = canonical_digest({"swap": True}) + output.write_text(json.dumps(payload), encoding="utf-8") + assert driver.main(["verify", "--certificate", str(output)]) == 1 + + +def test_seal_command_reports_a_sub_quorum_manifest(tmp_path: Path) -> None: + """A manifest short of the reviewer quorum fails the seal command.""" + manifest_path = _manifest(tmp_path) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["reviewers"]["attestations"] = manifest["reviewers"]["attestations"][:1] + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + output = tmp_path / "certificate.json" + assert driver.main(["seal", "--manifest", str(manifest_path), "--output", str(output)]) == 1 + assert not output.exists() + + +def test_seal_rejects_a_missing_manifest(tmp_path: Path) -> None: + """A manifest path that does not exist is reported, not crashed.""" + output = tmp_path / "certificate.json" + assert ( + driver.main(["seal", "--manifest", str(tmp_path / "absent.json"), "--output", str(output)]) + == 1 + ) + + +def test_seal_rejects_invalid_manifest_json(tmp_path: Path) -> None: + """A manifest that is not valid JSON is reported cleanly.""" + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text("{not json", encoding="utf-8") + output = tmp_path / "certificate.json" + assert driver.main(["seal", "--manifest", str(manifest_path), "--output", str(output)]) == 1 + + +def test_seal_rejects_a_forged_candidate(tmp_path: Path) -> None: + """A candidate with a non-Git commit is rejected before any replay.""" + manifest_path = _manifest(tmp_path) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["candidate"]["commit"] = "not-a-commit" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with pytest.raises(ValueError, match=r"candidate\.commit"): + driver.seal(manifest_path) + + +def test_surface_rejects_non_list_signing_bundles(tmp_path: Path) -> None: + """A surface whose signing bundles are not a list is rejected.""" + with pytest.raises(ValueError, match="signing_bundle_digests must be a list"): + driver._surface( + { + "signing_bundle_digests": "one", + "provenance_digest": canonical_digest({"p": "slsa"}), + "pages_digest": canonical_digest({"p": "docs"}), + } + ) + + +def test_reviewer_attestations_must_be_a_list() -> None: + """A reviewers block whose attestations are not a list is rejected.""" + with pytest.raises(ValueError, match=r"reviewers\.attestations must be a list"): + driver._reviewer_attestations("not-a-list") + + +def test_require_list_accepts_a_list() -> None: + """The list guard returns the value unchanged for a real list.""" + assert driver._require_list([1, 2], "field") == [1, 2] diff --git a/tools/seal_release_quorum.py b/tools/seal_release_quorum.py new file mode 100644 index 0000000..020dfb4 --- /dev/null +++ b/tools/seal_release_quorum.py @@ -0,0 +1,301 @@ +# SPDX-License-Identifier: Apache-2.0 +# Apache License 2.0; see LICENSE. +# © Concepts 1996–2026 Miroslav Šotek. All rights reserved. +# © Code 2020–2026 Miroslav Šotek. All rights reserved. +# ORCID: 0009-0009-3560-0851 +# Contact: www.anulum.li | protoscience@anulum.li +# RigorFoundry — turnkey release-quorum certificate driver +"""Assemble or verify one sealed release-quorum certificate from evidence.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from rigor_foundry.campaign_evidence import ToolchainIdentity +from rigor_foundry.campaign_models import AuditCampaign +from rigor_foundry.cross_repository_campaign import InterRepositoryEdge +from rigor_foundry.cross_repository_capture import ( + RepositoryCaptureRequest, + capture_cross_repository_campaign, +) +from rigor_foundry.cross_repository_execution import ( + CrossRepositoryExecution, + CrossRepositoryExecutionPlan, + adapter_lock_digest, +) +from rigor_foundry.cross_repository_runtime import execute_cross_repository_campaign +from rigor_foundry.model_primitives import ( + parse_utc_timestamp, + require_utc_timestamp, +) +from rigor_foundry.models import require_integer, require_mapping, require_string +from rigor_foundry.release_quorum import ( + ReleaseArtefacts, + ReleaseCandidateAnchor, + ReleaseQuorumCertificate, + ReleaseSurfaceReferences, +) +from rigor_foundry.review_attestation import ReviewerAttestation +from rigor_foundry.scanner import scan_repository +from rigor_foundry.trust import VerificationTrustStore + + +def _require_list(value: object, field: str) -> list[object]: + """Return a list value or raise for one manifest field.""" + if not isinstance(value, list): + raise ValueError(f"{field} must be a list") + return value + + +def _load_json(path: Path, field: str) -> object: + """Read and parse one JSON evidence file.""" + try: + text = path.read_text(encoding="utf-8") + except OSError as error: + raise ValueError(f"{field} could not be read at {path}") from error + try: + return json.loads(text) + except json.JSONDecodeError as error: + raise ValueError(f"{field} is not valid JSON at {path}") from error + + +def _candidate(value: object) -> ReleaseCandidateAnchor: + """Build the release-candidate anchor from the manifest.""" + data = require_mapping(value, "candidate") + return ReleaseCandidateAnchor.build( + commit=require_string(data.get("commit"), "candidate.commit"), + tree=require_string(data.get("tree"), "candidate.tree"), + version=require_string(data.get("version"), "candidate.version"), + ) + + +def _artefacts(value: object) -> ReleaseArtefacts: + """Build the release-artefact identities from the manifest.""" + data = require_mapping(value, "artefacts") + return ReleaseArtefacts.build( + wheel_sha256=require_string(data.get("wheel_sha256"), "artefacts.wheel_sha256"), + sdist_sha256=require_string(data.get("sdist_sha256"), "artefacts.sdist_sha256"), + sbom_sha256=require_string(data.get("sbom_sha256"), "artefacts.sbom_sha256"), + ) + + +def _surface(value: object) -> ReleaseSurfaceReferences: + """Build the release-surface references from the manifest.""" + data = require_mapping(value, "surface") + bundles = _require_list(data.get("signing_bundle_digests"), "surface.signing_bundle_digests") + return ReleaseSurfaceReferences.build( + signing_bundle_digests=tuple( + require_string(item, f"surface.signing_bundle_digests[{index}]") + for index, item in enumerate(bundles) + ), + provenance_digest=require_string( + data.get("provenance_digest"), "surface.provenance_digest" + ), + pages_digest=require_string(data.get("pages_digest"), "surface.pages_digest"), + ) + + +def _capture_request(value: object, index: int) -> tuple[RepositoryCaptureRequest, str]: + """Scan one repository at its current head and freeze its capture request.""" + data = require_mapping(value, f"replay.repositories[{index}]") + repository_id = require_string(data.get("repository_id"), f"replay.repositories[{index}].id") + repository_root = Path( + require_string(data.get("repository_root"), f"replay.repositories[{index}].root") + ) + policy_path = require_string( + data.get("policy_path"), f"replay.repositories[{index}].policy_path" + ) + report = scan_repository(repository_root, Path(policy_path)) + request = RepositoryCaptureRequest.build( + repository_id=repository_id, + repository_root=repository_root, + requested_commit=report.head, + policy_digest=report.policy_digest, + rule_pack_version=report.rule_pack_version, + rule_pack_digest=report.rule_pack_digest, + adapter_lock_digest=adapter_lock_digest(report.policy), + toolchain_digest=ToolchainIdentity.current().identity_digest, + ) + return request, policy_path + + +def _edge(value: object, index: int) -> InterRepositoryEdge: + """Build one declared inter-repository dependency edge.""" + data = require_mapping(value, f"replay.edges[{index}]") + return InterRepositoryEdge.build( + from_repository=require_string(data.get("from_repository"), f"replay.edges[{index}].from"), + to_repository=require_string(data.get("to_repository"), f"replay.edges[{index}].to"), + relationship=require_string( + data.get("relationship"), f"replay.edges[{index}].relationship" + ), + rationale=require_string(data.get("rationale"), f"replay.edges[{index}].rationale"), + ) + + +def _run_replay(value: object) -> CrossRepositoryExecution: + """Capture and replay the declared representative campaign live.""" + data = require_mapping(value, "replay") + repositories = _require_list(data.get("repositories"), "replay.repositories") + prepared = tuple(_capture_request(item, index) for index, item in enumerate(repositories)) + requests = tuple(request for request, _policy in prepared) + policy_paths = tuple(policy for _request, policy in prepared) + edge_specs = _require_list(data.get("edges", []), "replay.edges") + edges = tuple(_edge(item, index) for index, item in enumerate(edge_specs)) + capture = capture_cross_repository_campaign( + campaign_id=require_string(data.get("campaign_id"), "replay.campaign_id"), + frozen_at=require_utc_timestamp(data.get("frozen_at"), "replay.frozen_at"), + requests=requests, + edges=edges, + ) + plan = CrossRepositoryExecutionPlan.build( + capture=capture, + requests=requests, + policy_paths=policy_paths, + ) + return execute_cross_repository_campaign(plan=plan, capture=capture, requests=requests) + + +def _reviewer_attestations(value: object) -> tuple[ReviewerAttestation, ...]: + """Load every declared reviewer attestation from its JSON file.""" + paths = _require_list(value, "reviewers.attestations") + return tuple( + ReviewerAttestation.from_dict( + _load_json( + Path(require_string(item, f"reviewers.attestations[{index}]")), + f"reviewers.attestations[{index}]", + ) + ) + for index, item in enumerate(paths) + ) + + +def seal(manifest_path: Path) -> ReleaseQuorumCertificate: + """Build one sealed certificate from a release-evidence manifest. + + Parameters + ---------- + manifest_path: + Path to the JSON manifest declaring the candidate, the replay + repositories and edges, and the campaign, reviewer, trust-store, + artefact, and surface evidence. + + Returns + ------- + ReleaseQuorumCertificate + The sealed, validated certificate. + + Raises + ------ + ValueError + If the manifest is malformed or any release gate is unmet. + """ + manifest = require_mapping(_load_json(manifest_path, "manifest"), "manifest") + candidate = _candidate(manifest.get("candidate")) + artefacts = _artefacts(manifest.get("artefacts")) + surface = _surface(manifest.get("surface")) + campaign = AuditCampaign.from_dict( + _load_json( + Path(require_string(manifest.get("attestation_campaign"), "attestation_campaign")), + "attestation_campaign", + ) + ) + reviewers = require_mapping(manifest.get("reviewers"), "reviewers") + trust_store = VerificationTrustStore.from_dict( + _load_json( + Path(require_string(reviewers.get("trust_store"), "reviewers.trust_store")), + "reviewers.trust_store", + ) + ) + attestations = _reviewer_attestations(reviewers.get("attestations")) + instant = parse_utc_timestamp( + require_utc_timestamp(reviewers.get("instant"), "reviewers.instant"), + "reviewers.instant", + ) + execution = _run_replay(manifest.get("replay")) + return ReleaseQuorumCertificate.build( + candidate=candidate, + execution=execution, + campaign=campaign, + observed_runs=require_integer(manifest.get("observed_runs"), "observed_runs", minimum=1), + observed_model_witnesses=require_integer( + manifest.get("observed_model_witnesses"), + "observed_model_witnesses", + minimum=1, + ), + reviewer_attestations=attestations, + reviewer_assessment_digest=require_string( + reviewers.get("assessment_digest"), + "reviewers.assessment_digest", + ), + required_reviewers=require_integer( + reviewers.get("required"), + "reviewers.required", + minimum=1, + ), + trust_store=trust_store, + instant=instant, + artefacts=artefacts, + surface=surface, + ) + + +def verify(certificate_path: Path) -> ReleaseQuorumCertificate: + """Reload and integrity-check one sealed certificate JSON file.""" + return ReleaseQuorumCertificate.from_dict(_load_json(certificate_path, "certificate")) + + +def _seal_command(args: argparse.Namespace) -> int: + """Seal a certificate from a manifest and write it to the output path.""" + certificate = seal(args.manifest) + args.output.write_text( + json.dumps(certificate.to_dict(), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + print(f"sealed release-quorum certificate {certificate.certificate_digest} at {args.output}") + return 0 + + +def _verify_command(args: argparse.Namespace) -> int: + """Verify a sealed certificate file and report its bound identity.""" + certificate = verify(args.certificate) + print( + "verified release-quorum certificate " + f"{certificate.certificate_digest} for candidate {certificate.candidate.commit}" + ) + return 0 + + +def _parser() -> argparse.ArgumentParser: + """Build the release-quorum driver argument parser.""" + parser = argparse.ArgumentParser(description="Seal or verify a release-quorum certificate.") + subparsers = parser.add_subparsers(dest="command", required=True) + seal_parser = subparsers.add_parser( + "seal", help="Seal a certificate from an evidence manifest." + ) + seal_parser.add_argument("--manifest", type=Path, required=True) + seal_parser.add_argument("--output", type=Path, required=True) + seal_parser.set_defaults(handler=_seal_command) + verify_parser = subparsers.add_parser("verify", help="Verify one sealed certificate file.") + verify_parser.add_argument("--certificate", type=Path, required=True) + verify_parser.set_defaults(handler=_verify_command) + return parser + + +def main(argv: list[str] | None = None) -> int: + """Run the release-quorum driver over one command.""" + args = _parser().parse_args(argv) + handler: object = args.handler + if not callable(handler): # pragma: no cover - argparse always binds a handler + raise SystemExit(2) + try: + return int(handler(args)) + except ValueError as error: + print(f"release-quorum error: {error}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:]))