164 lines
8.1 KiB
Python
164 lines
8.1 KiB
Python
#!/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)
|