#!/usr/bin/env python3 """Validate a restricted GuestOps backup, monitoring and recovery record.""" 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" SCENARIOS = { "encrypted-manual-backup", "scheduled-backup", "production-service-recovery", "atomic-off-host-transfer", "off-host-checksum", "retention-and-legal-hold", "monitoring-coverage", "alert-escalation", "durable-secret-free-logs", "isolated-restore", "data-protection-recovery", "database-inventory", "image-rollback", "controlled-return-to-service", } 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_name(value: object, field: str) -> str: name = str(value or "").strip() require(2 <= len(name) <= 120 and "@" not in name and "/" not in name and "\\" not in name, f"{field} requires a name without an email address or path.") return name def validate(record: object, expected_commit: str, expected_release_sha256: str) -> None: require(isinstance(record, dict), "Acceptance record must be a JSON object.") require(record.get("schemaVersion") == 1, "Unsupported backup/recovery schema.") require(record.get("system") == "guestops-backup-recovery", "system must be guestops-backup-recovery.") require(record.get("evidenceId") == "backup-restore", "evidenceId must be backup-restore.") require(record.get("dataClassification") == "synthetic-only", "Recovery acceptance must use synthetic data only.") 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.") safe_name(record.get("hostIdentifier"), "hostIdentifier") images = record.get("images") require(isinstance(images, dict) and set(images) == {"api", "worker", "mongo"}, "images must contain exactly api, worker and mongo.") 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 immutable.") mongo = images["mongo"] require(isinstance(mongo, dict) and set(mongo) == {"reference", "id"} and mongo["reference"] == "mongo:8.0" and re.fullmatch(r"sha256:[0-9a-f]{64}", str(mongo["id"])) is not None, "images.mongo must identify the immutable mongo:8.0 image.") owners = record.get("owners") owner_keys = {"backup", "monitoring", "recoveryOperator", "technicalEscalation", "retention", "independentReviewer"} require(isinstance(owners, dict) and set(owners) == owner_keys, "owners must contain the exact operational and review roles.") names = {key: safe_name(value, f"owners.{key}") for key, value in owners.items()} reviewer = names["independentReviewer"].casefold() require(reviewer not in {names[key].casefold() for key in owner_keys - {"independentReviewer"}}, "independentReviewer must be different from every operational owner.") 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.") recovery = record.get("recoveryObjectives") require(isinstance(recovery, dict) and set(recovery) == { "targetRpoHours", "observedRpoHours", "targetRtoMinutes", "observedRtoMinutes", }, "recoveryObjectives must contain exact target and observed RPO/RTO values.") for field in recovery: require(isinstance(recovery[field], (int, float)) and not isinstance(recovery[field], bool) and recovery[field] >= 0, f"recoveryObjectives.{field} must be non-negative.") require(recovery["targetRpoHours"] == 24 and recovery["observedRpoHours"] <= 24, "Observed RPO must meet the approved 24-hour target.") require(recovery["targetRtoMinutes"] == 240 and recovery["observedRtoMinutes"] <= 240, "Observed RTO must meet the approved four-hour target.") retention = record.get("retention") require(retention == { "localVerifiedDays": 7, "offHostDaily": 35, "offHostMonthly": 12, "legalHoldOverrideTested": True, }, "Retention must record seven local days, 35 daily and 12 monthly off-host copies, and legal-hold testing.") backup = record.get("backup") require(isinstance(backup, dict) and set(backup) == { "createdAt", "sha256", "transferredSha256", "privateKeyPresentOnHost", }, "backup must contain exact creation, checksum, transfer and private-key fields.") created = timestamp(backup["createdAt"], "backup.createdAt") require(created <= started, "The accepted backup must exist when the timed exercise starts.") require(abs(recovery["observedRpoHours"] - (started - created).total_seconds() / 3600) < 0.01, "Observed RPO must match the backup and exercise timestamps.") require(abs(recovery["observedRtoMinutes"] - (ended - started).total_seconds() / 60) < 0.01, "Observed RTO must match the exercise timestamps.") require(SHA256.fullmatch(str(backup["sha256"])) is not None and backup["transferredSha256"] == backup["sha256"], "Local and transferred backup checksums must match.") require(backup["privateKeyPresentOnHost"] is False, "The recovery private key must not be present on the Debian host.") rollback = record.get("rollback") require(isinstance(rollback, dict) and set(rollback) == { "previousReleaseCommit", "previousArchiveSha256", "previousImages", "persistentVolumesReplaced", "restoredReleaseCommit", "unresolvedOperations", "finalControls", }, "rollback must contain the exact rehearsal and final-state fields.") require(GIT_SHA.fullmatch(str(rollback["previousReleaseCommit"])) is not None and rollback["previousReleaseCommit"] != expected_commit, "Rollback must use a different retained previous release.") require(SHA256.fullmatch(str(rollback["previousArchiveSha256"])) is not None, "Rollback requires the previous archive checksum.") previous_images = rollback["previousImages"] require(isinstance(previous_images, dict) and set(previous_images) == {"api", "worker"}, "Rollback requires exact previous API and worker images.") for name in ("api", "worker"): image = previous_images[name] require(isinstance(image, dict) and set(image) == {"reference", "id"} and image["reference"] == f"guestops-{name}:{rollback['previousReleaseCommit']}" and re.fullmatch(r"sha256:[0-9a-f]{64}", str(image["id"])) is not None, f"rollback.previousImages.{name} must use the retained previous release identity.") require(rollback["persistentVolumesReplaced"] is False, "Image rollback must not replace persistent volumes.") require(rollback["restoredReleaseCommit"] == expected_commit, "The exercise must finish on the approved candidate.") require(rollback["unresolvedOperations"] == 0, "The exercise must finish without unresolved operations.") require(rollback["finalControls"] == { "googleSending": "disabled", "faqLiveMode": "disabled", "pmsWrites": "disabled", "paymentCreation": "disabled", }, "The exercise must finish with all unaccepted external writes disabled.") 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)] require(len(ids) == len(scenarios) and len(ids) == len(set(ids)) and set(ids) == SCENARIOS, "Acceptance record requires the exact backup/recovery 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("monitoringState") == "healthy", "Monitoring must be healthy at acceptance completion.") require(record.get("unresolvedCriticalFindings") == 0, "Acceptance cannot pass with unresolved critical findings.") def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("record", type=Path) parser.add_argument("--expected-commit", required=True) parser.add_argument("--expected-release-record-sha256", required=True) args = parser.parse_args() validate(json.loads(args.record.read_text(encoding="utf-8")), args.expected_commit, args.expected_release_record_sha256) print("Backup, monitoring and recovery acceptance record is structurally complete and passed. " "This validates the record, not its restricted evidence.") if __name__ == "__main__": try: main() except (OSError, ValueError, json.JSONDecodeError) as error: print(f"Backup/recovery acceptance record rejected: {error}", file=__import__("sys").stderr) raise SystemExit(1)