#!/usr/bin/env python3 """Validate the redacted v1 readiness record and emit the release decision.""" from __future__ import annotations import json import pathlib import re import sys from datetime import datetime, timedelta from typing import Any LOCAL_GATES = { "immutable-release-artifacts", "debug-and-release-verification", "real-consumer-pilots", "candidate-capacity-resilience", "production-process-recovery", "security-privacy-observability", } EXTERNAL_GATES = { "public-package-empty-cache-restore", "signed-publication", "source-preserving-udp-ingress", "same-lan-direct-canary", "home-nat-direct-canary", "restrictive-cgnat-typed-failure", "firewall-blocked-udp-typed-failure", "ipv6-direct-canary", "public-rate-shaped-capacity", "one-hour-candidate-endurance", "alert-delivery", "cold-standby-rollback-drill", "documentation-only-runbook-exercise", } STATUSES = {"pass", "pending", "fail"} FORBIDDEN_KEY_PARTS = { "address", "credential", "endpoint", "listingid", "password", "playerid", "secret", "token", "userid", } UUID = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F-]{27,}\b") IPV4 = re.compile(r"(? None: if isinstance(value, dict): for key, child in value.items(): normalized = re.sub(r"[^a-z0-9]", "", key.lower()) if any(part in normalized for part in FORBIDDEN_KEY_PARTS): raise InvalidRecord(f"{path}.{key} uses a forbidden sensitive-data key") reject_sensitive(child, f"{path}.{key}") elif isinstance(value, list): for index, child in enumerate(value): reject_sensitive(child, f"{path}[{index}]") elif isinstance(value, str): if UUID.search(value) or IPV4.search(value) or IPV6.search(value) \ or "://" in value or "@" in value: raise InvalidRecord(f"{path} contains endpoint, identifier, or account-shaped data") def evidence_path(repository_root: pathlib.Path, value: str, path: str) -> pathlib.Path: relative = pathlib.PurePosixPath(value) if relative.is_absolute() or ".." in relative.parts or not value: raise InvalidRecord(f"{path} must be a repository-relative reference") candidate = (repository_root / pathlib.Path(*relative.parts)).resolve() if not candidate.is_relative_to(repository_root.resolve()) or not candidate.is_file(): raise InvalidRecord(f"{path} does not resolve to a repository evidence file") return candidate def load_json(path: pathlib.Path, label: str) -> Any: try: with path.open("r", encoding="utf-8") as source: return json.load(source) except (OSError, json.JSONDecodeError) as error: raise InvalidRecord(f"{label} is not readable JSON: {error}") from error def validate_gate_set( items: Any, expected: set[str], path: str, repository_root: pathlib.Path, ) -> list[dict[str, str]]: if not isinstance(items, list): raise InvalidRecord(f"{path} must be an array") gates: list[dict[str, str]] = [] for index, item in enumerate(items): if not isinstance(item, dict) or set(item) != {"id", "status", "evidenceRef", "note"}: raise InvalidRecord(f"{path}[{index}] has an invalid shape") if not all(isinstance(item[key], str) for key in item): raise InvalidRecord(f"{path}[{index}] fields must be strings") if item["status"] not in STATUSES: raise InvalidRecord(f"{path}[{index}] has an invalid status") evidence_path(repository_root, item["evidenceRef"], f"{path}[{index}].evidenceRef") if len(item["note"]) > 240: raise InvalidRecord(f"{path}[{index}].note is too long") gates.append(item) identifiers = [gate["id"] for gate in gates] if len(identifiers) != len(set(identifiers)): raise InvalidRecord(f"{path} contains duplicate gate identifiers") if set(identifiers) != expected: missing = sorted(expected - set(identifiers)) extra = sorted(set(identifiers) - expected) raise InvalidRecord(f"{path} gate mismatch; missing={missing}, extra={extra}") return gates def validate_local_evidence( record: dict[str, Any], gates: list[dict[str, str]], repository_root: pathlib.Path, ) -> None: if any(gate["status"] != "pass" for gate in gates): return commit = record["evaluatedCommit"] release_path = evidence_path( repository_root, "docs/evidence/releases/v1.0.0-local-candidate.json", "local release evidence", ) release = load_json(release_path, "local release evidence") if not isinstance(release, dict) or release.get("schemaVersion") != 1 \ or release.get("kind") != "rendezvous-local-release-candidate" \ or release.get("sourceCommit") != commit \ or release.get("treeState") != "clean" \ or release.get("result") != "pass": raise InvalidRecord("local release evidence is not a passing clean build of evaluatedCommit") verification = release.get("verification") if not isinstance(verification, dict): raise InvalidRecord("local release evidence has no verification object") exact_passes = { "lockedRestore": "pass", "format": "pass", "byteReproduciblePackages": "pass", "byteReproducibleServerArchive": "pass", "sbomChecksumsAndProvenance": "pass", "candidateConsumerFixtures": "pass", "realConsumerRestores": "pass", } if any(verification.get(key) != value for key, value in exact_passes.items()) \ or verification.get("reportedVulnerabilities") != 0 \ or verification.get("releaseBuildWarnings") != 0 \ or verification.get("releaseBuildErrors") != 0 \ or verification.get("debugTestsPassed", 0) < 300 \ or verification.get("debugTestsFailed") != 0 \ or verification.get("releaseTestsPassed", 0) < 300 \ or verification.get("releaseTestsFailed") != 0 \ or verification.get("selectedProductionFaultTestsPassed", 0) < 17: raise InvalidRecord("local release evidence does not satisfy every required verification") consumers = release.get("consumers") if not isinstance(consumers, list) or { item.get("name") for item in consumers if isinstance(item, dict) } != {"SpaceGame", "Unscouted"} or any( not isinstance(item, dict) or item.get("candidateRestore") != "pass" or item.get("directTrafficPilot") != "pass" for item in consumers ): raise InvalidRecord("local release evidence does not prove both required consumers") capacity_path = evidence_path( repository_root, "docs/evidence/capacity/v2/candidate-2cpu.json", "candidate capacity evidence", ) capacity = load_json(capacity_path, "candidate capacity evidence") runtime = capacity.get("runtime") if isinstance(capacity, dict) else None state = capacity.get("state") if isinstance(capacity, dict) else None if not isinstance(runtime, dict) or not isinstance(state, dict) \ or capacity.get("schemaVersion") != 2 \ or capacity.get("profile") != "candidate" \ or capacity.get("passed") is not True \ or capacity.get("failures") != [] \ or runtime.get("commitSha") != commit \ or runtime.get("treeState") != "clean" \ or runtime.get("processorCount") != 2 \ or state.get("soakDurationSeconds", 0) < 300 \ or state.get("finalListings") != 0 \ or state.get("finalAttempts") != 0 \ or state.get("finalReplayMarkers") != 0 \ or state.get("restartStartedEmpty") is not True \ or state.get("overloadWasTyped") is not True \ or state.get("recoverySucceeded") is not True: raise InvalidRecord("candidate capacity evidence does not satisfy the clean evaluated commit") def validate_external_attestations( record: dict[str, Any], gates: list[dict[str, str]], repository_root: pathlib.Path, ) -> None: for gate in gates: if gate["status"] != "pass": continue path = evidence_path(repository_root, gate["evidenceRef"], f"{gate['id']} evidence") attestation = load_json(path, f"{gate['id']} evidence") if not isinstance(attestation, dict) or set(attestation) != { "schemaVersion", "kind", "gateId", "candidateCommit", "result", "performedAtUtc", "artifactDigest", "evidenceLocation", "reviewerRole", }: raise InvalidRecord(f"{gate['id']} requires a complete external-gate attestation") reject_sensitive(attestation, f"external evidence {gate['id']}") if attestation["schemaVersion"] != 1 \ or attestation["kind"] != "rendezvous-external-gate-attestation" \ or attestation["gateId"] != gate["id"] \ or attestation["candidateCommit"] != record["evaluatedCommit"] \ or attestation["result"] != "pass" \ or not isinstance(attestation["artifactDigest"], str) \ or not DIGEST.fullmatch(attestation["artifactDigest"]) \ or attestation["evidenceLocation"] not in { "protected-operations-record", "public-release-record", } \ or attestation["reviewerRole"] not in { "release-operator", "network-operator", "security-operator", "independent-operator", }: raise InvalidRecord(f"{gate['id']} external attestation does not match the candidate gate") try: performed = datetime.fromisoformat(attestation["performedAtUtc"].replace("Z", "+00:00")) except (AttributeError, ValueError) as error: raise InvalidRecord(f"{gate['id']} has an invalid performedAtUtc") from error if performed.tzinfo is None or performed.utcoffset() != timedelta(0): raise InvalidRecord(f"{gate['id']} performedAtUtc must be UTC") def validate(record: Any, repository_root: pathlib.Path) -> tuple[bool, list[str]]: if not isinstance(record, dict) or set(record) != { "schemaVersion", "kind", "evaluatedCommit", "decision", "localGates", "externalGates", }: raise InvalidRecord("The top-level readiness record shape is invalid") if record["schemaVersion"] != 1 or record["kind"] != "rendezvous-production-readiness": raise InvalidRecord("The readiness schema identity is invalid") if not isinstance(record["evaluatedCommit"], str) or not COMMIT.fullmatch(record["evaluatedCommit"]): raise InvalidRecord("evaluatedCommit must be a full lowercase Git commit") reject_sensitive(record) local_gates = validate_gate_set( record["localGates"], LOCAL_GATES, "$.localGates", repository_root ) external_gates = validate_gate_set( record["externalGates"], EXTERNAL_GATES, "$.externalGates", repository_root ) validate_local_evidence(record, local_gates, repository_root) validate_external_attestations(record, external_gates, repository_root) gates = local_gates + external_gates blockers = sorted(gate["id"] for gate in gates if gate["status"] != "pass") ready = not blockers expected_decision = "ready" if ready else "not-ready" if record["decision"] != expected_decision: raise InvalidRecord( f"decision must be {expected_decision!r} for the recorded gate statuses" ) return ready, blockers def main() -> int: if len(sys.argv) != 2: print("usage: check_production_readiness.py RECORD", file=sys.stderr) return 2 try: with open(sys.argv[1], "r", encoding="utf-8") as source: record = json.load(source) ready, blockers = validate(record, pathlib.Path(__file__).resolve().parent.parent) except (OSError, json.JSONDecodeError, InvalidRecord) as error: print(f"INVALID: {error}", file=sys.stderr) return 2 if not ready: print(f"NOT READY: {len(blockers)} required gate(s) are not passing.") for blocker in blockers: print(f"- {blocker}") return 3 print("READY: every required v1 production gate is recorded as passing.") return 0 if __name__ == "__main__": raise SystemExit(main())