221 lines
11 KiB
Python
221 lines
11 KiB
Python
#!/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)
|