"""Lesson 2: finite two-state fault-model lab. Python 3, standard library only. This is not an RTL simulator, a formal proof, or a physical injection model. Every run is ONE reset-to-edge-7 boot attempt with ZERO or ONE event. Requests can be accepted at edges 4 and 6; those are not separate attempts. At each edge: apply pulse controls; sample pre-update commit; update state. For state_upset only, flip the new stored result after the normal update. Reference image/policy, completion, clock/reset and acceptance are trusted. No fault detector is implemented. NO_UNAUTHORIZED_COMMIT_IN_WINDOW is bounded. """ from dataclasses import asdict, dataclass import argparse import json from pathlib import Path @dataclass(frozen=True) class Fault: kind: str start: int duration: int = 1 def __post_init__(self): if self.kind not in {"q_pulse", "state_upset", "source_pulse", "grant_pulse"}: raise ValueError("unknown fault target/effect") if self.start not in range(2, 8) or self.duration not in (1, 2): raise ValueError("campaign: start 2..7, pulse duration 1 or 2 edges") if self.kind == "state_upset" and self.duration != 1: raise ValueError("one state-upset event; retention is not pulse duration") def active(self, edge): return self.start <= edge < self.start + self.duration def run(authorized=False, fault=None, horizon=8, fetch_edges=(4, 6)): checked = result = ref_complete = 0 trace = [] for edge in range(horizon): reset_n = int(edge != 0) done = int(edge == 2) active = fault is not None and fault.active(edge) source_xor = int(active and fault.kind == "source_pulse") q_xor = int(active and fault.kind == "q_pulse") grant_xor = int(active and fault.kind == "grant_pulse") # A state upset happens once, after this edge's normal update. state_xor = int(fault is not None and fault.kind == "state_upset" and edge == fault.start) auth_seen = int(authorized) ^ source_xor result_seen = result ^ q_xor grant = (checked & result_seen) ^ grant_xor valid = int(edge in fetch_edges) ready = 1 commit = bool(reset_n and valid and ready and grant) unauthorized = commit and not (ref_complete and authorized) row = dict(edge=edge, checked_pre=checked, raw_pre=result, result_seen=result_seen, auth_seen=auth_seen, ref_complete_pre=ref_complete, ref_pass=int(authorized), verify_done=done, valid=valid, ready=ready, grant=grant, commit=commit, unauthorized=unauthorized, source_xor=source_xor, q_xor=q_xor, grant_xor=grant_xor, state_xor=state_xor) if not reset_n: checked = result = ref_complete = 0 else: if done and not checked: checked, result = 1, auth_seen result ^= state_xor # Independent policy completion; verify_done itself is NOT faulted. if done: ref_complete = 1 row.update(raw_post=result, checked_post=checked, ref_complete_post=ref_complete) trace.append(row) bad_edges = [r["edge"] for r in trace if r["unauthorized"]] return dict(authorized=authorized, fault=asdict(fault) if fault else None, budget=dict(events_per_attempt=int(fault is not None), locations=int(fault is not None), target_bits=int(fault is not None)), observation_edges=[0, horizon - 1], accepted_edges=[r["edge"] for r in trace if r["commit"]], unauthorized_edges=bad_edges, outcome="UNAUTHORIZED_COMMIT" if bad_edges else "NO_UNAUTHORIZED_COMMIT_IN_WINDOW", detector="NOT_IMPLEMENTED", trace=trace) def exercise(): controls = [run(True), run(False)] assert controls[0]["accepted_edges"] == [4, 6] assert controls[1]["accepted_edges"] == [] specs = [Fault(kind, start, duration) for kind in ("q_pulse", "state_upset", "source_pulse", "grant_pulse") for start in range(2, 8) for duration in ((1,) if kind == "state_upset" else (1, 2))] campaign = [run(False, f) for f in specs] assert len(campaign) == 42 for f, r in zip(specs, campaign): assert r["budget"] == dict(events_per_attempt=1, locations=1, target_bits=1) assert r["trace"][2]["ref_complete_pre"] == 0 assert r["trace"][2]["ref_complete_post"] == 1 assert all(t["ref_pass"] == 0 for t in r["trace"]) if f.kind in ("q_pulse", "grant_pulse"): assert all(t["raw_pre"] == t["raw_post"] == 0 for t in r["trace"]) if f.kind == "state_upset": assert all(t["raw_post"] == int(t["edge"] >= f.start) for t in r["trace"]) if f.kind == "source_pulse": assert r["accepted_edges"] == ([4, 6] if f.start == 2 else []) examples = { "q pulse at 3, one edge": (Fault("q_pulse", 3), []), "stored upset after 3": (Fault("state_upset", 3), [4, 6]), "source pulse before 2": (Fault("source_pulse", 2), [4, 6]), "source pulse before 3": (Fault("source_pulse", 3), []), "grant pulse before 4": (Fault("grant_pulse", 4), [4]), "stored upset after 6": (Fault("state_upset", 6), []), } for name, (f, expected) in examples.items(): result = run(False, f) assert result["unauthorized_edges"] == expected print(f"{name}: {result['outcome']}; unauthorized edges={expected}; " "window=0..7; detector=NOT_IMPLEMENTED") # Same late upset, changed request schedule: the earlier negative was bounded. assert run(False, Fault("state_upset", 6), fetch_edges=(4, 6, 7))["unauthorized_edges"] == [7] bad = sum(bool(r["unauthorized_edges"]) for r in campaign) assert bad == 18 print("PASS: 2 controls + 42 fault cases; 18 unauthorized, 24 no unauthorized " "commit observed in window. Expected counterexamples reproduced.") return dict(model="lesson02-v1", controls=controls, campaign=campaign, counts=dict(controls=2, fault_cases=42, unauthorized=18, no_unauthorized_commit_in_window=24), caveat="Finite two-state Python model; detector not implemented; " "not RTL/formal/silicon security validation.") if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output-dir", type=Path) args = parser.parse_args() evidence = exercise() if args.output_dir: args.output_dir.mkdir(parents=True, exist_ok=True) (args.output_dir / "lesson02-results.json").write_text( json.dumps(evidence, indent=2) + "\n", encoding="utf-8")