285 lines
13 KiB
Python
285 lines
13 KiB
Python
#!/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())
|