#!/usr/bin/env python3 """Validate a restricted five-business-day GuestOps pilot run record.""" from __future__ import annotations import argparse import datetime as dt import json from pathlib import Path import re from urllib.parse import urlparse STOP_CONDITIONS = { "tenant-leakage", "credential-exposure", "data-loss", "unapproved-send", "duplicate-send", "unreconciled-uncertain-send", "failed-rollback", "monitoring-loss", } def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def date_value(value: object, field: str) -> dt.date: require(isinstance(value, str), f"{field} must be an ISO date.") try: return dt.date.fromisoformat(value) except ValueError as error: raise ValueError(f"{field} must be an ISO date.") from error 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: return dt.datetime.fromisoformat(value.removesuffix("Z") + "+00:00") except ValueError as error: raise ValueError(f"{field} is not a valid timestamp.") from error def safe_name(value: object, field: str) -> str: name = str(value or "").strip() require(2 <= len(name) <= 120 and "@" not in name, f"{field} requires a name without an email address.") return name def references(value: object, field: str) -> None: require(isinstance(value, list) and 1 <= len(value) <= 10 and all( isinstance(item, str) and 3 <= len(item) <= 200 and "@" not in item for item in value ), f"{field} requires one to ten safe opaque references without email addresses.") def business_dates(start: dt.date, end: dt.date) -> list[dt.date]: days = [] current = start while current <= end: if current.weekday() < 5: days.append(current) current += dt.timedelta(days=1) return days def validate(record: object) -> None: require(isinstance(record, dict), "Pilot run record must be a JSON object.") require(record.get("schemaVersion") == 1, "Unsupported pilot run schema.") require(record.get("system") == "guestops-supervised-pilot", "Pilot run system must be guestops-supervised-pilot.") require(record.get("targetGate") == "B", "This pilot run record is restricted to Gate B.") 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.") origin = urlparse(str(record.get("environment", ""))) require(origin.scheme == "https" and origin.hostname and origin.path in ("", "/") and not origin.query and not origin.fragment and origin.username is None and origin.password is None, "environment must be an HTTPS origin without credentials, path, query or fragment.") label = str(record.get("hotelLabel", "")).strip() require(3 <= len(label) <= 80 and "@" not in label, "hotelLabel must be a non-email alias.") require(record.get("hotelCount") == 1, "Gate B pilot must contain exactly one hotel.") require(record.get("plannedBusinessDays") == 5, "Gate B pilot must require five business days.") owners = record.get("owners") require(isinstance(owners, dict) and set(owners) == {"hotelOwner", "technicalOwner", "rollbackDecisionMaker"}, "owners must contain hotelOwner, technicalOwner and rollbackDecisionMaker.") names = {role: safe_name(value, f"owners.{role}") for role, value in owners.items()} require(names["hotelOwner"].casefold() != names["technicalOwner"].casefold(), "hotelOwner and technicalOwner must be different people.") start = date_value(record.get("startedOn"), "startedOn") end = date_value(record.get("endedOn"), "endedOn") expected_days = business_dates(start, end) require(len(expected_days) == 5, "Pilot window must contain exactly five business days.") reviews = record.get("dailyReviews") require(isinstance(reviews, list) and len(reviews) == 5, "Exactly five daily reviews are required.") review_dates = [] for index, review in enumerate(reviews): require(isinstance(review, dict), f"dailyReviews[{index}] must be an object.") review_date = date_value(review.get("date"), f"dailyReviews[{index}].date") review_dates.append(review_date) require(review.get("status") == "pass", f"Daily review {review_date} has not passed.") safe_name(review.get("reviewedBy"), f"dailyReviews[{index}].reviewedBy") references(review.get("evidence"), f"dailyReviews[{index}].evidence") require(review_dates == expected_days, "Daily reviews must cover each business day in chronological order.") controls = record.get("pilotControls") require(controls == {"pmsWrites": "disabled", "paymentCreation": "disabled", "faqMode": "off", "googleReviewedSending": "accepted"}, "Pilot controls require accepted reviewed Google sending with PMS, payments and FAQ live mode disabled.") stop_conditions = record.get("stopConditions") require(isinstance(stop_conditions, dict) and set(stop_conditions) == STOP_CONDITIONS, "stopConditions must contain the exact Gate B stop-condition set.") require(all(value is False for value in stop_conditions.values()), "A pilot with an observed stop condition cannot pass.") findings = record.get("findings") require(isinstance(findings, list), "findings must be a list, including an empty list when none were found.") finding_ids = [] for finding in findings: require(isinstance(finding, dict), "Each finding must be an object.") finding_id = str(finding.get("id", "")) require(re.fullmatch(r"[a-z0-9][a-z0-9-]{2,79}", finding_id) is not None, "Finding IDs must be safe opaque identifiers.") finding_ids.append(finding_id) severity = finding.get("severity") disposition = finding.get("disposition") require(severity in ("critical", "high", "medium", "low"), f"Finding {finding_id} has an invalid severity.") require(disposition in ("resolved", "contained"), f"Finding {finding_id} must be resolved or contained.") references(finding.get("evidence"), f"Finding {finding_id} evidence") require(not (severity in ("critical", "high") and disposition == "contained"), f"Finding {finding_id} is too severe for containment.") if disposition == "contained": containment = finding.get("containment") require(isinstance(containment, dict), f"Finding {finding_id} requires containment details.") safe_name(containment.get("owner"), f"Finding {finding_id} containment owner") require(timestamp(containment.get("expiresAt"), f"Finding {finding_id}.containment.expiresAt").date() > end, f"Finding {finding_id} containment must expire after the pilot.") require(5 <= len(str(containment.get("rollbackTrigger", ""))) <= 300, f"Finding {finding_id} containment requires a rollback trigger.") require(len(finding_ids) == len(set(finding_ids)), "Finding IDs must be unique.") require(record.get("postPilotState") == {"pmsWrites": "disabled", "paymentCreation": "disabled", "faqMode": "off"}, "Pilot must end with PMS writes, payment creation and FAQ live mode disabled.") 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("Supervised pilot record is structurally complete: five business days passed without a stop condition. This validates the record, not its restricted evidence.") if __name__ == "__main__": try: main() except (OSError, ValueError, json.JSONDecodeError) as error: print(f"Supervised pilot record rejected: {error}", file=__import__("sys").stderr) raise SystemExit(1)