#!/usr/bin/env python3 """Validate the complete release-bound GuestOps Gate B acceptance bundle.""" from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import re import sys from urllib.parse import urlparse DEPLOY = Path(__file__).resolve().parent if str(DEPLOY) not in sys.path: sys.path.insert(0, str(DEPLOY)) import automation_acceptance import backup_restore_acceptance import debian_acceptance import desktop_acceptance import google_acceptance import identity_privacy_acceptance import incident_exercise import pilot_approval import pilot_run import verify_release import verify_source_package VERSION = "0.2.1" SHA256 = re.compile(r"[0-9a-f]{64}") RECORD_KEYS = { "debian-host", "persistence", "backup-restore", "google-mailbox", "automation", "identity-privacy", "inbox-usability", "capacity", "incident-support", "pilot-findings", "pilot-approval", } def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def digest(path: Path) -> str: with path.open("rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def load_json(path: Path) -> object: require(path.is_file() and not path.is_symlink(), f"Required bundle file is missing or is a symlink: {path.name}") return json.loads(path.read_text(encoding="utf-8")) def release_binding(record: object, name: str, commit: str, release_sha: str) -> None: require(isinstance(record, dict), f"{name} must be a JSON object.") require(record.get("releaseCommit") == commit, f"{name} uses a different releaseCommit.") require(record.get("releaseRecordSha256") == release_sha, f"{name} uses a different releaseRecordSha256.") def environment_origin(record: object, name: str) -> str: require(isinstance(record, dict), f"{name} must be a JSON object.") value = str(record.get("environment", "")).rstrip("/") parsed = urlparse(value) require(parsed.scheme == "https" and parsed.hostname and parsed.path in ("", "/"), f"{name} has no valid HTTPS environment.") return value def validate_capacity(report: object, commit: str, release_sha: str) -> None: require(isinstance(report, dict), "capacity must be a JSON object.") require(report.get("schemaVersion") == 1 and report.get("kind") == "guestops-read-only-capacity", "capacity has an unsupported schema or kind.") require(report.get("releaseCommit") == commit, "capacity uses a different releaseCommit.") require(report.get("releaseRecordSha256") == release_sha, "capacity uses a different releaseRecordSha256.") require(report.get("paths") == ["/health/ready", "/api/hotel", "/api/conversations/page"], "capacity must use the exact read-only path set.") for field in ("concurrency", "requests", "successes", "failures"): require(isinstance(report.get(field), int) and not isinstance(report.get(field), bool), f"capacity.{field} must be an integer.") require(1 <= report["concurrency"] <= 20 and report["requests"] > 0, "capacity has invalid concurrency or request count.") require(report["successes"] + report["failures"] == report["requests"], "capacity success and failure counts do not match requests.") expected_error = round(report["failures"] / report["requests"], 6) require(report.get("errorRate") == expected_error, "capacity.errorRate does not match its counts.") latency = report.get("latencyMs") require(isinstance(latency, dict) and set(latency) == {"median", "p95", "maximum"} and all(isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0 for value in latency.values()), "capacity.latencyMs is invalid.") require(latency["median"] <= latency["p95"] <= latency["maximum"], "capacity latency percentiles are out of order.") def validate_bundle( source_archive: Path, source_record_path: Path, release_record_path: Path, records: dict[str, tuple[Path, object]], host_metrics_path: Path, ) -> dict[str, object]: require(set(records) == RECORD_KEYS, "Gate B bundle requires the exact record set.") release_record = load_json(release_record_path) require(isinstance(release_record, dict), "release-record must be a JSON object.") commit = str(release_record.get("commit", "")) version = str(release_record.get("version", "")) require(version == VERSION, f"Gate B bundle version must be {VERSION}.") release_result = verify_release.validate(source_archive, release_record_path, commit, version) source_result = verify_source_package.validate(source_archive, source_record_path, commit, version) release_sha = str(release_result["releaseRecord"]["sha256"]) require(source_result["artifact"]["sha256"] == release_result["archive"]["sha256"], "Source and release records identify different archives.") values = {name: value for name, (_, value) in records.items()} for name, record in values.items(): release_binding(record, name, commit, release_sha) debian_acceptance.validate_pair(values["debian-host"], values["persistence"], commit, release_sha) backup_restore_acceptance.validate(values["backup-restore"], commit, release_sha) google_acceptance.validate(values["google-mailbox"]) automation_acceptance.validate(values["automation"]) identity_privacy_acceptance.validate(values["identity-privacy"]) desktop_acceptance.validate(values["inbox-usability"]) validate_capacity(values["capacity"], commit, release_sha) incident_exercise.validate(values["incident-support"]) pilot_run.validate(values["pilot-findings"]) pilot_approval.validate(values["pilot-approval"]) environment_names = { "debian-host", "persistence", "backup-restore", "google-mailbox", "automation", "identity-privacy", "inbox-usability", "incident-support", "pilot-findings", } origins = {environment_origin(values[name], name) for name in environment_names} require(len(origins) == 1, "Gate B records use different environments.") origin = next(iter(origins)) require(values["capacity"].get("originHost") == urlparse(origin).hostname, "capacity originHost does not match the accepted environment.") archive_sha = str(release_result["archive"]["sha256"]) for name in ("debian-host", "persistence", "backup-restore"): require(values[name].get("archiveSha256") == archive_sha, f"{name} uses a different archiveSha256.") approved_images = release_record["images"] for name in ("debian-host", "persistence", "backup-restore"): images = values[name].get("images", {}) require(images.get("api") == approved_images["api"] and images.get("worker") == approved_images["worker"], f"{name} uses different API or worker image identities.") approval_capacity = values["pilot-approval"]["capacity"] capacity = values["capacity"] require(approval_capacity["reportSha256"] == digest(records["capacity"][0]), "Pilot approval capacity report checksum does not match the retained report.") require(approval_capacity["hostMetricsSha256"] == digest(host_metrics_path), "Pilot approval host metrics checksum does not match the retained file.") require(approval_capacity["observedConcurrency"] == capacity["concurrency"], "Pilot approval concurrency does not match the capacity report.") require(approval_capacity["observedP95Ms"] == capacity["latencyMs"]["p95"], "Pilot approval p95 does not match the capacity report.") require(approval_capacity["observedErrorRate"] == capacity["errorRate"], "Pilot approval error rate does not match the capacity report.") require(values["pilot-approval"]["pilot"]["recordSha256"] == digest(records["pilot-findings"][0]), "Pilot approval checksum does not match the retained pilot run record.") summary_records = {name: digest(path) for name, (path, _) in sorted(records.items())} return { "archiveSha256": archive_sha, "environmentHost": urlparse(origin).hostname, "releaseCommit": commit, "releaseRecordSha256": release_sha, "records": summary_records, "schemaVersion": 1, "sourceRecordSha256": source_result["sourceRecord"]["sha256"], "validated": True, "version": version, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source-archive", required=True, type=Path) parser.add_argument("--source-record", required=True, type=Path) parser.add_argument("--release-record", required=True, type=Path) parser.add_argument("--host-metrics", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) for name in sorted(RECORD_KEYS): parser.add_argument("--" + name, required=True, type=Path) args = parser.parse_args() try: require(args.host_metrics.is_file() and not args.host_metrics.is_symlink(), "Host metrics file is missing or is a symlink.") require(not args.output.exists() and args.output.parent.is_dir(), "Output must be a new file in an existing restricted directory.") records = {} for name in RECORD_KEYS: path = getattr(args, name.replace("-", "_")) records[name] = (path, load_json(path)) summary = validate_bundle( args.source_archive, args.source_record, args.release_record, records, args.host_metrics, ) descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as output: output.write(json.dumps(summary, indent=2, sort_keys=True) + "\n") print("Gate B bundle is structurally complete, consistently release-bound and approved. " "This validates records and checksums, not the underlying restricted evidence.") except (OSError, ValueError, json.JSONDecodeError) as error: print(f"Gate B bundle rejected: {error}", file=sys.stderr) raise SystemExit(1) if __name__ == "__main__": main()