"""Run Eggcracker's local full qualification and produce a portable evidence pack.""" from __future__ import annotations import argparse import hashlib import json import os import platform import pwd import subprocess import sys import time from datetime import UTC, datetime from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] CLI = Path("/usr/local/bin/eggcracker") POLICY = Path("/etc/lumi-eggcracker/policy.json") def run(argv: list[str], *, timeout: int = 900) -> None: result = subprocess.run(argv, capture_output=True, text=True, check=False, timeout=timeout) if result.returncode: raise RuntimeError( result.stderr.strip() or result.stdout.strip() or "qualification command failed" ) def digest(path: Path) -> str: value = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(64 * 1024), b""): value.update(block) return value.hexdigest() def new_directory(path: Path) -> Path: if not path.is_absolute() or path.exists() or path.is_symlink() or not path.parent.is_dir(): raise RuntimeError( "evidence directory must be a new absolute child of an existing directory" ) path.mkdir(mode=0o700) return path def doctor(operator: str, *, timeout: float = 30) -> dict[str, Any]: """Wait through a bounded post-restart scan and require a real PASS.""" deadline = time.monotonic() + timeout latest = "Eggcracker doctor did not pass" while time.monotonic() < deadline: result = subprocess.run( ["/usr/sbin/runuser", "-u", operator, "--", str(CLI), "doctor"], capture_output=True, text=True, check=False, timeout=3, ) latest = result.stderr.strip() or result.stdout.strip() or latest if result.returncode == 0: try: value = json.loads(result.stdout) except json.JSONDecodeError: value = None if isinstance(value, dict) and value.get("result") == "PASS": return value time.sleep(0.05) raise RuntimeError(latest) def main() -> int: parser = argparse.ArgumentParser( description="Run Eggcracker's complete local qualification against an installed instance." ) parser.add_argument("--operator", required=True) parser.add_argument("--assets-manifest", required=True, type=Path) parser.add_argument("--safetensors-assets-manifest", required=True, type=Path) parser.add_argument("--evidence-dir", required=True, type=Path) args = parser.parse_args() if not hasattr(os, "geteuid") or os.geteuid() != 0: raise SystemExit("self-validation must run as root") evidence: Path | None = None summary: dict[str, Any] = { "result": "FAIL", "schema_version": "lumi-eggcracker.self-validation.v1", } try: evidence = new_directory(args.evidence_dir) if not CLI.is_file() or not POLICY.is_file() or POLICY.is_symlink(): raise RuntimeError("Eggcracker is not installed") policy = json.loads(POLICY.read_text(encoding="utf-8")) if policy.get("operator_uid") is None or policy.get("source_commit") is None: raise RuntimeError("installed policy is invalid") workload = pwd.getpwuid(int(policy["workload_uid"])).pw_name if workload == args.operator: raise RuntimeError("installed workload identity must differ from operator") checked = doctor(args.operator) environment = { "catalogue": checked["catalogue"], "platform": platform.platform(), "python": sys.version.split()[0], "source_commit": policy["source_commit"], "validated_utc": datetime.now(UTC).isoformat().replace("+00:00", "Z"), "version": checked["version"], } (evidence / "environment.json").write_text( json.dumps(environment, sort_keys=True) + "\n", encoding="utf-8" ) (evidence / "doctor.json").write_text( json.dumps(checked, sort_keys=True) + "\n", encoding="utf-8" ) jobs = ( ( "real-ai-smoke.json", [ "smoke_content_ai.py", "--assets-manifest", str(args.assets_manifest), "--user", workload, "--repetitions", "5", ], ), ( "safetensors-ai-smoke.json", [ "smoke_safetensors_ai.py", "--assets-manifest", str(args.safetensors_assets_manifest), "--user", workload, "--repetitions", "5", ], ), ( "safetensors-aten-negative.json", [ "smoke_safetensors_aten_negative.py", "--assets-manifest", str(args.safetensors_assets_manifest), "--user", workload, "--repetitions", "5", ], ), ( "content-matrix.json", [ "run_content_matrix.py", "--assets-manifest", str(args.assets_manifest), "--user", workload, "--repetitions", "100", ], ), ( "benign-model-matrix.json", [ "run_content_benign_matrix.py", "--assets-manifest", str(args.assets_manifest), "--user", workload, "--repetitions", "300", ], ), ( "content-adversarial-matrix.json", [ "run_content_adversarial_matrix.py", "--assets-manifest", str(args.assets_manifest), "--user", workload, "--tree-repetitions", "100", "--startup-repetitions", "20", "--restart-repetitions", "20", ], ), ( "autonomous-regression.json", [ "run_autonomous_matrix.py", "--assets-manifest", str(args.assets_manifest), "--discoveries", "100", "--approved", "50", "--benign", "200", ], ), ( "selected-workload-regression.json", [ "run_native_matrix.py", "--fork-race-repetitions", "100", "--benign-repetitions", "50", "--restart-repetitions", "20", "--socket-attempts", "100", ], ), ( "overhead-benchmark.json", [ "benchmark_overhead.py", "--policy", str(POLICY), ], ), ) for filename, arguments in jobs: doctor(args.operator) run( [ sys.executable, str(ROOT / "scripts" / arguments[0]), *arguments[1:], "--output", str(evidence / filename), ] ) summary.update( { "result": "PASS", "source_commit": policy["source_commit"], "version": checked["version"], } ) (evidence / "validation.json").write_text( json.dumps(summary, sort_keys=True) + "\n", encoding="utf-8" ) report = "# Lumi Eggcracker local validation\n\nPASS: the packaged deterministic, recognition, containment, boundary, recovery and overhead qualification jobs completed on the installed instance.\n" (evidence / "report.md").write_text(report, encoding="utf-8") sums = "".join( f"{digest(path)} {path.name}\n" for path in sorted(evidence.iterdir()) if path.name != "SHA256SUMS" ) (evidence / "SHA256SUMS").write_text(sums, encoding="utf-8") print(json.dumps({"evidence_dir": str(evidence), **summary}, sort_keys=True)) return 0 except ( OSError, RuntimeError, TypeError, ValueError, json.JSONDecodeError, subprocess.SubprocessError, ) as error: summary["error"] = str(error) if evidence is not None: (evidence / "validation.json").write_text( json.dumps(summary, sort_keys=True) + "\n", encoding="utf-8" ) raise SystemExit(f"self-validation failed: {error}") from error if __name__ == "__main__": raise SystemExit(main())