270 lines
11 KiB
Python
270 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Host-only policy tests for the inactive Phase-1.0W architecture."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
|
|
def load(path: Path):
|
|
spec = importlib.util.spec_from_file_location("phase10w_policy", 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"))
|
|
sys.path.insert(0, str(root / "tests"))
|
|
module = load(root / "tools/phase10w_shsrv_client_policy.py")
|
|
from phase10w_fake_transport import FakeTransport, FakeTransportError
|
|
now = datetime(2026, 7, 22, 12, 0, tzinfo=timezone.utc)
|
|
|
|
def records(window="T2_GREETING_AND_HELP"):
|
|
path = None if window == "T2_GREETING_AND_HELP" else "/data/exact.elf"
|
|
commands = list(module.WINDOW_COMMANDS[window])
|
|
common = {
|
|
"active": True,
|
|
"policy_sha256": "a" * 64,
|
|
"collector_sha256": module.COLLECTOR_SHA256,
|
|
"run_id": "synthetic_run_001",
|
|
"target_address": "device.invalid",
|
|
"target_port": 2323,
|
|
"window": window,
|
|
"exact_literal_path": path,
|
|
"commands": commands,
|
|
"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 common, approval
|
|
|
|
cases = []
|
|
|
|
def case(name):
|
|
def register(function):
|
|
cases.append((name, function))
|
|
return function
|
|
return register
|
|
|
|
@case("01 tracked inactive shape is inert")
|
|
def _(): require(module.inactive_record_is_inert({"active": False, "policy_sha256": None, "collector_sha256": None, "run_id": None, "target_address": None, "target_port": None, "window": None, "exact_literal_path": None, "commands": [], "deadline_seconds": None, "expires_at": None}), "inactive shape rejected")
|
|
|
|
@case("02 exact help window produces immutable plan")
|
|
def _():
|
|
activation, approval = records(); plan = module.build_session_plan(activation, approval, now)
|
|
require(plan.commands == ("help",) and plan.target_port == 2323, "help plan mismatch")
|
|
|
|
@case("03 exact-path window validates path")
|
|
def _():
|
|
activation, approval = records("T3_ONE_EXACT_PATH"); plan = module.build_session_plan(activation, approval, now)
|
|
require(plan.exact_literal_path == "/data/exact.elf", "path lost")
|
|
|
|
@case("04 inactive record is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["active"] = False
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("inactive activation accepted")
|
|
|
|
@case("05 missing attestation is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["attested"] = False
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("missing attestation accepted")
|
|
|
|
@case("06 record mismatch is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["run_id"] = "different_run"
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("mismatch accepted")
|
|
|
|
@case("07 collector hash mismatch is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["collector_sha256"] = approval["collector_sha256"] = "0" * 64
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("wrong hash accepted")
|
|
|
|
@case("08 retry is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["automatic_retry"] = True
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("retry accepted")
|
|
|
|
@case("09 execution authority is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["device_execution_authorized"] = True
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("execution authority accepted")
|
|
|
|
@case("10 missing side-effect acceptance is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["automatic_serial_query_accepted"] = False
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("serial effect hidden")
|
|
|
|
@case("11 command injection is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["commands"] = approval["commands"] = ["help; hbldr"]
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("command injection accepted")
|
|
|
|
@case("12 target syntax injection is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["target_address"] = approval["target_address"] = "device.invalid\nother"
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("target injection accepted")
|
|
|
|
@case("13 wrong port is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["target_port"] = approval["target_port"] = 9999
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("wrong port accepted")
|
|
|
|
@case("14 unsafe path is rejected")
|
|
def _():
|
|
activation, approval = records("T3_ONE_EXACT_PATH"); activation["exact_literal_path"] = approval["exact_literal_path"] = "/data/../escape"
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except (module.PolicyError, RuntimeError): return
|
|
raise RuntimeError("unsafe path accepted")
|
|
|
|
@case("15 excessive deadline is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["deadline_seconds"] = approval["deadline_seconds"] = 11
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("deadline relaxation accepted")
|
|
|
|
@case("16 expired approval is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["expires_at"] = approval["expires_at"] = "2026-07-22T11:59:00Z"
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("expired approval accepted")
|
|
|
|
@case("17 overlong approval lifetime is rejected")
|
|
def _():
|
|
activation, approval = records(); activation["expires_at"] = approval["expires_at"] = "2026-07-22T12:16:00Z"
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("long approval accepted")
|
|
|
|
@case("18 policy plan exposes no transport method")
|
|
def _():
|
|
activation, approval = records(); plan = module.build_session_plan(activation, approval, now)
|
|
require(not any(hasattr(plan, name) for name in ("connect", "send", "recv", "open")), "transport method present")
|
|
|
|
@case("18b unknown approval field is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["unexpected_authority"] = True
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("unknown approval field accepted")
|
|
|
|
@case("18c missing listener attestation is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["listener_already_running_attested"] = False
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("missing listener attestation accepted")
|
|
|
|
@case("18d policy hash mismatch is rejected")
|
|
def _():
|
|
activation, approval = records(); approval["policy_sha256"] = "b" * 64
|
|
try: module.build_session_plan(activation, approval, now)
|
|
except module.PolicyError: return
|
|
raise RuntimeError("policy hash mismatch accepted")
|
|
|
|
@case("19 fake transport models exactly one session")
|
|
def _():
|
|
activation, approval = records(); plan = module.build_session_plan(activation, approval, now)
|
|
transport = FakeTransport([b"synthetic"]); transport.open_once(plan)
|
|
transport.send_command_token("help"); require(transport.receive_chunk() == b"synthetic", "fake input lost")
|
|
require(transport.receive_chunk() is None, "fake EOF missing"); transport.close_once()
|
|
require(transport.events == ["OPEN", "COMMAND_HELP", "RECEIVE", "CLOSE"], "event sequence mismatch")
|
|
|
|
@case("20 fake second open is rejected")
|
|
def _():
|
|
activation, approval = records(); plan = module.build_session_plan(activation, approval, now)
|
|
transport = FakeTransport([]); transport.open_once(plan)
|
|
try: transport.open_once(plan)
|
|
except FakeTransportError: return
|
|
raise RuntimeError("second fake open accepted")
|
|
|
|
@case("21 fake unexpected command is rejected")
|
|
def _():
|
|
activation, approval = records(); plan = module.build_session_plan(activation, approval, now)
|
|
transport = FakeTransport([]); transport.open_once(plan)
|
|
try: transport.send_command_token("hbldr")
|
|
except FakeTransportError: return
|
|
raise RuntimeError("unexpected fake command accepted")
|
|
|
|
@case("22 fake transport has no retry or reconnect API")
|
|
def _():
|
|
transport = FakeTransport([])
|
|
require(not any(hasattr(transport, name) for name in ("retry", "reconnect", "resume")), "retry API present")
|
|
|
|
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.0W client-policy tests passed: {len(cases)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|