#!/usr/bin/env python3 """Governed release evidence producer (V2R2C2 P0, operation governed.release.evidence.produce). This module is the *producer* half of the governed evidence pipeline. It never invents candidate/release evidence: every digest and pass/fail flag in the closure is derived from real build artifacts and real verification executions supplied as input. It assembles a self-consistent, root-owned evidence closure below ``source_root/`` that ``TrustedBuilderEvidenceMaterializer`` (the projection half) can re-validate and project byte-for-byte. Contract (see docs/V2R2C2-P0-convergence-20260909.md section 1): - operation_id : governed.release.evidence.produce - input : governed build artifacts + deployment conventions (no hand-write) - output : source_root// full evidence_layout closure + receipt.json - fail-closed : mismatched secret-scan digest, tampered artifact, non-passing tests/canary/rollback/secret-scan, or inconsistent gitea all abort without writing a closure. - idempotent : an existing identical closure is accepted; a conflicting one fails. """ from __future__ import annotations import argparse import hashlib import json import os import secrets from dataclasses import dataclass, field from pathlib import Path import shutil import stat from typing import Any, Mapping from aios_baseline.operations import OperationDefinition, OperationRegistry from aios_trusted_candidate_builder import ( load_builder_policy, TrustedCandidateBuilderConfig, ) from release_evidence_materializer import EvidenceMaterializationError from release_version import validate_release_version PRODUCE_SCHEMA = "aios.governed-release-evidence-producer.v1" EVIDENCE_RECEIPT_SCHEMA = "aios.trusted-candidate-builder.evidence.v1" # Each governed evidence type (tests/canary/gitea/rollback/secret_scan) must be # bound to a trusted-provider receipt. A plain caller-supplied {"passed": true} # is never accepted on its own: the producer requires the receipt that proves a # trusted provider actually ran the check and produced the referenced result. TRUSTED_PROVIDER_RECEIPT_SCHEMA = "aios.trusted-provider.evidence.v1" TRUSTED_PROVIDER_EVIDENCE_TYPES = ("tests", "canary", "gitea", "rollback", "secret_scan") class EvidenceProduceError(RuntimeError): """Stable, non-sensitive producer failure.""" def _canonical_json(value: Mapping[str, Any]) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=True, ).encode("utf-8") def _file_sha256(path: Path) -> str: digest = hashlib.sha256() with Path(path).open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _validate_oid(value: object, name: str) -> str: normalized = str(value or "").strip().lower() if ( len(normalized) != 40 or any(char not in "0123456789abcdef" for char in normalized) ): raise EvidenceProduceError(f"{name.upper()}_INVALID") return normalized def _validate_identifier(value: object, name: str) -> str: normalized = str(value or "").strip() if ( not normalized or len(normalized) > 255 or normalized[0] in ".-" or any( char not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._:-" for char in normalized ) ): raise EvidenceProduceError(f"{name.upper()}_INVALID") return normalized def _validate_sha256(value: object, name: str, *, allow_empty: bool = False) -> str: normalized = str(value or "").strip().lower() if allow_empty and not normalized: return "" if ( len(normalized) != 64 or any(char not in "0123456789abcdef" for char in normalized) ): raise EvidenceProduceError(f"{name.upper()}_INVALID") return normalized @dataclass(frozen=True) class GovernedBuildArtifacts: """Real build artifacts + real verification results. Producer never invents these.""" version: str candidate_id: str source_commit: str tree_oid: str artifact: bytes manifest: Mapping[str, Any] package_closure: bytes policy: Mapping[str, Any] tests: Mapping[str, Any] canary: Mapping[str, Any] gitea: Mapping[str, Any] rollback: Mapping[str, Any] secret_scan: Mapping[str, Any] # Deployment conventions (supplied by release planning, not evidence invention). deployment: Mapping[str, Any] tag: Mapping[str, Any] finalization: Mapping[str, Any] rollback_sha256: str = "" issued_at: str = "" @dataclass(frozen=True) class ProduceResult: status: str idempotent: bool receipt_sha256: str closure_dir: str evidence_digests: Mapping[str, str] = field(default_factory=dict) def _require_trusted_provider_receipt( name: str, evidence: Mapping[str, Any], source_commit: str, tree_oid: str, ) -> None: """Require a trusted-provider binding for `evidence` (second-layer check). A caller may never pass a bare ``{"passed": true}``: each governed evidence type must be bound to a trusted provider. Two accepted forms, both fail-closed on any missing/structurally invalid/digest-mismatched binding: 1. Round-2 native ``receipt`` (schema ``aios.trusted-provider.evidence.v1``) — a caller-carried receipt. 2. Round-3 resolver-authenticated ``provider_receipt`` provenance (schema ``aios.trusted-evidence-provider-receipt.v2``) — produced by ``TrustedProviderReceiptResolver`` from an authenticated store. Its input is the resolver-authenticated provenance, never a caller-supplied body. This helper is a *second-layer* binding validator: for the Round-3 path the authoritative proof is the resolver's authenticated provenance embedded as ``provider_receipt``, and this check only re-asserts content/OID binding on top of it. """ receipt = evidence.get("receipt") if isinstance(receipt, Mapping) and receipt: body = {k: v for k, v in dict(evidence).items() if k != "receipt"} expected = _sha256_bytes(_canonical_json(body)) if receipt.get("schema") != TRUSTED_PROVIDER_RECEIPT_SCHEMA: raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID") if receipt.get("content_sha256") != expected: raise EvidenceProduceError("EVIDENCE_RECEIPT_CONTENT_MISMATCH") for f in ("provider_identity", "operation_run_id", "result_reference", "environment", "issued_at"): if not str(receipt.get(f, "") or "").strip(): raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID") if receipt.get("source_oid") not in (None, "", source_commit): raise EvidenceProduceError("EVIDENCE_RECEIPT_SOURCE_OID_MISMATCH") if receipt.get("tree_oid") not in (None, "", tree_oid): raise EvidenceProduceError("EVIDENCE_RECEIPT_TREE_OID_MISMATCH") return provider_receipt = evidence.get("provider_receipt") if isinstance(provider_receipt, Mapping) and provider_receipt: if ( provider_receipt.get("schema") != "aios.trusted-evidence-provider-receipt.v2" ): raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID") for f in ("provider_identity", "provider_operation", "operation_run_id", "result_reference", "content_sha256", "source_oid", "tree_oid", "issued_at"): if not str(provider_receipt.get(f, "") or "").strip(): raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID") if provider_receipt.get("source_oid") != source_commit: raise EvidenceProduceError("EVIDENCE_RECEIPT_SOURCE_OID_MISMATCH") if provider_receipt.get("tree_oid") != tree_oid: raise EvidenceProduceError("EVIDENCE_RECEIPT_TREE_OID_MISMATCH") return raise EvidenceProduceError("EVIDENCE_RECEIPT_MISSING") def _verify_artifacts( config: TrustedCandidateBuilderConfig, request: GovernedBuildArtifacts, ) -> dict[str, str]: """Validate input consistency (fail-closed) and return the evidence digests.""" try: version = validate_release_version(request.version) except ValueError as exc: raise EvidenceProduceError("VERSION_INVALID") from exc if version != request.version: raise EvidenceProduceError("VERSION_INVALID") candidate_id = _validate_identifier(request.candidate_id, "candidate_id") source_commit = _validate_oid(request.source_commit, "source_commit") tree_oid = _validate_oid(request.tree_oid, "tree_oid") if ( set(request.manifest) != { "artifact_sha256", "package_closure_sha256", } ): raise EvidenceProduceError("MANIFEST_SCHEMA_INVALID") closure_sha = _validate_sha256( request.manifest.get("package_closure_sha256"), "package_closure_sha256" ) artifact_sha = _sha256_bytes(request.artifact) if request.manifest["artifact_sha256"] != artifact_sha: raise EvidenceProduceError("MANIFEST_ARTIFACT_DIGEST_MISMATCH") # manifest_sha256 is the digest of the manifest evidence file itself. It is # derived (never invented) from the manifest content; the manifest does not # embed its own digest to avoid a self-referential fixed point. manifest_sha = _sha256_bytes(_canonical_json(dict(request.manifest))) if _sha256_bytes(request.package_closure) != closure_sha: raise EvidenceProduceError("PACKAGE_CLOSURE_DIGEST_MISMATCH") # policy_sha256 is the digest of the policy evidence file content; derived # here (never invented, never self-referential). policy_sha = _sha256_bytes(_canonical_json(dict(request.policy))) # Secret scan must cover the exact artifact bytes. secret_scan_artifact = _validate_sha256( request.secret_scan.get("artifact_sha256"), "secret_scan.artifact_sha256" ) if secret_scan_artifact != artifact_sha: raise EvidenceProduceError("SECRET_SCAN_ARTIFACT_DIGEST_MISMATCH") # Every governed evidence type must be bound to a trusted-provider receipt. # A bare caller-supplied {"passed": true} (no receipt) fails closed here; # the producer never accepts an invented pass flag. for etype in TRUSTED_PROVIDER_EVIDENCE_TYPES: _require_trusted_provider_receipt( etype, getattr(request, etype), source_commit, tree_oid ) if request.secret_scan.get("passed") is not True: raise EvidenceProduceError("SECRET_SCAN_NOT_PASSED") if request.tests.get("passed") is not True: raise EvidenceProduceError("TESTS_NOT_PASSED") if request.canary.get("passed") is not True: raise EvidenceProduceError("CANARY_NOT_PASSED") if request.rollback.get("ready") is not True: raise EvidenceProduceError("ROLLBACK_NOT_READY") if ( request.gitea.get("consistent") is not True or request.gitea.get("repository") != config.repository or request.gitea.get("source_commit") != source_commit or request.gitea.get("remote_ref_oid") != source_commit or request.gitea.get("tree_oid") != tree_oid ): raise EvidenceProduceError("GITEA_INCONSISTENT") digests = { "artifact": artifact_sha, "manifest": manifest_sha, "package_closure": closure_sha, "policy": policy_sha, "tests": _sha256_bytes(_canonical_json(dict(request.tests))), "canary": _sha256_bytes(_canonical_json(dict(request.canary))), "gitea": _sha256_bytes(_canonical_json(dict(request.gitea))), "rollback": _sha256_bytes(_canonical_json(dict(request.rollback))), "secret_scan": _sha256_bytes(_canonical_json(dict(request.secret_scan))), } return digests def _build_receipt( config: TrustedCandidateBuilderConfig, request: GovernedBuildArtifacts, ) -> Mapping[str, Any]: """Assemble the self-consistent receipt from verified artifacts.""" from silent_release_controller import ReleaseCandidate, ReleaseState artifact_sha = _sha256_bytes(request.artifact) manifest_sha = _sha256_bytes(_canonical_json(dict(request.manifest))) closure_sha = _validate_sha256( request.manifest.get("package_closure_sha256"), "package_closure_sha256" ) policy_sha = _sha256_bytes(_canonical_json(dict(request.policy))) candidate_kwargs: dict[str, Any] = {} issued_at = str(request.issued_at or "").strip() if issued_at: # The evidence was produced at the build execution instant; bind it so the # same verified artifacts reproduce an identical, idempotent closure. candidate_kwargs["created_at"] = issued_at candidate_kwargs["updated_at"] = issued_at candidate = ReleaseCandidate( candidate_id=request.candidate_id, source_commit=request.source_commit, tree_oid=request.tree_oid, artifact_sha256=artifact_sha, manifest_sha256=manifest_sha, tests=dict(request.tests), canary=dict(request.canary), rollback_ready=request.rollback.get("ready") is True, git_consistency={ "consistent": True, "repository": config.repository, "source_commit": request.source_commit, "tree_oid": request.tree_oid, }, gitea_consistency=dict(request.gitea), state=ReleaseState.READY_FOR_APPROVAL, **candidate_kwargs, ) deployment_payload = dict(request.deployment) deployment_payload.update( { "service": request.deployment["service"], "source_commit": request.source_commit, "tree_oid": request.tree_oid, "artifact_sha256": artifact_sha, "manifest_sha256": manifest_sha, "package_closure_sha256": closure_sha, "policy_sha256": policy_sha, "rollback_sha256": _validate_sha256( request.rollback_sha256, "rollback_sha256" ) or request.rollback.get("rollback_sha256", ""), "repository_identity": config.repository, "authoritative_git_dir": request.deployment["authoritative_git_dir"], } ) if not deployment_payload["rollback_sha256"]: raise EvidenceProduceError("ROLLBACK_SHA256_REQUIRED") from silent_release_hardening import ( DeploymentPlan, FinalizationPlan, ReleasePlan, TagPlan, TrustedCandidateBundle, ) release_plan = ReleasePlan( deployment=DeploymentPlan.from_dict(deployment_payload), tag=TagPlan.from_dict(dict(request.tag)), finalization=FinalizationPlan.from_dict(dict(request.finalization)), policy_sha256=policy_sha, ) checks = {name: True for name in config.required_checks} TrustedCandidateBundle(candidate, release_plan, checks).validate() return { "schema": EVIDENCE_RECEIPT_SCHEMA, "version": request.version, "candidate": candidate.to_dict(), "release_plan": release_plan.to_dict(), } def _safe_target(source_root: Path, version: str) -> Path: relative = Path(version) if relative.is_absolute() or any( part in {"", ".", ".."} for part in relative.parts ): raise EvidenceProduceError("CLOSURE_PATH_INVALID") candidate = Path(source_root) / version try: metadata = Path(source_root).stat(follow_symlinks=False) except OSError as exc: raise EvidenceProduceError("SOURCE_ROOT_UNAVAILABLE") from exc if not stat.S_ISDIR(metadata.st_mode) or Path(source_root).is_symlink(): raise EvidenceProduceError("SOURCE_ROOT_INVALID") return candidate def _write_durable_json(path: Path, payload: Mapping[str, Any], mode: int = 0o600) -> str: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) # The deterministic collector compares file digests against the canonical # payload hash (no trailing newline); write the exact canonical bytes so the # producer's closure re-validates under TrustedBuilderEvidenceMaterializer. raw = _canonical_json(payload) descriptor = os.open( path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0), mode, ) try: os.write(descriptor, raw) os.fsync(descriptor) finally: os.close(descriptor) os.chmod(path, mode) return hashlib.sha256(raw).hexdigest() def _write_durable_bytes(path: Path, value: bytes, mode: int = 0o600) -> str: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) descriptor = os.open( path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0), mode, ) try: os.write(descriptor, value) os.fsync(descriptor) finally: os.close(descriptor) os.chmod(path, mode) return hashlib.sha256(value).hexdigest() def produce_evidence_closure( *, builder_policy: Path, source_root: Path, request: GovernedBuildArtifacts, expected_owner_uid: int | None = 0, ) -> ProduceResult: """Produce a self-consistent, root-owned evidence closure, fail-closed and idempotent.""" try: config = TrustedCandidateBuilderConfig.from_policy( load_builder_policy(builder_policy) ) except Exception as exc: raise EvidenceProduceError(str(exc)) from exc try: version = validate_release_version(request.version) except ValueError as exc: raise EvidenceProduceError("VERSION_INVALID") from exc digests = _verify_artifacts(config, request) receipt = _build_receipt(config, request) # Resolve layout paths from policy (aligned with evidence_layout; never invented). target = _safe_target(source_root, version) layout_paths = { name: target / Path(template.format(version=version)).relative_to(version) for name, template in config.evidence_layout.items() } if target.exists() or target.is_symlink(): # Idempotency: an existing identical closure (same receipt digest + same # artifact bytes) is accepted; anything else is a conflict. existing_receipt = target / "receipt.json" if not existing_receipt.is_file(): raise EvidenceProduceError("CLOSURE_CONFLICT") try: existing = json.loads(existing_receipt.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise EvidenceProduceError("CLOSURE_CONFLICT") from exc if _canonical_json(dict(existing)) != _canonical_json(dict(receipt)): raise EvidenceProduceError("CLOSURE_CONFLICT") for name, path in layout_paths.items(): if not (path.is_file() and _file_sha256(path) == digests[name]): raise EvidenceProduceError("CLOSURE_CONFLICT") return ProduceResult( status="EVIDENCE_PRODUCED", idempotent=True, receipt_sha256=_sha256_bytes(_canonical_json(dict(receipt))), closure_dir=str(target), evidence_digests=dict(digests), ) temporary = source_root / ( f".{version}.{os.getpid()}.{secrets.token_hex(8)}.prod.tmp" ) if temporary.exists() or temporary.is_symlink(): raise EvidenceProduceError("PRODUCE_TEMP_CONFLICT") temporary.mkdir(mode=0o700) # The closure directory mirrors source_root//; every written file uses # the policy-declared basename so the layout is aligned with evidence_layout. try: _write_durable_json(temporary / "receipt.json", receipt) _write_durable_bytes(temporary / "artifact.tar.gz", request.artifact) _write_durable_json(temporary / "manifest.json", request.manifest) _write_durable_bytes(temporary / "checksums.sha256", request.package_closure) _write_durable_json(temporary / "trusted-release-policy.json", request.policy) _write_durable_json(temporary / "tests.json", request.tests) _write_durable_json(temporary / "canary.json", request.canary) _write_durable_json(temporary / "gitea.json", request.gitea) _write_durable_json(temporary / "rollback.json", request.rollback) _write_durable_json(temporary / "secret-scan.json", request.secret_scan) os.replace(temporary, target) except Exception: shutil.rmtree(temporary, ignore_errors=True) raise return ProduceResult( status="EVIDENCE_PRODUCED", idempotent=False, receipt_sha256=_sha256_bytes(_canonical_json(dict(receipt))), closure_dir=str(target), evidence_digests=dict(digests), ) def producer_operation_definition() -> OperationDefinition: """Register the governed.release.evidence.produce operation (lookup-only registry). This is the *legacy closure assembler* entrypoint. V2R2C2 Round-3 keeps it registered for backward compatibility with the existing runtime chain; the governed production path now resolves ``governed_authenticated_evidence_operation.produce_authenticated_evidence_closure`` (see ``authenticated_producer_operation_definition``). """ return OperationDefinition.from_dict( { "operation_id": "governed.release.evidence.produce", "version": 1, "effect_class": "RECONCILABLE", "handler_ref": "governed_release_evidence_producer.produce_evidence_closure", "input_schema": PRODUCE_SCHEMA, "required_permissions": ["governed.release.evidence.produce"], "timeout_seconds": 60.0, "readback": {"mode": "STATE", "required": True}, "retry": {"mode": "NEVER"}, "reconcile": {"mode": "EXACT_READBACK"}, "rollback": {"mode": "COMPENSATE"}, "owner_boundary": {"mode": "WITHIN_GRANT"}, } ) AUTHENTICATED_EVIDENCE_HANDLER_REF = ( "governed_authenticated_evidence_operation.produce_authenticated_evidence_closure" ) def authenticated_producer_operation_definition() -> OperationDefinition: """Register the governed, provider-authenticated release evidence operation. V2R2C2 Round-3 production entrypoint. Its ``handler_ref`` points at the authenticated wrapper, which refuses to invoke the legacy closure assembler until every final-release provider receipt has been resolved from an authenticated store (Gate-W only authenticates tests/Gitea/secret-scan; canary/rollback require Gate-R). See Round-3 Provider Authentication. """ return OperationDefinition.from_dict( { "operation_id": "governed.authenticated.release.evidence.produce", "version": 1, "effect_class": "RECONCILABLE", "handler_ref": AUTHENTICATED_EVIDENCE_HANDLER_REF, "input_schema": PRODUCE_SCHEMA, "required_permissions": ["governed.release.evidence.produce"], "timeout_seconds": 60.0, "readback": {"mode": "STATE", "required": True}, "retry": {"mode": "NEVER"}, "reconcile": {"mode": "EXACT_READBACK"}, "rollback": {"mode": "COMPENSATE"}, "owner_boundary": {"mode": "WITHIN_GRANT"}, } ) def producer_registry() -> OperationRegistry: """Union of the legacy assembler and the authenticated production operation.""" return OperationRegistry( [producer_operation_definition(), authenticated_producer_operation_definition()] ) def _main() -> None: parser = argparse.ArgumentParser() parser.add_argument("operation", choices=("produce",)) parser.add_argument("--version", required=True) parser.add_argument("--builder-policy", required=True, type=Path) parser.add_argument("--source-root", required=True, type=Path) parser.add_argument("--artifacts-json", required=True, type=Path) arguments = parser.parse_args() try: payload = json.loads(arguments.artifacts_json.read_text(encoding="utf-8")) request = GovernedBuildArtifacts( version=arguments.version, candidate_id=payload["candidate_id"], source_commit=payload["source_commit"], tree_oid=payload["tree_oid"], artifact=bytes.fromhex(payload["artifact_hex"]), manifest=payload["manifest"], package_closure=bytes.fromhex(payload["package_closure_hex"]), policy=payload["policy"], tests=payload["tests"], canary=payload["canary"], gitea=payload["gitea"], rollback=payload["rollback"], secret_scan=payload["secret_scan"], deployment=payload["deployment"], tag=payload["tag"], finalization=payload["finalization"], rollback_sha256=payload.get("rollback_sha256", ""), ) result = produce_evidence_closure( builder_policy=arguments.builder_policy, source_root=arguments.source_root, request=request, ) except (EvidenceProduceError, KeyError, ValueError, OSError, json.JSONDecodeError) as exc: raise SystemExit(str(exc)) from exc from dataclasses import asdict print(json.dumps(asdict(result), sort_keys=True, separators=(",", ":"))) if __name__ == "__main__": _main()