Rendezvous/eng/check_production_readiness.py

#!/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 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"(?<![0-9])(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?![0-9])")
COMMIT = re.compile(r"[0-9a-f]{40}")


class InvalidRecord(ValueError):
    pass


def reject_sensitive(value: Any, path: str = "$") -> 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 "://" in value or "@" in value:
            raise InvalidRecord(f"{path} contains endpoint, identifier, or account-shaped data")


def validate_gate_set(items: Any, expected: set[str], path: str) -> 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 = pathlib.PurePosixPath(item["evidenceRef"])
        if evidence.is_absolute() or ".." in evidence.parts or not item["evidenceRef"]:
            raise InvalidRecord(f"{path}[{index}].evidenceRef must be a repository-relative reference")
        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(record: Any) -> 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)
    gates = validate_gate_set(record["localGates"], LOCAL_GATES, "$.localGates")
    gates += validate_gate_set(record["externalGates"], EXTERNAL_GATES, "$.externalGates")
    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)
    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())