#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Synthetic offline tests for the Phase-1.0V collector model.""" from __future__ import annotations import argparse import importlib.util import json from pathlib import Path import sys def load(path: Path): spec = importlib.util.spec_from_file_location("phase10v_model", 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) def greeting(extra: str = "") -> bytes: return ( "Welcome to shsrv.elf running on pid 123, compiled Jul 22 2026 at 12:34:56\n" "Model: synthetic-model\n" "S/N: SYNTHETIC-SERIAL-NEVER-RETAIN\n" "S/W: 9.60\n" "SoC temp: 40 C\n" "CPU temp: 41 C\n" "CPU freq: 3500 MHz\n" + extra).encode("utf-8") 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")) module = load(root / "tools/phase10v_shsrv_collector_model.py") cases = [] def case(name): def register(function): cases.append((name, function)) return function return register @case("01 plain greeting is parsed offline") def _(): collector = module.OfflineCollector(); collector.feed(greeting()) require(collector.finalize()["classification"] == "COMPILE_METADATA_ONLY", "greeting rejected") @case("02 Telnet negotiation is removed") def _(): collector = module.OfflineCollector(); collector.feed(bytes([255, 251, 1]) + greeting()) require(collector.finalize()["compile_metadata"]["firmware"] == "9.60", "negotiation leaked") @case("03 fragmented Telnet negotiation is handled") def _(): collector = module.OfflineCollector(); collector.feed(bytes([255])); collector.feed(bytes([253])); collector.feed(bytes([3]) + greeting()) require(collector.finalize()["classification"] == "COMPILE_METADATA_ONLY", "fragmentation failed") @case("04 Telnet subnegotiation is removed") def _(): control = bytes([255, 250, 24, 1, 2, 3, 255, 240]) collector = module.OfflineCollector(); collector.feed(control + greeting()) require(collector.finalize()["classification"] == "COMPILE_METADATA_ONLY", "subnegotiation leaked") @case("04b doubled IAC stays inside subnegotiation") def _(): control = bytes([255, 250, 24, 255, 255, 1, 255, 240]) collector = module.OfflineCollector(); collector.feed(control + greeting()) require(collector.finalize()["classification"] == "COMPILE_METADATA_ONLY", "doubled IAC state failed") @case("05 serial is absent from output") def _(): collector = module.OfflineCollector(); collector.feed(greeting()); result = collector.finalize() require("SYNTHETIC-SERIAL" not in json.dumps(result), "serial retained") @case("06 telemetry values are absent from output") def _(): collector = module.OfflineCollector(); collector.feed(greeting()); result = collector.finalize() require("3500" not in json.dumps(result), "telemetry retained") @case("07 exact identity always remains false") def _(): collector = module.OfflineCollector(); collector.feed(greeting()) require(collector.finalize()["exact_identity"] is False, "identity promoted") @case("08 raw byte limit is enforced") def _(): collector = module.OfflineCollector() try: collector.feed(b"x" * (module.MAX_RAW_BYTES + 1)) except module.CollectorError: return raise RuntimeError("oversized input accepted") @case("09 chunk limit is enforced") def _(): collector = module.OfflineCollector() try: for _index in range(module.MAX_CHUNKS + 1): collector.feed(b"x") except module.CollectorError: return raise RuntimeError("excess chunks accepted") @case("09b empty chunks do not consume the limit") def _(): collector = module.OfflineCollector() for _index in range(module.MAX_CHUNKS + 1): collector.feed(b"") require(collector.chunk_count == 0, "empty chunks counted") @case("10 incomplete Telnet sequence is rejected") def _(): collector = module.OfflineCollector(); collector.feed(bytes([255])) try: collector.finalize() except module.CollectorError: return raise RuntimeError("incomplete control accepted") @case("11 invalid UTF-8 is rejected") def _(): collector = module.OfflineCollector(); collector.feed(bytes([0xC3, 0x28])) try: collector.finalize() except module.CollectorError: return raise RuntimeError("invalid UTF-8 accepted") @case("12 collector finalizes only once") def _(): collector = module.OfflineCollector(); collector.feed(greeting()); collector.finalize() try: collector.finalize() except module.CollectorError: return raise RuntimeError("second finalize accepted") @case("13 feed after finalization is rejected") def _(): collector = module.OfflineCollector(); collector.finalize() try: collector.feed(b"later") except module.CollectorError: return raise RuntimeError("post-finalize feed accepted") @case("14 abort prevents output") def _(): collector = module.OfflineCollector(); collector.feed(greeting()); collector.abort() try: collector.finalize() except module.CollectorError: return raise RuntimeError("aborted collector finalized") @case("15 approved literal-path metadata is retained") def _(): collector = module.OfflineCollector(); collector.feed(greeting("filename: /approved\nsize: 123\n")) result = collector.finalize({"/approved"}) require(result["file_observations"][0]["size"] == 123, "approved metadata lost") @case("16 unknown path is discarded and memory erasure unproven") def _(): collector = module.OfflineCollector(); collector.feed(greeting("filename: /unknown\nsize: 123\n")) result = collector.finalize({"/approved"}) require(result["file_observations"] == [] and result["collector_model"]["physical_memory_erasure_proven"] is False, "boundary promoted") @case("17 unsafe or non-normalized expected paths are rejected") def _(): for path in ("relative", "/safe/../escape", "/wild*card", "/line\nfeed"): collector = module.OfflineCollector(); collector.feed(greeting()) try: collector.finalize({path}) except module.CollectorError: continue raise RuntimeError(f"unsafe path accepted: {path!r}") @case("18 unexpected firmware metadata is rejected") def _(): collector = module.OfflineCollector(); collector.feed(greeting().replace(b"S/W: 9.60", b"S/W: secret")) try: collector.finalize() except module.CollectorError: return raise RuntimeError("unexpected firmware accepted") @case("19 malformed compile metadata is rejected") def _(): data = greeting().replace(b"compiled Jul 22 2026", b"compiled SERIAL-IN-DATE") collector = module.OfflineCollector(); collector.feed(data) try: collector.finalize() except module.CollectorError: return raise RuntimeError("malformed compile metadata accepted") failures = [] for name, function in cases: try: function() print(f"PASS {name}") except Exception as error: # noqa: BLE001 - synthetic harness failures.append(f"{name}: {error}") print(f"FAIL {name}: {error}") if failures: return 1 print(f"Phase-1.0V collector-model tests passed: {len(cases)}") return 0 if __name__ == "__main__": raise SystemExit(main())