119 lines
5.2 KiB
Python
119 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail-closed validator for the Offline Workstation Gate 2 payload lock."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
IMMUTABLE_IMAGE = re.compile(r"^[^\s]+@sha256:([0-9a-f]{64})$")
|
|
FORBIDDEN_SOURCE = re.compile(r"(^|[/_.:-])(latest|main|master)([/_.:@-]|$)", re.I)
|
|
REQUIRED_OBJECT_FIELDS = {"id", "version", "architecture", "source", "sha256", "bytes", "license"}
|
|
OBJECT_GROUPS = ("apt", "python", "runtime", "containers", "models", "source")
|
|
|
|
|
|
def fail(errors: list[str], message: str) -> None:
|
|
errors.append(message)
|
|
|
|
|
|
def validate(path: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError:
|
|
return [f"payload lock does not exist: {path}"]
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
return [f"payload lock is unreadable or invalid JSON: {exc}"]
|
|
|
|
if data.get("schema") != "nexus-one-ai.offline-workstation-payload-lock.v1":
|
|
fail(errors, "schema must be nexus-one-ai.offline-workstation-payload-lock.v1")
|
|
if data.get("product") != "offline-workstation" or data.get("architecture") != "amd64":
|
|
fail(errors, "product/architecture must be offline-workstation/amd64")
|
|
if data.get("ubuntu_release") != "22.04.5":
|
|
fail(errors, "ubuntu_release must be 22.04.5")
|
|
if data.get("approval_record") != "CT-G1-2026-009":
|
|
fail(errors, "approval_record must bind the approved payload boundary CT-G1-2026-009")
|
|
if data.get("status") != "reviewed":
|
|
fail(errors, "status must be reviewed before acquisition or assembly")
|
|
if data.get("unresolved"):
|
|
fail(errors, "unresolved must be an empty array")
|
|
|
|
objects = data.get("objects")
|
|
if not isinstance(objects, dict):
|
|
return errors + ["objects must be an object"]
|
|
|
|
seen_ids: set[str] = set()
|
|
seen_paths: set[str] = set()
|
|
for group in OBJECT_GROUPS:
|
|
entries = objects.get(group)
|
|
if not isinstance(entries, list) or not entries:
|
|
fail(errors, f"objects.{group} must be a non-empty array")
|
|
continue
|
|
for index, entry in enumerate(entries):
|
|
label = f"objects.{group}[{index}]"
|
|
if not isinstance(entry, dict):
|
|
fail(errors, f"{label} must be an object")
|
|
continue
|
|
missing = sorted(REQUIRED_OBJECT_FIELDS - entry.keys())
|
|
if missing:
|
|
fail(errors, f"{label} missing fields: {', '.join(missing)}")
|
|
continue
|
|
object_id = entry["id"]
|
|
if not isinstance(object_id, str) or not object_id:
|
|
fail(errors, f"{label}.id must be non-empty")
|
|
elif object_id in seen_ids:
|
|
fail(errors, f"duplicate object id: {object_id}")
|
|
else:
|
|
seen_ids.add(object_id)
|
|
if entry.get("architecture") not in {"amd64", "all"}:
|
|
fail(errors, f"{label}.architecture must be amd64 or all")
|
|
if not isinstance(entry.get("bytes"), int) or entry["bytes"] <= 0:
|
|
fail(errors, f"{label}.bytes must be a positive integer")
|
|
if data.get("status") == "reviewed" and entry.get("bytes_basis") != "measured":
|
|
fail(errors, f"{label}.bytes_basis must be measured for a reviewed lock")
|
|
if not isinstance(entry.get("sha256"), str) or not SHA256.fullmatch(entry["sha256"]):
|
|
fail(errors, f"{label}.sha256 must be 64 lowercase hexadecimal characters")
|
|
for field in ("version", "source", "license"):
|
|
value = entry.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
fail(errors, f"{label}.{field} must be non-empty")
|
|
source = entry.get("source", "")
|
|
version = entry.get("version", "")
|
|
if FORBIDDEN_SOURCE.search(source) or FORBIDDEN_SOURCE.search(version):
|
|
fail(errors, f"{label} contains mutable latest/main/master identity")
|
|
target = entry.get("target")
|
|
if not isinstance(target, str) or not target or target.startswith("/") or ".." in Path(target).parts:
|
|
fail(errors, f"{label}.target must be a safe relative path")
|
|
elif target in seen_paths:
|
|
fail(errors, f"duplicate target path: {target}")
|
|
else:
|
|
seen_paths.add(target)
|
|
if group == "containers":
|
|
match = IMMUTABLE_IMAGE.fullmatch(source)
|
|
if not match:
|
|
fail(errors, f"{label}.source must be an OCI reference pinned with @sha256")
|
|
elif match.group(1) != entry["sha256"]:
|
|
fail(errors, f"{label} OCI digest must match sha256")
|
|
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) != 2:
|
|
print(f"usage: {Path(sys.argv[0]).name} PAYLOAD_LOCK.json", file=sys.stderr)
|
|
return 2
|
|
errors = validate(Path(sys.argv[1]))
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}", file=sys.stderr)
|
|
return 1
|
|
print("Offline Workstation payload lock preflight passed.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|