207 lines
9.5 KiB
Python
207 lines
9.5 KiB
Python
#!/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.1"
|
|
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)
|