#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Fault-injection tests for the inactive Phase-1.0X transport layer.""" from __future__ import annotations import argparse from datetime import datetime, timezone import importlib.util import json from pathlib import Path import sys import tempfile def load(path: Path, name: str): spec = importlib.util.spec_from_file_location(name, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def require(value: bool, message: str) -> None: if not value: raise RuntimeError(message) class FakeClock: def __init__(self) -> None: self.value = 100.0 def monotonic(self) -> float: return self.value class FakeAdapter: def __init__(self, boundaries, events, clock, advance=0.0, fail=None): self.boundaries = dict(boundaries) self.events = events self.clock = clock self.advance = advance self.fail = fail self.open_count = 0 self.closed = False def open_once(self, _plan, _remaining): self.events.append("OPEN") self.open_count += 1 if self.fail == "OPEN": raise RuntimeError("SENSITIVE RAW OPEN ERROR") def receive_boundary(self, boundary, _remaining): self.events.append(boundary) self.clock.value += self.advance if self.fail == boundary: raise RuntimeError("SENSITIVE RAW RECEIVE ERROR") return self.boundaries.get(boundary, []) def send_command_token(self, command, _path, _remaining): self.events.append(f"SEND_{command.upper()}") if self.fail == f"SEND_{command.upper()}": raise RuntimeError("SENSITIVE RAW SEND ERROR") def close_once(self): self.events.append("CLOSE") self.closed = True if self.fail == "CLOSE": raise RuntimeError("SENSITIVE RAW CLOSE ERROR") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() root = args.root.resolve() sys.path.insert(0, str(root / "tools")) policy = load(root / "tools/phase10w_shsrv_client_policy.py", "phase10w_policy_x") transport = load(root / "tools/phase10x_inactive_transport.py", "phase10x_transport") now = datetime(2026, 7, 22, 12, 0, tzinfo=timezone.utc) def plan(window="T2_GREETING_AND_HELP", run_id="synthetic_run_x"): path = None if window == "T2_GREETING_AND_HELP" else "/data/exact.elf" common = { "active": True, "policy_sha256": "a" * 64, "collector_sha256": policy.COLLECTOR_SHA256, "run_id": run_id, "target_address": "device.invalid", "target_port": 2323, "window": window, "exact_literal_path": path, "commands": list(policy.WINDOW_COMMANDS[window]), "deadline_seconds": 10, "expires_at": "2026-07-22T12:10:00Z", } approval = dict(common) approval.update({ "attested": True, "listener_already_running_attested": True, "ps5_connection_authorized": True, "device_request_authorized": True, "result_receive_authorized": True, "spawned_shell_effects_accepted": True, "automatic_serial_query_accepted": True, "automatic_telemetry_query_accepted": True, "sanitized_output_only_accepted": True, "physical_memory_erasure_unproven_accepted": True, "target_build_authorized": False, "device_transfer_authorized": False, "device_execution_authorized": False, "installation_authorized": False, "autoload_authorized": False, "device_write_authorized": False, "automatic_retry": False, "reconnect_authorized": False, "resume_authorized": False, "fallback_authorized": False, }) return policy.build_session_plan(common, approval, now) greeting = ( b"Welcome to shsrv.elf running on pid 1, compiled Jul 22 2026 at 12:34:56\n" b"S/N: SENSITIVE-SERIAL\nS/W: 9.60\nCPU freq: 3500 MHz\n") help_output = b"Builtin commands:\n help\n\n" cases = [] def case(name): def register(function): cases.append((name, function)) return function return register def execute(directory, session_plan=None, adapter=None, clock=None): session_plan = session_plan or plan() clock = clock or FakeClock() events = [] adapter = adapter or FakeAdapter({"INITIAL_PROMPT": [greeting], "AFTER_HELP": [help_output]}, events, clock) store = transport.ExclusiveEvidenceStore(Path(directory)) return transport.run_injected_session(session_plan, adapter, clock, store), events @case("01 receipt exists before fake open") def _(): with tempfile.TemporaryDirectory() as directory: session_plan = plan(); clock = FakeClock(); events = [] class InspectAdapter(FakeAdapter): def open_once(self, p, remaining): require((Path(directory) / f"{p.run_id}.consumed.json").exists(), "receipt missing before open") super().open_once(p, remaining) adapter = InspectAdapter({"INITIAL_PROMPT": [greeting], "AFTER_HELP": [help_output]}, events, clock) execute(directory, session_plan, adapter, clock) @case("02 help flow is one shot") def _(): with tempfile.TemporaryDirectory() as directory: outcome, events = execute(directory) require(events == ["OPEN", "INITIAL_PROMPT", "SEND_HELP", "AFTER_HELP", "CLOSE"] and outcome.exact_identity is False, "help sequence mismatch") @case("03 exact-path flow preserves command order") def _(): with tempfile.TemporaryDirectory() as directory: p = plan("T3_ONE_EXACT_PATH"); clock = FakeClock(); events = [] adapter = FakeAdapter({"INITIAL_PROMPT": [greeting], "AFTER_STAT": [b"filename: /data/exact.elf\nsize: 123\n"], "AFTER_SUM": [b"12345 /data/exact.elf\n"]}, events, clock) outcome, _ = execute(directory, p, adapter, clock) require(events == ["OPEN", "INITIAL_PROMPT", "SEND_STAT", "AFTER_STAT", "SEND_SUM", "AFTER_SUM", "CLOSE"] and outcome.classification == "WEAK_FILE_CORRELATION_ONLY", "path sequence mismatch") @case("04 sanitized output excludes serial and telemetry") def _(): with tempfile.TemporaryDirectory() as directory: outcome, _ = execute(directory); data = outcome.output.path.read_text(encoding="ascii") require("SENSITIVE-SERIAL" not in data and "3500" not in data, "sensitive output retained") @case("05 receipt excludes target") def _(): with tempfile.TemporaryDirectory() as directory: outcome, _ = execute(directory); data = outcome.receipt.path.read_text(encoding="ascii") require("device.invalid" not in data and '"target_retained":false' in data, "target retained") @case("06 second run is blocked before fake open") def _(): with tempfile.TemporaryDirectory() as directory: execute(directory); clock = FakeClock(); events = [] adapter = FakeAdapter({}, events, clock) try: execute(directory, plan(), adapter, clock) except transport.EvidenceFailure: require(events == [], "adapter opened after consumed receipt"); return raise RuntimeError("second run accepted") @case("07 existing output cannot be overwritten") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); store = transport.ExclusiveEvidenceStore(Path(directory)); receipt = store.create_consumed_receipt(p, 100.0) store.create_sanitized_output(p, receipt, {"classification": "x", "exact_identity": False}) try: store.create_sanitized_output(p, receipt, {"classification": "y"}) except transport.EvidenceFailure: return raise RuntimeError("output overwrite accepted") @case("08 deadline expiry fails and closes") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); clock = FakeClock(); events = [] adapter = FakeAdapter({"INITIAL_PROMPT": [greeting]}, events, clock, advance=11.0) try: execute(directory, p, adapter, clock) except transport.SessionFailure: require(adapter.closed, "adapter not closed"); return raise RuntimeError("expired session accepted") @case("09 adapter open failure is normalized and closed") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); clock = FakeClock(); events = [] adapter = FakeAdapter({}, events, clock, fail="OPEN") try: execute(directory, p, adapter, clock) except transport.SessionFailure as error: require( "SENSITIVE" not in str(error) and error.__cause__ is None and error.__context__ is None and adapter.closed, "adapter detail/cause leaked or close skipped") return raise RuntimeError("adapter failure accepted") @case("10 adapter closes after send failure") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); clock = FakeClock(); events = [] adapter = FakeAdapter({"INITIAL_PROMPT": [greeting]}, events, clock, fail="SEND_HELP") try: execute(directory, p, adapter, clock) except transport.SessionFailure: require(adapter.closed and adapter.open_count == 1, "cleanup/retry mismatch"); return raise RuntimeError("send failure accepted") @case("11 excessive boundary chunks are rejected") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); clock = FakeClock(); events = [] adapter = FakeAdapter({"INITIAL_PROMPT": [b"x"] * 65}, events, clock) try: execute(directory, p, adapter, clock) except transport.SessionFailure: return raise RuntimeError("excess chunks accepted") @case("12 malformed transcript creates no output") def _(): with tempfile.TemporaryDirectory() as directory: p = plan(); clock = FakeClock(); events = [] adapter = FakeAdapter({"INITIAL_PROMPT": [b"\xff"]}, events, clock) try: execute(directory, p, adapter, clock) except transport.SessionFailure: require(not (Path(directory) / f"{p.run_id}.sanitized.json").exists(), "invalid output created"); return raise RuntimeError("malformed transcript accepted") @case("13 output reopens as valid JSON") def _(): with tempfile.TemporaryDirectory() as directory: outcome, _ = execute(directory) receipt = json.loads(outcome.receipt.path.read_text(encoding="ascii")) value = json.loads(outcome.output.path.read_text(encoding="ascii")) require( receipt["status"] == "CONSUMED_BEFORE_ADAPTER_OPEN" and value["status"] == "SANITIZED_OUTPUT_COMPLETE" and value["receipt_sha256"] == outcome.receipt.sha256, "output binding mismatch") @case("14 output byte count and hash match reopen") def _(): with tempfile.TemporaryDirectory() as directory: outcome, _ = execute(directory); payload = outcome.output.path.read_bytes() import hashlib require(len(payload) == outcome.output.size and hashlib.sha256(payload).hexdigest() == outcome.output.sha256, "reopen identity mismatch") @case("15 evidence store exposes no cleanup") def _(): require(not any(hasattr(transport.ExclusiveEvidenceStore, name) for name in ("delete", "cleanup", "overwrite")), "cleanup API present") @case("16 transport module exposes no live CLI") def _(): require(not hasattr(transport, "main"), "live CLI present") @case("17 transport module imports no socket") def _(): source = (root / "tools/phase10x_inactive_transport.py").read_text(encoding="utf-8") require("import socket" not in source and "from socket" not in source, "socket import present") @case("18 host-only result is not exact identity") def _(): with tempfile.TemporaryDirectory() as directory: outcome, _ = execute(directory); require(outcome.exact_identity is False, "identity promoted") failures = [] for name, function in cases: try: function(); print(f"PASS {name}") except Exception as error: # noqa: BLE001 - fault harness failures.append(f"{name}: {error}"); print(f"FAIL {name}: {error}") if failures: return 1 print(f"Phase-1.0X inactive-transport tests passed: {len(cases)}") return 0 if __name__ == "__main__": raise SystemExit(main())