227 lines
10 KiB
Python
227 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Host-only integration tests for the Phase-1.0AA exact fake adapter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
CURRENT_COMMANDS = "authid browse cat cd chgrp chmod chown chroot cmp cp df echo env exec exit export file find grep hbdbg hbldr hexdump http2_get id kill launch ln ls mkdir mknod mount mv notify pkg_install procstat ps pwd reptyr rm rmdir sfocreate sfoinfo sleep stat sum suspend sync sysctl touch umount".split()
|
|
|
|
|
|
def load(path: Path):
|
|
spec = importlib.util.spec_from_file_location("phase10aa_fake", 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 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/phase10aa_offline_fake_batch.py")
|
|
from phase10w_shsrv_client_policy import SessionPlan
|
|
|
|
def plan(window="T2_GREETING_AND_HELP", path=None):
|
|
commands = ("help",) if window == "T2_GREETING_AND_HELP" else ("stat", "sum")
|
|
return SessionPlan("synthetic_aa", "must-not-persist.invalid", 2323,
|
|
window, path, commands, 10,
|
|
datetime(2030, 1, 1, tzinfo=timezone.utc))
|
|
|
|
def greeting(extra="", eol="\n"):
|
|
text = ("Welcome to shsrv.elf running on pid 1, compiled Jul 22 2026 at 12:34:56\n"
|
|
"Model: synthetic\nS/N: SECRET-SERIAL\nS/W: 9.60\n"
|
|
"SoC temp: 40 C\nCPU temp: 41 C\nCPU freq: 3500 MHz\n" + extra)
|
|
return text.replace("\n", eol).encode("ascii")
|
|
|
|
def help_text():
|
|
return "Builtin commands:\n" + "".join(
|
|
f" {command} - synthetic\n" for command in CURRENT_COMMANDS) + "\n"
|
|
|
|
def events(data, data_advance=1.0, deadline_advance=9.0):
|
|
return (module.FakeReceiveEvent(module.EVENT_DATA, data, data_advance),
|
|
module.FakeReceiveEvent(module.EVENT_HARD_DEADLINE, b"", deadline_advance))
|
|
|
|
def execute(directory, session_plan=None, scripted=None):
|
|
p = session_plan or plan(); clock = module.OfflineFakeClock()
|
|
adapter = module.OfflineFakeBatchAdapter(clock, scripted or events(greeting(help_text())))
|
|
store = module.OfflineFakeEvidenceStore(Path(directory))
|
|
return module.run_offline_fake_batch(p, adapter, clock, store), adapter
|
|
|
|
def expect_failure(function):
|
|
try: function()
|
|
except (module.OfflineIntegrationError, module.FakeAdapterError): return
|
|
raise RuntimeError("invalid fake integration was accepted")
|
|
|
|
cases = []
|
|
def case(name):
|
|
def register(function): cases.append((name, function)); return function
|
|
return register
|
|
|
|
@case("01 exact fake help integration succeeds")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, adapter = execute(d)
|
|
require(outcome.classification == "SOURCE_FAMILY_FINGERPRINT_ONLY" and adapter.send_count == 1, "help integration failed")
|
|
|
|
@case("02 exact path integration sends one batch")
|
|
def _():
|
|
path = "/data/a.elf"; data = greeting(f"filename: {path}\nsize: 123\n12345 {path}\n")
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, adapter = execute(d, plan("T3_ONE_EXACT_PATH", path), events(data))
|
|
require(outcome.classification == "WEAK_FILE_CORRELATION_ONLY" and adapter.send_count == 1, "path integration failed")
|
|
|
|
@case("03 receipt precedes fake open")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d)
|
|
require(outcome.trace[:3] == ("RECEIPT_CREATED", "FAKE_OPEN", "FAKE_SEND_ONE_BATCH"), "ordering mismatch")
|
|
|
|
@case("04 deadline is the only seal event")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d)
|
|
output = json.loads(outcome.output.path.read_text())
|
|
require(output["result"]["passive_batch_contract"]["sealed_by_synthetic_hard_deadline"] is True, "deadline seal missing")
|
|
|
|
@case("05 early deadline fails")
|
|
def _():
|
|
scripted = (module.FakeReceiveEvent(module.EVENT_DATA, greeting(help_text()), 1), module.FakeReceiveEvent(module.EVENT_HARD_DEADLINE, b"", 1))
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=scripted))
|
|
|
|
@case("06 remote EOF fails")
|
|
def _():
|
|
scripted = (module.FakeReceiveEvent(module.EVENT_DATA, greeting(help_text()), 1), module.FakeReceiveEvent(module.EVENT_REMOTE_EOF, b"", 1))
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=scripted))
|
|
|
|
@case("07 blocked fake receive fails")
|
|
def _():
|
|
scripted = (module.FakeReceiveEvent(module.EVENT_BLOCKED, b"", 10),)
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=scripted))
|
|
|
|
@case("08 data at deadline fails")
|
|
def _():
|
|
scripted = (module.FakeReceiveEvent(module.EVENT_DATA, greeting(help_text()), 10),)
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=scripted))
|
|
|
|
@case("09 missing deadline event fails")
|
|
def _():
|
|
scripted = (module.FakeReceiveEvent(module.EVENT_DATA, greeting(help_text()), 1),)
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=scripted))
|
|
|
|
@case("10 incoming IAC fails")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=events(b"x\xffy")))
|
|
|
|
@case("11 partial help fails at deadline")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d: expect_failure(lambda: execute(d, scripted=events(greeting("Builtin commands:\n help - partial\n\n"))))
|
|
|
|
@case("12 CRLF transcript succeeds")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d, scripted=events(greeting(help_text(), "\r\n")))
|
|
require(outcome.exact_identity is False, "CRLF result promoted")
|
|
|
|
@case("13 close occurs after success")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
_, adapter = execute(d); require(adapter.close_count == 1 and adapter.state == "CLOSED", "success not closed")
|
|
|
|
@case("14 close occurs after failure")
|
|
def _():
|
|
clock = module.OfflineFakeClock(); adapter = module.OfflineFakeBatchAdapter(clock, (module.FakeReceiveEvent(module.EVENT_REMOTE_EOF),))
|
|
with tempfile.TemporaryDirectory() as d:
|
|
store = module.OfflineFakeEvidenceStore(Path(d)); expect_failure(lambda: module.run_offline_fake_batch(plan(), adapter, clock, store))
|
|
require(adapter.close_count == 1 and adapter.state == "CLOSED", "failure not closed")
|
|
|
|
@case("15 second fake send is impossible")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
_, adapter = execute(d); expect_failure(lambda: adapter.send_one_batch(None))
|
|
|
|
@case("16 exclusive receipt collision fails")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
execute(d); expect_failure(lambda: execute(d))
|
|
|
|
@case("17 target is absent from evidence")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d); combined = outcome.receipt.path.read_text() + outcome.output.path.read_text()
|
|
require("must-not-persist" not in combined and "target_address" not in combined, "target retained")
|
|
|
|
@case("18 serial and raw transcript are absent from output")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d); value = outcome.output.path.read_text()
|
|
require("SECRET-SERIAL" not in value and "Welcome to shsrv" not in value and '"raw_transcript_persisted":false' in value, "sensitive input retained")
|
|
|
|
@case("19 receipt binds batch hash and size")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d); receipt = json.loads(outcome.receipt.path.read_text())
|
|
require(receipt["batch_sha256"] == outcome.batch_sha256 and receipt["batch_size"] == 5, "batch binding mismatch")
|
|
|
|
@case("20 failure leaves consumed receipt and no output")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
expect_failure(lambda: execute(d, scripted=(module.FakeReceiveEvent(module.EVENT_REMOTE_EOF),)))
|
|
files = sorted(path.name for path in Path(d).iterdir())
|
|
require(files == ["synthetic_aa.aa-consumed.json"], "failure evidence mismatch")
|
|
|
|
@case("21 fake event buffer is logically discarded")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, adapter = execute(d); output = json.loads(outcome.output.path.read_text())
|
|
require(adapter.logical_event_buffer_discarded and output["result"]["phase10aa_fake_integration"]["physical_memory_erasure_proven"] is False, "erasure promoted")
|
|
|
|
@case("22 exact identity and device proof stay false")
|
|
def _():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
outcome, _ = execute(d); require(outcome.exact_identity is False and outcome.device_behavior_proven is False, "proof promoted")
|
|
|
|
@case("23 custom adapter is rejected")
|
|
def _():
|
|
class Custom: pass
|
|
clock = module.OfflineFakeClock()
|
|
with tempfile.TemporaryDirectory() as d:
|
|
store = module.OfflineFakeEvidenceStore(Path(d)); expect_failure(lambda: module.run_offline_fake_batch(plan(), Custom(), clock, store))
|
|
|
|
@case("24 fake event validation is bounded")
|
|
def _():
|
|
expect_failure(lambda: module.FakeReceiveEvent(module.EVENT_DATA, b"x", float("inf")))
|
|
|
|
@case("25 no live API or address exists")
|
|
def _():
|
|
names = set(dir(module)); require(not ({"connect", "send", "recv", "main"} & names), "live API exists")
|
|
|
|
failures = []
|
|
for name, function in cases:
|
|
try: function(); print(f"PASS {name}")
|
|
except Exception as error: failures.append(f"{name}: {error}"); print(f"FAIL {name}: {error}")
|
|
if failures: return 1
|
|
print(f"Phase-1.0AA offline fake-batch tests passed: {len(cases)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|