#!/usr/bin/env python3 """Validate a GuestOps supervised-pilot go/no-go record.""" from __future__ import annotations import argparse import datetime as dt import json from pathlib import Path import re GATE_B = { "release-ci", "debian-host", "persistence", "backup-restore", "google-mailbox", "automation", "identity-privacy", "inbox-usability", "capacity", "incident-support", "pilot-findings", } GATE_C = GATE_B | {"pms-provider", "payment-provider"} def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def timestamp(value: object, field: str) -> dt.datetime: require(isinstance(value, str) and value.endswith("Z"), f"{field} must be a UTC timestamp ending in Z.") try: parsed = dt.datetime.fromisoformat(value[:-1] + "+00:00") except ValueError as error: raise ValueError(f"{field} is not a valid timestamp.") from error return parsed def validate(record: object) -> None: require(isinstance(record, dict), "Approval record must be a JSON object.") require(record.get("schemaVersion") == 2, "Unsupported approval schema; Gate B 0.2.0 requires schemaVersion 2.") gate = record.get("targetGate") require(gate in ("B", "C"), "targetGate must be B or C.") require(record.get("decision") == "approved", "Only an explicit approved decision passes validation.") require(re.fullmatch(r"[0-9a-f]{40}", str(record.get("releaseCommit", ""))) is not None, "releaseCommit must be a full lowercase Git SHA.") require(re.fullmatch(r"[0-9a-f]{64}", str(record.get("releaseRecordSha256", ""))) is not None, "releaseRecordSha256 must be a SHA-256 digest.") decided = timestamp(record.get("decidedAt"), "decidedAt") approvals = record.get("approvals") require(isinstance(approvals, dict) and set(approvals) == {"hotelOwner", "technicalOwner"}, "Separate hotelOwner and technicalOwner approvals are required.") approver_names = [] for role, approval in approvals.items(): name = str(approval.get("name", "")).strip() if isinstance(approval, dict) else "" require(2 <= len(name) <= 120 and "@" not in name, f"{role} approval requires a named owner without an email address.") approver_names.append(name.casefold()) require(timestamp(approval.get("approvedAt"), f"{role}.approvedAt") <= decided, f"{role} approval cannot occur after the decision.") require(len(set(approver_names)) == 2, "hotelOwner and technicalOwner must be different people.") evidence = record.get("evidence") require(isinstance(evidence, list), "evidence must be a list.") ids = [item.get("id") for item in evidence if isinstance(item, dict)] required = GATE_C if gate == "C" else GATE_B require(len(ids) == len(evidence) and len(ids) == len(set(ids)) and set(ids) == required, f"Gate {gate} requires the exact evidence set.") for item in evidence: item_id = item["id"] status = item.get("status") require(status in ("pass", "contained"), f"Evidence {item_id} must pass or have approved containment.") references = item.get("references") require(isinstance(references, list) and 1 <= len(references) <= 10 and all(isinstance(value, str) and 3 <= len(value) <= 200 and "@" not in value for value in references), f"Evidence {item_id} requires safe opaque references without email addresses.") if status == "contained": containment = item.get("containment") require(isinstance(containment, dict), f"Evidence {item_id} requires containment details.") require(2 <= len(str(containment.get("owner", ""))) <= 120, f"Evidence {item_id} containment requires an owner.") require(timestamp(containment.get("expiresAt"), f"{item_id}.containment.expiresAt") > decided, f"Evidence {item_id} containment must expire after the decision.") require(5 <= len(str(containment.get("rollbackTrigger", ""))) <= 300, f"Evidence {item_id} containment requires a rollback trigger.") capacity = record.get("capacity") require(isinstance(capacity, dict), "Capacity thresholds and observations are required.") require(re.fullmatch(r"[0-9a-f]{64}", str(capacity.get("reportSha256", ""))) is not None, "capacity.reportSha256 must identify the retained probe report.") require(re.fullmatch(r"[0-9a-f]{64}", str(capacity.get("hostMetricsSha256", ""))) is not None, "capacity.hostMetricsSha256 must identify the retained host metrics.") for observed, target in (("observedP95Ms", "targetP95Ms"), ("observedErrorRate", "targetErrorRate")): values = (capacity.get(observed), capacity.get(target)) require(all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values) and 0 <= capacity[observed] <= capacity[target], f"Capacity {observed} must be within its approved {target}.") require(capacity["targetP95Ms"] > 0, "Capacity targetP95Ms must be positive.") require(capacity["targetErrorRate"] <= 1, "Capacity targetErrorRate must be a ratio no greater than 1.") concurrency = (capacity.get("observedConcurrency"), capacity.get("targetConcurrency")) require(all(isinstance(value, int) and not isinstance(value, bool) for value in concurrency) and capacity["observedConcurrency"] >= capacity["targetConcurrency"] > 0, "Observed concurrency must meet the approved positive target.") for resource in ("Cpu", "Memory"): target = capacity.get(f"target{resource}HeadroomPercent") observed = capacity.get(f"observed{resource}HeadroomPercent") require(all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in (target, observed)) and 25 <= target <= observed <= 100, f"Observed {resource.lower()} headroom must meet the approved target of at least 25 percent.") pilot = record.get("pilot") require(isinstance(pilot, dict), "A supervised pilot summary is required.") require(re.fullmatch(r"[0-9a-f]{64}", str(pilot.get("recordSha256", ""))) is not None, "pilot.recordSha256 must identify the retained pilot run record.") require(pilot.get("businessDaysObserved") == 5, "The pilot must cover exactly five business days.") require(pilot.get("hotelsObserved") == 1, "The Gate B pilot must cover exactly one hotel.") require(pilot.get("stopConditionsObserved") == 0, "A pilot with a stop condition cannot be approved.") require(pilot.get("unresolvedFindings") == 0, "All pilot findings must be resolved or explicitly contained.") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("record", type=Path) args = parser.parse_args() validate(json.loads(args.record.read_text(encoding="utf-8"))) print("Pilot approval record is structurally complete and approved. This validates the record, not its restricted evidence.") if __name__ == "__main__": try: main() except (OSError, ValueError, json.JSONDecodeError) as error: print(f"Pilot approval rejected: {error}", file=__import__("sys").stderr) raise SystemExit(1)