130 lines
7.2 KiB
Python
130 lines
7.2 KiB
Python
#!/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)
|