#!/usr/bin/env python3 """Validate restricted GuestOps Debian-host and persistence acceptance records.""" from __future__ import annotations import argparse import datetime as dt import json from pathlib import Path import re from urllib.parse import urlparse VERSION = "0.2.0" SYSTEMS = { "guestops-debian-host": { "evidenceId": "debian-host", "scenarios": { "host-baseline", "network-exposure", "https-and-redirect", "proxy-trust", "boot-services", "controlled-reboot", "workspace-health", "durable-log-retrieval", "secret-free-logs", }, }, "guestops-persistence": { "evidenceId": "persistence", "scenarios": { "separate-volume-layout", "service-restart", "container-recreation", "database-inventory", "data-protection-key", "image-identity", "post-reboot-persistence", }, }, } SHA256 = re.compile(r"[0-9a-f]{64}") GIT_SHA = re.compile(r"[0-9a-f]{40}") 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.removesuffix("Z") + "+00:00") except ValueError as error: raise ValueError(f"{field} is not a valid timestamp.") from error require(parsed.tzinfo == dt.timezone.utc, f"{field} must be UTC.") return parsed def safe_text(value: object, field: str, minimum: int = 2, maximum: int = 160) -> str: text = str(value or "").strip() require(minimum <= len(text) <= maximum and "@" not in text and "\\" not in text, f"{field} must be safe text without an email address or local path.") return text def validate_common(record: object, expected_commit: str, expected_release_sha256: str) -> str: require(isinstance(record, dict), "Acceptance record must be a JSON object.") require(record.get("schemaVersion") == 1, "Unsupported Debian acceptance schema.") system = record.get("system") require(system in SYSTEMS, "Unknown Debian acceptance record system.") require(record.get("evidenceId") == SYSTEMS[system]["evidenceId"], f"{system} has the wrong evidenceId.") require(record.get("releaseVersion") == VERSION, f"releaseVersion must be {VERSION}.") require(GIT_SHA.fullmatch(str(expected_commit)) is not None, "Expected release commit must be a full lowercase Git SHA.") require(SHA256.fullmatch(str(expected_release_sha256)) is not None, "Expected release-record checksum must be a lowercase SHA-256 digest.") require(record.get("releaseCommit") == expected_commit, "releaseCommit does not match the approved candidate.") require(record.get("releaseRecordSha256") == expected_release_sha256, "releaseRecordSha256 does not match the retained release record.") require(SHA256.fullmatch(str(record.get("archiveSha256", ""))) is not None, "archiveSha256 must be a lowercase 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.") host = safe_text(record.get("hostIdentifier"), "hostIdentifier") images = record.get("images") require(isinstance(images, dict) and set(images) == {"api", "worker"}, "images must contain exactly api and worker.") for name in ("api", "worker"): image = images[name] require(isinstance(image, dict) and set(image) == {"reference", "id"}, f"images.{name} must contain exactly reference and id.") require(image["reference"] == f"guestops-{name}:{expected_commit}", f"images.{name}.reference must use the full approved commit.") require(re.fullmatch(r"sha256:[0-9a-f]{64}", str(image["id"])) is not None, f"images.{name}.id must be an immutable image ID.") operator = safe_text(record.get("operator"), "operator") reviewer = safe_text(record.get("reviewedBy"), "reviewedBy") require(operator.casefold() != reviewer.casefold(), "operator and reviewedBy must be different people.") started = timestamp(record.get("startedAt"), "startedAt") ended = timestamp(record.get("endedAt"), "endedAt") reviewed = timestamp(record.get("reviewedAt"), "reviewedAt") require(started <= ended <= reviewed, "Acceptance timestamps are out of order.") scenarios = record.get("scenarios") require(isinstance(scenarios, list), "scenarios must be a list.") ids = [item.get("id") for item in scenarios if isinstance(item, dict)] required = SYSTEMS[system]["scenarios"] require(len(ids) == len(scenarios) and len(ids) == len(set(ids)) and set(ids) == required, f"{system} requires its exact acceptance scenario set.") for item in scenarios: scenario_id = item["id"] require(item.get("status") == "pass", f"Scenario {scenario_id} has not passed.") evidence = item.get("evidence") require(isinstance(evidence, list) and 1 <= len(evidence) <= 10 and all( isinstance(value, str) and 3 <= len(value) <= 200 and "@" not in value and "\\" not in value and not value.startswith("/") for value in evidence ), f"Scenario {scenario_id} requires safe opaque evidence references.") require(record.get("unresolvedCriticalFindings") == 0, "Acceptance cannot pass with unresolved critical findings.") return host def validate_host(record: object, expected_commit: str, expected_release_sha256: str) -> str: host = validate_common(record, expected_commit, expected_release_sha256) require(record.get("system") == "guestops-debian-host", "First record must be guestops-debian-host.") facts = record.get("hostFacts") require(isinstance(facts, dict) and set(facts) == { "debianMajor", "cpuCores", "memoryBytes", "freeDiskBytes", "publicTcpPorts", }, "hostFacts must contain the exact reviewed host facts.") require(isinstance(facts["debianMajor"], int) and facts["debianMajor"] >= 12, "Debian 12 or newer is required.") require(isinstance(facts["cpuCores"], int) and facts["cpuCores"] >= 4, "At least four CPU cores are required.") require(isinstance(facts["memoryBytes"], int) and facts["memoryBytes"] >= 7_500_000_000, "At least 7.5 GB of memory is required.") require(isinstance(facts["freeDiskBytes"], int) and facts["freeDiskBytes"] >= 8 * 1024**3, "At least 8 GiB of free disk space is required.") require(facts["publicTcpPorts"] == [80, 443], "Only TCP ports 80 and 443 may be public.") require(record.get("featureControls") == { "googleSending": "disabled", "faqLiveMode": "disabled", "pmsWrites": "disabled", "paymentCreation": "disabled", }, "Unaccepted external writes and FAQ live mode must remain disabled.") return host def validate_persistence(record: object, expected_commit: str, expected_release_sha256: str) -> str: host = validate_common(record, expected_commit, expected_release_sha256) require(record.get("system") == "guestops-persistence", "Second record must be guestops-persistence.") require(record.get("drillCommand") == "python3 deploy/ops.py persistence-drill --confirm-restart", "drillCommand must identify the confirmation-gated persistence drill.") return host def validate_pair(host_record: object, persistence_record: object, expected_commit: str, expected_release_sha256: str) -> None: host = validate_host(host_record, expected_commit, expected_release_sha256) persistence_host = validate_persistence( persistence_record, expected_commit, expected_release_sha256) require(host == persistence_host, "Both records must identify the same host.") for field in ("environment", "releaseVersion", "releaseCommit", "releaseRecordSha256", "archiveSha256", "images"): require(host_record.get(field) == persistence_record.get(field), f"Both records must use the same {field}.") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("host_record", type=Path) parser.add_argument("persistence_record", type=Path) parser.add_argument("--expected-commit", required=True) parser.add_argument("--expected-release-record-sha256", required=True) args = parser.parse_args() host_record = json.loads(args.host_record.read_text(encoding="utf-8")) persistence_record = json.loads(args.persistence_record.read_text(encoding="utf-8")) validate_pair(host_record, persistence_record, args.expected_commit, args.expected_release_record_sha256) print("Debian-host and persistence acceptance records are structurally complete and passed. " "This validates the records, not their restricted evidence.") if __name__ == "__main__": try: main() except (OSError, ValueError, json.JSONDecodeError) as error: print(f"Debian acceptance records rejected: {error}", file=__import__("sys").stderr) raise SystemExit(1)