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