"""Maintainer exercise: observe real reads, original receipts and interrupted work. The normal reader workflow uses the two binaries directly. This driver retains its raw results and private state in --output; it never simulates a kernel result. """ import argparse import hashlib import json import os import platform import shutil import signal import sqlite3 import subprocess import time from pathlib import Path def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", type=Path, required=True) parser.add_argument("--client", type=Path, required=True) parser.add_argument("--chio", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() host_binary, client_binary, chio = [p.resolve(strict=True) for p in [args.host, args.client, args.chio]] output = args.output.resolve() output.mkdir(mode=0o700, parents=True, exist_ok=False) state = output / "state" repository = output / "repository" shutil.copytree(Path(__file__).parent / "repository", repository) (repository / "README.md").write_text("This file is outside the source grant.\n") evidence = output / "evidence" evidence.mkdir(mode=0o700) journal = state / "resource-reads.jsonl" processes = [] records = {} def start(label): log_path = evidence / f"{label}.log" with log_path.open("wb") as log: process = subprocess.Popen([str(host_binary), str(state), str(repository)], stdout=log, stderr=subprocess.STDOUT) processes.append(process) deadline = time.monotonic() + 30 while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError(log_path.read_text()) if '"ready":true' in log_path.read_text(): return process time.sleep(.05) raise RuntimeError("host did not become ready") def stop(process, sig=signal.SIGINT): process.send_signal(sig) process.wait(timeout=20) def invoke(label, key, path, receipt): command = [str(client_binary), key, path, str(receipt)] with (state / "connection.json").open("rb") as boot: result = subprocess.run(command, stdin=boot, capture_output=True, timeout=40) (evidence / f"{label}.stdout").write_bytes(result.stdout) (evidence / f"{label}.stderr").write_bytes(result.stderr) records[label] = {"command": command, "exit": result.returncode} return result def verify(label, receipt): result = subprocess.run([str(chio), "receipt", "verify", "--input", str(receipt), "--trusted-kernel-pubkey", str(state / "kernel.pub")], capture_output=True) (evidence / f"{label}.verify.stdout").write_bytes(result.stdout) (evidence / f"{label}.verify.stderr").write_bytes(result.stderr) assert result.returncode == 0, result.stderr try: host = start("initial-host") receipt = evidence / "original.receipt.json" original = invoke("original", "read-pricing-v1", "src/pricing.py", receipt) assert original.returncode == 0, original.stderr value = json.loads(original.stdout) source = (repository / "src/pricing.py").read_bytes() assert value["text"].encode() == source assert value["sha256"] == hashlib.sha256(source).hexdigest() assert len(journal.read_text().splitlines()) == 1 verify("original", receipt) signed = receipt.read_bytes() denied = invoke("refused", "read-parent-v1", "../README.md", evidence / "refused.receipt.json") assert denied.returncode == 2 and json.loads(denied.stdout)["status"] == "denied" assert len(journal.read_text().splitlines()) == 1 verify("refused", evidence / "refused.receipt.json") (repository / "src/link.py").symlink_to("../README.md") symlink = invoke("symlink", "read-symlink-v1", "src/link.py", evidence / "symlink.receipt.json") assert symlink.returncode != 0 assert len(journal.read_text().splitlines()) == 1 conflict = invoke("conflict", "read-pricing-v1", "src/other.py", evidence / "conflict.receipt.json") assert conflict.returncode != 0 and b"conflict" in conflict.stderr assert len(journal.read_text().splitlines()) == 1 competing = subprocess.run([str(host_binary), str(state), str(repository)], capture_output=True, timeout=10) assert competing.returncode != 0 and b"another host owns this state" in competing.stderr (evidence / "competing-host.stderr").write_bytes(competing.stderr) stop(host) (repository / "src/pricing.py").write_text(source.decode().replace("TIER_PERCENT = 10", "TIER_PERCENT = 15")) host = start("replacement-host") replay = invoke("replay", "read-pricing-v1", "src/pricing.py", receipt) assert replay.returncode == 0 and replay.stdout == original.stdout assert receipt.read_bytes() == signed assert len(journal.read_text().splitlines()) == 1 pause = state / "pause-after-read" pause.touch(mode=0o600) with (state / "connection.json").open("rb") as boot: client = subprocess.Popen([str(client_binary), "read-crash-v1", "src/pricing.py", str(evidence / "interrupted.receipt.json")], stdin=boot, stdout=subprocess.PIPE, stderr=subprocess.PIPE) processes.append(client) deadline = time.monotonic() + 15 while len(journal.read_text().splitlines()) != 2: assert client.poll() is None, "client exited before the read observation" assert time.monotonic() < deadline, "read observation did not arrive" time.sleep(.05) stop(host, signal.SIGKILL) stdout, stderr = client.communicate(timeout=10) assert client.returncode != 0 (evidence / "interrupted.stdout").write_bytes(stdout) (evidence / "interrupted.stderr").write_bytes(stderr) pause.unlink() host = start("crash-replacement-host") unknown_receipt = evidence / "unknown.receipt.json" unknown = invoke("unknown", "read-crash-v1", "src/pricing.py", unknown_receipt) assert unknown.returncode == 2 and "OutcomeUnknownAfterDispatch" in json.loads(unknown.stdout)["reason"] assert len(journal.read_text().splitlines()) == 2 verify("unknown", unknown_receipt) stop(host) # All serving processes are stopped. Immutable reads cannot create WAL files. with sqlite3.connect(f"file:{state / 'authority.db'}?mode=ro&immutable=1", uri=True) as db: operations = [{"state": state_name, "record": json.loads(record)} for state_name, record in db.execute("SELECT state, operation_json FROM admission_operations ORDER BY created_at_unix_ms")] assert {x["state"] for x in operations} == {"completed", "outcome_unknown_after_dispatch"} (evidence / "operations.json").write_text(json.dumps(operations, indent=2) + "\n") shutil.copyfile(journal, evidence / "resource-reads.jsonl") shutil.copyfile(state / "kernel.pub", evidence / "kernel.pub") report = {"host": platform.platform(), "binaries": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in [host_binary, client_binary, chio]}, "invocations": records, "sourceBytes": len(source), "resourceReads": 2, "checks": ["returned source and digest", "scope refusal without read", "in-scope symlink refused by connector without read", "changed-request conflict", "exclusive host owner", "original receipt after restart and source change", "unknown outcome after killed dispatch without reread", "independent receipt signature and signer verification"], "scope": "embedded connector; no native cage or Linux launch qualification"} (evidence / "report.json").write_text(json.dumps(report, indent=2) + "\n") print(json.dumps({"passed": report["checks"], "evidence": str(evidence)}, indent=2)) finally: for process in processes: if process.poll() is None: process.kill() process.wait() if __name__ == "__main__": main()