"""Run the release-blocking native Priority-0 adversarial campaigns. The root controller is only an oracle and fixture coordinator. Every target process runs as the dedicated workload identity and wins only while it retains a complete supported content/runtime profile. """ from __future__ import annotations import argparse import ctypes import fcntl import hashlib import json import mmap import os import pwd import resource import secrets import shutil import signal import subprocess import sys import time from pathlib import Path from typing import Any CLI = Path("/usr/local/bin/eggcracker") DETECTIONS = Path("/var/lib/lumi-eggcracker/detections") POLICY = Path("/etc/lumi-eggcracker/policy.json") INSTALLED_ARTIFACT = Path("/usr/local/lib/lumi-eggcracker/lumi-eggcracker.pyz") SCHEMA = "lumi-eggcracker.p0-native.v1" PROFILE = "content.gguf-llama" HOLD_SECONDS = 180 def digest(path: Path) -> str: value = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): value.update(block) return value.hexdigest() def gguf(seed: int) -> bytes: # Plausible GGUF v3: one tensor, no metadata, plus the minimum tensor # prefix required by the bounded public validator. return b"GGUF" + (3).to_bytes(4, "little") + (1).to_bytes( 8, "little" ) + (0).to_bytes(8, "little") + bytes([seed & 0xFF]) * 12 def write_all(descriptor: int, value: bytes) -> None: pending = memoryview(value) while pending: written = os.write(descriptor, pending) if written < 1: raise OSError("fixture write made no progress") pending = pending[written:] os.lseek(descriptor, 0, os.SEEK_SET) def copy_to_descriptor(source: Path, descriptor: int) -> None: with source.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): write_all_at_current_offset(descriptor, block) os.lseek(descriptor, 0, os.SEEK_SET) def write_all_at_current_offset(descriptor: int, value: bytes) -> None: pending = memoryview(value) while pending: written = os.write(descriptor, pending) if written < 1: raise OSError("fixture write made no progress") pending = pending[written:] def wait_gate(path: Path, timeout: float = 90) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if path.is_file(): return time.sleep(0.005) raise RuntimeError("fixture gate did not open") def marker(path: Path, value: dict[str, Any]) -> None: path.write_text(json.dumps(value, sort_keys=True) + "\n", encoding="utf-8") def map_runtime(descriptor: int) -> mmap.mmap: return mmap.mmap( descriptor, 0, flags=mmap.MAP_PRIVATE, prot=mmap.PROT_READ | mmap.PROT_EXEC, ) def fixture_evidence(argv: list[str]) -> int: parser = argparse.ArgumentParser() parser.add_argument("--mode", required=True) parser.add_argument("--model", required=True, type=Path) parser.add_argument("--runtime", required=True, type=Path) parser.add_argument("--gate", required=True, type=Path) parser.add_argument("--ready", required=True, type=Path) parser.add_argument("--work", required=True, type=Path) parser.add_argument("--pre-fds", type=int, default=0) parser.add_argument("--post-fds", type=int, default=0) parser.add_argument("--decoy-maps", type=int, default=0) args = parser.parse_args(argv) # CPython's mmap object retains its own duplicate of the backing # descriptor, so each decoy consumes two slots until teardown. required_descriptors = ( args.pre_fds + args.post_fds + args.decoy_maps * 2 + 64 ) soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE) if required_descriptors > hard_limit: raise RuntimeError("fixture descriptor requirement exceeds the hard limit") if soft_limit < required_descriptors: resource.setrlimit( resource.RLIMIT_NOFILE, (required_descriptors, hard_limit), ) descriptors: list[int] = [] mappings: list[mmap.mmap] = [] for _ in range(args.pre_fds): descriptors.append(os.open("/dev/null", os.O_RDONLY)) marker(args.ready, {"pid": os.getpid(), "state": "READY"}) wait_gate(args.gate) if args.mode in {"regular", "high-fd", "high-maps"}: model_descriptor = os.open(args.model, os.O_RDONLY) elif args.mode in {"memfd-model", "sealed-memfd-model"}: flags = getattr(os, "MFD_ALLOW_SEALING", 2) model_descriptor = os.memfd_create("p0-model", flags=flags) write_all(model_descriptor, args.model.read_bytes()) if args.mode == "sealed-memfd-model": seals = ( getattr(fcntl, "F_SEAL_SEAL", 0x0001) | getattr(fcntl, "F_SEAL_SHRINK", 0x0002) | getattr(fcntl, "F_SEAL_GROW", 0x0004) | getattr(fcntl, "F_SEAL_WRITE", 0x0008) ) fcntl.fcntl(model_descriptor, getattr(fcntl, "F_ADD_SEALS", 1033), seals) elif args.mode == "otmpfile-model": model_descriptor = os.open( args.work, os.O_TMPFILE | os.O_RDWR, 0o600, ) write_all(model_descriptor, args.model.read_bytes()) elif args.mode == "deleted-model": private = args.work / f"deleted-model-{os.getpid()}" private.write_bytes(args.model.read_bytes()) model_descriptor = os.open(private, os.O_RDONLY) private.unlink() elif args.mode == "unlink-model": model_descriptor = os.open(args.model, os.O_RDONLY) args.model.unlink() else: model_descriptor = os.open(args.model, os.O_RDONLY) descriptors.append(model_descriptor) for _ in range(args.post_fds): descriptors.append(os.open("/dev/null", os.O_RDONLY)) for index in range(args.decoy_maps): decoy = args.work / f"map-{os.getpid()}-{index}" descriptor = os.open(decoy, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600) os.ftruncate(descriptor, 4096) mappings.append( mmap.mmap( descriptor, 4096, flags=mmap.MAP_PRIVATE, prot=mmap.PROT_READ | mmap.PROT_EXEC, ) ) descriptors.append(descriptor) if args.mode in {"memfd-runtime", "otmpfile-runtime", "deleted-runtime", "unlink-runtime"}: if args.mode == "memfd-runtime": runtime_descriptor = os.memfd_create("p0-runtime", flags=0) elif args.mode == "otmpfile-runtime": runtime_descriptor = os.open( args.work, os.O_TMPFILE | os.O_RDWR, 0o700, ) elif args.mode == "unlink-runtime": runtime_descriptor = os.open(args.runtime, os.O_RDONLY) else: private_runtime = args.work / f"deleted-runtime-{os.getpid()}" shutil.copyfile(args.runtime, private_runtime) runtime_descriptor = os.open(private_runtime, os.O_RDONLY) private_runtime.unlink() if args.mode in {"memfd-runtime", "otmpfile-runtime"}: copy_to_descriptor(args.runtime, runtime_descriptor) mappings.append(map_runtime(runtime_descriptor)) descriptors.append(runtime_descriptor) if args.mode == "unlink-runtime": args.runtime.unlink() else: runtime_descriptor = os.open(args.runtime, os.O_RDONLY) mappings.append(map_runtime(runtime_descriptor)) descriptors.append(runtime_descriptor) marker( args.ready.with_suffix(".armed"), { "descriptors": len(descriptors), "maps": len(mappings), "pid": os.getpid(), "state": "ARMED", }, ) time.sleep(HOLD_SECONDS) return 0 def fixture_storm(argv: list[str]) -> int: parser = argparse.ArgumentParser() parser.add_argument("--count", required=True, type=int) parser.add_argument("--role", choices=("complete", "content"), required=True) parser.add_argument("--model", required=True, type=Path) parser.add_argument("--runtime", required=True, type=Path) parser.add_argument("--gate", required=True, type=Path) parser.add_argument("--ready", required=True, type=Path) args = parser.parse_args(argv) children: list[int] = [] for _ in range(args.count): pid = os.fork() if pid: children.append(pid) continue wait_gate(args.gate) descriptors = [os.open(args.model, os.O_RDONLY)] mappings: list[mmap.mmap] = [] if args.role == "complete": runtime_descriptor = os.open(args.runtime, os.O_RDONLY) descriptors.append(runtime_descriptor) mappings.append(map_runtime(runtime_descriptor)) time.sleep(HOLD_SECONDS) os._exit(0) marker( args.ready, {"children": children, "count": len(children), "pid": os.getpid(), "state": "READY"}, ) time.sleep(HOLD_SECONDS) return 0 def execveat(descriptor: int, command: list[str], environment: dict[str, str]) -> None: libc = ctypes.CDLL(None, use_errno=True) argv = (ctypes.c_char_p * (len(command) + 1))( *(item.encode() for item in command), None, ) env_values = [f"{key}={value}".encode() for key, value in environment.items()] envp = (ctypes.c_char_p * (len(env_values) + 1))(*env_values, None) result = libc.syscall( 322, # x86-64 SYS_execveat descriptor, ctypes.c_char_p(b""), argv, envp, 0x1000, # AT_EMPTY_PATH ) if result != 0: error = ctypes.get_errno() raise OSError(error, os.strerror(error)) raise RuntimeError("execveat unexpectedly returned") def runner_command(executable: str, model: str) -> list[str]: return [ executable, "-m", model, "-p", "Name a Linux cgroup property.", "-n", "4096", "-t", "4", "-tb", "4", "-c", "512", "--simple-io", "--single-turn", "--no-warmup", "--no-display-prompt", "--ignore-eos", "--seed", "1234", ] def fixture_exec(argv: list[str]) -> int: parser = argparse.ArgumentParser() parser.add_argument("--mode", required=True) parser.add_argument("--model", required=True, type=Path) parser.add_argument("--runtime", required=True, type=Path) parser.add_argument("--gate", required=True, type=Path) parser.add_argument("--ready", required=True, type=Path) parser.add_argument("--work", required=True, type=Path) args = parser.parse_args(argv) environment = { "HOME": "/nonexistent", "LANG": "C.UTF-8", "PATH": "/usr/bin:/bin", } if args.mode in {"procfd-model", "sealed-procfd-model"}: flags = getattr(os, "MFD_ALLOW_SEALING", 2) model_descriptor = os.memfd_create("p0-real-model", flags=flags) copy_to_descriptor(args.model, model_descriptor) if args.mode == "sealed-procfd-model": seals = ( getattr(fcntl, "F_SEAL_SEAL", 0x0001) | getattr(fcntl, "F_SEAL_SHRINK", 0x0002) | getattr(fcntl, "F_SEAL_GROW", 0x0004) | getattr(fcntl, "F_SEAL_WRITE", 0x0008) ) fcntl.fcntl(model_descriptor, getattr(fcntl, "F_ADD_SEALS", 1033), seals) os.set_inheritable(model_descriptor, True) marker(args.ready, {"pid": os.getpid(), "state": "READY"}) wait_gate(args.gate) model = f"/proc/self/fd/{model_descriptor}" os.execve(args.runtime, runner_command(str(args.runtime), model), environment) copied = False if args.mode in {"memfd-exec", "execveat-memfd"}: runtime_descriptor = os.memfd_create("p0-executable", flags=0) elif args.mode == "otmpfile-exec": writer = os.open(args.work, os.O_TMPFILE | os.O_RDWR, 0o700) try: copy_to_descriptor(args.runtime, writer) os.fchmod(writer, 0o700) # A regular O_TMPFILE held open for writing cannot be executed # (ETXTBSY). Reopen the same anonymous inode read-only through # procfs before dropping the writer, preserving its pathless # identity while producing a valid execution fixture. runtime_descriptor = os.open(f"/proc/self/fd/{writer}", os.O_RDONLY) copied = True finally: os.close(writer) elif args.mode == "deleted-exec": private = args.work / f"deleted-exec-{os.getpid()}" shutil.copyfile(args.runtime, private) os.chmod(private, 0o700) runtime_descriptor = os.open(private, os.O_RDONLY) private.unlink() else: raise RuntimeError("unknown pathless execution mode") if args.mode != "deleted-exec" and not copied: copy_to_descriptor(args.runtime, runtime_descriptor) os.fchmod(runtime_descriptor, 0o700) marker(args.ready, {"pid": os.getpid(), "state": "READY"}) wait_gate(args.gate) command = runner_command("p0-pathless-runtime", str(args.model)) if args.mode == "execveat-memfd": execveat(runtime_descriptor, command, environment) os.execve(runtime_descriptor, command, environment) raise RuntimeError("fexecve unexpectedly returned") def control(argv: list[str], *, operator: str | None = None, environment: dict[str, str] | None = None) -> tuple[int, str, str]: command = [str(CLI), *argv] if operator is not None: command = ["/usr/sbin/runuser", "-u", operator, "--", *command] result = subprocess.run( command, capture_output=True, text=True, check=False, timeout=60, env=environment, ) return result.returncode, result.stdout, result.stderr def json_control(argv: list[str], *, operator: str | None = None) -> dict[str, Any]: code, stdout, stderr = control(argv, operator=operator) if code: raise RuntimeError(stderr.strip() or stdout.strip() or "Eggcracker control failed") value = json.loads(stdout) if not isinstance(value, dict): raise TypeError("Eggcracker control response is invalid") return value def stop(process: subprocess.Popen[bytes] | None) -> None: if process is None or process.poll() is not None: return try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) def alive(pid: int) -> bool: try: fields = Path(f"/proc/{pid}/stat").read_text(encoding="ascii").split() except OSError: return False return len(fields) > 2 and fields[2] != "Z" def wait_ready(path: Path, timeout: float = 90) -> dict[str, Any]: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): time.sleep(0.01) continue if isinstance(value, dict): return value raise RuntimeError(f"fixture readiness failed for {path.name}") def receipts_after(before: set[Path]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for path in sorted(set(DETECTIONS.glob("*.json")) - before): result.append(json.loads(path.read_text(encoding="utf-8"))) return result def validate_receipts( values: list[dict[str, Any]], *, expected: int, source_commit: str, ) -> list[float]: if len(values) != expected: raise RuntimeError(f"expected {expected} receipts, observed {len(values)}") latencies: list[float] = [] event_ids: set[str] = set() for value in values: containment = value.get("containment", {}) if ( value.get("result") != "TERMINATED" or value.get("source_commit") != source_commit or value.get("detector", {}).get("profile") != PROFILE or containment.get("root_populated") != 0 or containment.get("surviving_pids") != [] or "cgroup.kill" not in str(containment.get("primitive")) ): raise RuntimeError("receipt did not prove exact supported-profile containment") event_id = value.get("event_id") if not isinstance(event_id, str) or event_id in event_ids: raise RuntimeError("receipt event identity is invalid or duplicated") event_ids.add(event_id) latencies.append(float(containment["trigger_to_empty_ms"])) return latencies def launch_fixture( script: Path, workload: str, subcommand: str, arguments: list[str], ) -> subprocess.Popen[bytes]: return subprocess.Popen( [ "/usr/sbin/runuser", "-u", workload, "--", "/usr/bin/python3", "-I", "-S", str(script), subcommand, *arguments, ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) def wait_for_kills( target_pids: list[int], before: set[Path], expected: int, timeout: float, ) -> list[dict[str, Any]]: deadline = time.monotonic() + timeout while time.monotonic() < deadline: values = receipts_after(before) if len(values) >= expected and not any(alive(pid) for pid in target_pids): return values time.sleep(0.05) raise RuntimeError( f"containment timed out: receipts={len(receipts_after(before))}, " f"survivors={sum(alive(pid) for pid in target_pids)}" ) class Campaign: def __init__( self, *, script: Path, workload: str, operator: str, runtime: Path, real_model: Path, output: Path, ) -> None: self.workload = workload self.operator = operator self.runtime = runtime self.real_model = real_model self.output = output self.policy = json.loads(POLICY.read_text(encoding="utf-8")) token = secrets.token_hex(8) self.root = Path(f"/opt/lumi-eggcracker-p0-{token}") self.work = self.root / "work" self.root.mkdir(mode=0o755) self.work.mkdir(mode=0o733) os.chmod(self.work, 0o733) self.script = self.root / "fixture.py" shutil.copyfile(script, self.script) self.script.chmod(0o555) self.model = self.root / "synthetic.gguf" self.model.write_bytes(gguf(1)) self.model.chmod(0o444) self.canary = subprocess.Popen( ["/usr/sbin/runuser", "-u", workload, "--", "/bin/sleep", "900"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) self.results: dict[str, Any] = { "approval_material_substitution": [], "pathless_deleted": [], "saturation": [], } def assert_canary(self) -> None: if self.canary.poll() is not None: raise RuntimeError("unrelated workload-identity canary was terminated") def doctor(self) -> dict[str, Any]: value = json_control(["doctor"]) if value.get("result") != "PASS": raise RuntimeError("Eggcracker health did not recover to PASS") return value def one_evidence( self, mode: str, *, model: Path | None = None, runtime: Path | None = None, pre_fds: int = 0, post_fds: int = 0, decoy_maps: int = 0, timeout: float = 90, ) -> dict[str, Any]: token = secrets.token_hex(6) gate = self.root / f"gate-{token}" ready = self.work / f"ready-{token}.json" before = set(DETECTIONS.glob("*.json")) process = launch_fixture( self.script, self.workload, "fixture-evidence", [ "--mode", mode, "--model", str(model or self.model), "--runtime", str(runtime or self.runtime), "--gate", str(gate), "--ready", str(ready), "--work", str(self.work), "--pre-fds", str(pre_fds), "--post-fds", str(post_fds), "--decoy-maps", str(decoy_maps), ], ) try: identity = wait_ready(ready) gate.write_bytes(b"go\n") gate.chmod(0o444) receipts = wait_for_kills( [int(identity["pid"])], before, 1, timeout, ) latencies = validate_receipts( receipts, expected=1, source_commit=self.policy["source_commit"], ) self.assert_canary() return { "case": mode, "result": "PASS", "trigger_to_empty_ms": latencies[0], } finally: stop(process) def one_exec(self, mode: str, timeout: float = 150) -> dict[str, Any]: token = secrets.token_hex(6) gate = self.root / f"gate-{token}" ready = self.work / f"ready-{token}.json" before = set(DETECTIONS.glob("*.json")) process = launch_fixture( self.script, self.workload, "fixture-exec", [ "--mode", mode, "--model", str(self.real_model), "--runtime", str(self.runtime), "--gate", str(gate), "--ready", str(ready), "--work", str(self.work), ], ) try: identity = wait_ready(ready, timeout=120) gate.write_bytes(b"go\n") gate.chmod(0o444) receipts = wait_for_kills( [int(identity["pid"])], before, 1, timeout, ) latencies = validate_receipts( receipts, expected=1, source_commit=self.policy["source_commit"], ) self.assert_canary() return { "case": mode, "result": "PASS", "trigger_to_empty_ms": latencies[0], } finally: stop(process) def wave(self, count: int) -> dict[str, Any]: token = secrets.token_hex(6) gate = self.root / f"wave-gate-{token}" before = set(DETECTIONS.glob("*.json")) processes: list[subprocess.Popen[bytes]] = [] ready_paths: list[Path] = [] try: for index in range(count): ready = self.work / f"wave-{token}-{index}.json" ready_paths.append(ready) processes.append( launch_fixture( self.script, self.workload, "fixture-evidence", [ "--mode", "regular", "--model", str(self.model), "--runtime", str(self.runtime), "--gate", str(gate), "--ready", str(ready), "--work", str(self.work), ], ) ) identities = [wait_ready(path) for path in ready_paths] gate.write_bytes(b"go\n") gate.chmod(0o444) receipts = wait_for_kills( [int(item["pid"]) for item in identities], before, count, 120, ) latencies = validate_receipts( receipts, expected=count, source_commit=self.policy["source_commit"], ) self.assert_canary() return { "case": f"unrelated-complete-wave-{count}", "max_trigger_to_empty_ms": max(latencies), "receipts": len(receipts), "result": "PASS", } finally: for process in processes: stop(process) def storm(self, count: int, role: str) -> dict[str, Any]: token = secrets.token_hex(6) gate = self.root / f"storm-gate-{token}" ready = self.work / f"storm-{token}.json" before = set(DETECTIONS.glob("*.json")) process = launch_fixture( self.script, self.workload, "fixture-storm", [ "--count", str(count), "--role", role, "--model", str(self.model), "--runtime", str(self.runtime), "--gate", str(gate), "--ready", str(ready), ], ) try: identity = wait_ready(ready) children = [int(item) for item in identity["children"]] if len(children) != count: raise RuntimeError("storm did not create the requested process count") gate.write_bytes(b"go\n") gate.chmod(0o444) if role == "complete": receipts = wait_for_kills( [int(identity["pid"]), *children], before, 1, 120, ) validate_receipts( receipts, expected=1, source_commit=self.policy["source_commit"], ) if int(receipts[0].get("capture", {}).get("captured_processes", 0)) < count + 1: raise RuntimeError("related component receipt omitted armed processes") else: time.sleep(12) if receipts_after(before) or not all(alive(pid) for pid in children): raise RuntimeError("partial-profile storm was killed or produced a receipt") self.assert_canary() return { "case": f"related-{role}-storm-{count}", "processes": count, "receipts": len(receipts_after(before)), "result": "PASS", } finally: stop(process) def reject_mutation(self, kind: str) -> dict[str, Any]: root = self.root / f"approval-{kind}-{secrets.token_hex(4)}" root.mkdir(mode=0o755) model = root / "model.gguf" hostile = root / "hostile.gguf" model.write_bytes(gguf(10)) hostile.write_bytes(gguf(20)) model.chmod(0o444) hostile.chmod(0o444) name = f"p0-{kind}-{secrets.token_hex(4)}" run_name = f"p0-run-{kind}-{secrets.token_hex(4)}" command = [str(self.runtime), "-m", str(model), "--version"] json_control( [ "approve", "--name", name, "--uid", str(self.policy["workload_uid"]), "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ] ) mounted = False try: if kind == "rename": os.replace(hostile, model) elif kind == "hardlink": model.unlink() os.link(hostile, model) elif kind == "symlink": model.unlink() model.symlink_to(hostile) elif kind == "exchange": libc = ctypes.CDLL(None, use_errno=True) result = libc.renameat2( -100, os.fsencode(model), -100, os.fsencode(hostile), 2, ) if result: error = ctypes.get_errno() raise OSError(error, os.strerror(error)) elif kind == "bind": result = subprocess.run( ["/usr/bin/mount", "--bind", str(hostile), str(model)], capture_output=True, text=True, check=False, timeout=30, ) if result.returncode: raise RuntimeError(result.stderr.strip() or "bind mount failed") mounted = True else: raise RuntimeError("unknown approval substitution") code, _stdout, _stderr = control( [ "start", "--name", run_name, "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ], operator=self.operator, ) if code == 0: raise RuntimeError(f"{kind} material substitution inherited approval") self.assert_canary() return {"case": f"model-{kind}-substitution", "result": "PASS"} finally: if mounted: subprocess.run( ["/usr/bin/umount", str(model)], capture_output=True, check=False, timeout=30, ) control(["revoke", "--name", name]) def overlay_mutation(self) -> dict[str, Any]: root = self.root / f"overlay-{secrets.token_hex(4)}" lower, upper, work, merged = ( root / "lower", root / "upper", root / "work", root / "merged", ) for path in (lower, upper, work, merged): path.mkdir(parents=True, mode=0o755) model = lower / "model.gguf" model.write_bytes(gguf(30)) model.chmod(0o444) result = subprocess.run( [ "/usr/bin/mount", "-t", "overlay", "overlay", "-o", f"lowerdir={lower},upperdir={upper},workdir={work}", str(merged), ], capture_output=True, text=True, check=False, timeout=30, ) if result.returncode: raise RuntimeError(result.stderr.strip() or "overlay mount failed") name = f"p0-overlay-{secrets.token_hex(4)}" run_name = f"p0-run-overlay-{secrets.token_hex(4)}" merged_model = merged / "model.gguf" command = [str(self.runtime), "-m", str(merged_model), "--version"] try: json_control( [ "approve", "--name", name, "--uid", str(self.policy["workload_uid"]), "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ] ) merged_model.chmod(0o644) merged_model.write_bytes(gguf(31)) merged_model.chmod(0o444) code, _stdout, _stderr = control( [ "start", "--name", run_name, "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ], operator=self.operator, ) if code == 0: raise RuntimeError("overlay material substitution inherited approval") return {"case": "model-overlay-copy-up-substitution", "result": "PASS"} finally: control(["revoke", "--name", name]) subprocess.run( ["/usr/bin/umount", str(merged)], capture_output=True, check=False, timeout=30, ) def environment_injection(self) -> dict[str, Any]: root = self.root / f"environment-{secrets.token_hex(4)}" inputs, outputs, hostile = root / "inputs", root / "outputs", root / "hostile" inputs.mkdir(parents=True, mode=0o755) outputs.mkdir(mode=0o733) hostile.mkdir(mode=0o755) os.chmod(outputs, 0o733) script = inputs / "approved.py" output = outputs / "result" marker_path = outputs / "sitecustomize-loaded" script.write_text( "import pathlib,time\n" "try:\n import p0_untrusted\n state='UNSAFE'\n" "except ModuleNotFoundError:\n state='SAFE'\n" f"pathlib.Path({str(output)!r}).write_text(state)\n" "time.sleep(120)\n", encoding="utf-8", ) script.chmod(0o444) (hostile / "p0_untrusted.py").write_text("VALUE='hostile'\n", encoding="utf-8") (hostile / "sitecustomize.py").write_text( f"open({str(marker_path)!r},'w').write('loaded')\n", encoding="utf-8", ) python = Path("/opt/lumi-eggcracker-torch-smoke/bin/python") command = [str(python), str(script), str(output)] name = f"p0-env-{secrets.token_hex(4)}" run_name = f"p0-run-env-{secrets.token_hex(4)}" json_control( [ "approve", "--name", name, "--uid", str(self.policy["workload_uid"]), "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ] ) started = False try: environment = dict(os.environ) environment.update( { "BASH_ENV": str(hostile / "sitecustomize.py"), "LD_AUDIT": "/definitely/absent.so", "LD_LIBRARY_PATH": str(hostile), "LD_PRELOAD": "/definitely/absent.so", "PYTHONPATH": str(hostile), "PYTHONSTARTUP": str(hostile / "sitecustomize.py"), "PYTHONUSERBASE": str(hostile), } ) code, stdout, stderr = control( [ "start", "--name", run_name, "--max-pids", "32", "--max-memory-mib", "1024", "--cpu-quota-percent", "400", "--", *command, ], operator=self.operator, environment=environment, ) if code: raise RuntimeError(stderr.strip() or stdout.strip() or "approved start failed") started = True deadline = time.monotonic() + 30 while time.monotonic() < deadline and not output.is_file(): time.sleep(0.02) if output.read_text(encoding="utf-8") != "SAFE" or marker_path.exists(): raise RuntimeError("operator-controlled launch environment reached the workload") return {"case": "python-loader-and-import-environment-injection", "result": "PASS"} finally: if started: receipt = self.work / f"operator-{secrets.token_hex(6)}.json" control( ["kill", "--name", run_name, "--receipt", str(receipt)], operator=self.operator, ) receipt.unlink(missing_ok=True) control(["revoke", "--name", name]) def approved_parent_unapproved_child(self) -> dict[str, Any]: root = self.root / f"parent-child-{secrets.token_hex(4)}" root.mkdir(mode=0o755) gate = root / "gate" gate.write_bytes(b"go\n") gate.chmod(0o444) ready = self.work / f"parent-child-{secrets.token_hex(4)}.json" parent = root / "approved-parent.py" parent.write_text( "import subprocess,time\n" "subprocess.Popen(" + repr( [ "/usr/bin/python3", "-I", "-S", str(self.script), "fixture-evidence", "--mode", "regular", "--model", str(self.model), "--runtime", str(self.runtime), "--gate", str(gate), "--ready", str(ready), "--work", str(self.work), ] ) + ")\n" "time.sleep(120)\n", encoding="utf-8", ) parent.chmod(0o444) python = Path("/opt/lumi-eggcracker-torch-smoke/bin/python") command = [str(python), str(parent)] name = f"p0-parent-{secrets.token_hex(4)}" run_name = f"p0-run-parent-{secrets.token_hex(4)}" json_control( [ "approve", "--name", name, "--uid", str(self.policy["workload_uid"]), "--max-pids", "64", "--max-memory-mib", "2048", "--cpu-quota-percent", "400", "--", *command, ] ) before = set(DETECTIONS.glob("*.json")) started = False try: response = json_control( [ "start", "--name", run_name, "--max-pids", "64", "--max-memory-mib", "2048", "--cpu-quota-percent", "400", "--", *command, ], operator=self.operator, ) started = True if response.get("state") != "RUNNING": raise RuntimeError("approved parent did not start") wait_ready(ready) deadline = time.monotonic() + 90 receipts: list[dict[str, Any]] = [] while time.monotonic() < deadline: receipts = receipts_after(before) if receipts: break time.sleep(0.05) validate_receipts( receipts, expected=1, source_commit=self.policy["source_commit"], ) # The detector receipt is written immediately after authoritative # cgroup containment. Run-state reconciliation follows in the # same supervisor worker, so a receipt observer can briefly win a # race with the terminal state update. Wait only for that # bounded bookkeeping transition; do not treat the transient # RUNNING response as a surviving parent. status = None deadline = time.monotonic() + 5 while time.monotonic() < deadline: status = json_control(["status", "--name", run_name], operator=self.operator) if status.get("state") == "TERMINATED": break time.sleep(0.01) if status is None or status.get("state") != "TERMINATED": raise RuntimeError("approved parent survived its unapproved supported child") started = False self.assert_canary() return {"case": "approved-parent-unapproved-supported-child", "result": "PASS"} finally: if started: receipt = self.work / f"operator-{secrets.token_hex(6)}.json" control( ["kill", "--name", run_name, "--receipt", str(receipt)], operator=self.operator, ) receipt.unlink(missing_ok=True) control(["revoke", "--name", name]) def run(self) -> dict[str, Any]: self.doctor() for kind in ("rename", "exchange", "hardlink", "symlink", "bind"): self.results["approval_material_substitution"].append( self.reject_mutation(kind) ) self.results["approval_material_substitution"].append(self.overlay_mutation()) self.results["approval_material_substitution"].append( self.environment_injection() ) self.results["approval_material_substitution"].append( self.approved_parent_unapproved_child() ) for mode in ( "memfd-model", "sealed-memfd-model", "otmpfile-model", "deleted-model", "memfd-runtime", "otmpfile-runtime", "deleted-runtime", ): self.results["pathless_deleted"].append(self.one_evidence(mode)) for mode in ( "memfd-exec", "execveat-memfd", "otmpfile-exec", "deleted-exec", "procfd-model", "sealed-procfd-model", ): self.results["pathless_deleted"].append(self.one_exec(mode)) for count in (17, 32, 64): self.results["saturation"].append(self.wave(count)) self.results["saturation"].append(self.storm(96, "complete")) self.results["saturation"].append(self.storm(512, "content")) self.results["saturation"].append( self.one_evidence("high-fd", pre_fds=700, post_fds=324, timeout=120) ) self.results["saturation"].append( self.one_evidence("high-maps", decoy_maps=600, timeout=180) ) final_doctor = self.doctor() self.assert_canary() return { "artifact_sha256": digest(INSTALLED_ARTIFACT), "doctor": final_doctor, "families": self.results, "result": "PASS", "schema_version": SCHEMA, "source_commit": self.policy["source_commit"], "version": self.policy["version"], } def close(self) -> None: stop(self.canary) if self.root.exists() and not self.root.is_symlink(): shutil.rmtree(self.root) def campaign_main(argv: list[str]) -> int: if os.geteuid() != 0: raise SystemExit("P0 native campaign must run as root") parser = argparse.ArgumentParser() parser.add_argument("--workload-user", required=True) parser.add_argument("--operator", required=True) parser.add_argument("--runtime", required=True, type=Path) parser.add_argument("--real-model", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) args = parser.parse_args(argv) if ( args.output.exists() or args.output.is_symlink() or not args.output.parent.is_dir() or any(path.is_symlink() or not path.is_file() for path in (args.runtime, args.real_model)) ): raise SystemExit("P0 output must be new and fixtures must be regular files") pwd.getpwnam(args.workload_user) pwd.getpwnam(args.operator) campaign = Campaign( script=Path(__file__).resolve(), workload=args.workload_user, operator=args.operator, runtime=args.runtime, real_model=args.real_model, output=args.output, ) try: value = campaign.run() args.output.write_text( json.dumps(value, sort_keys=True) + "\n", encoding="utf-8", ) print(json.dumps({"output": str(args.output), "result": "PASS"}, sort_keys=True)) return 0 except BaseException as error: if not args.output.exists(): args.output.write_text( json.dumps( { "error": f"{type(error).__name__}: {error}", "result": "FAIL", "schema_version": SCHEMA, }, sort_keys=True, ) + "\n", encoding="utf-8", ) raise finally: campaign.close() def main() -> int: if len(sys.argv) >= 2 and sys.argv[1] == "fixture-evidence": return fixture_evidence(sys.argv[2:]) if len(sys.argv) >= 2 and sys.argv[1] == "fixture-storm": return fixture_storm(sys.argv[2:]) if len(sys.argv) >= 2 and sys.argv[1] == "fixture-exec": return fixture_exec(sys.argv[2:]) return campaign_main(sys.argv[1:]) if __name__ == "__main__": raise SystemExit(main())