363 lines
11 KiB
Python
363 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Deterministic, networkless tests for the Phase-0.9C host protocol model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
from types import ModuleType
|
|
from typing import Callable
|
|
|
|
|
|
def load_module(path: Path) -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location("phase09c_model", path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"could not load {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise RuntimeError(message)
|
|
|
|
|
|
def request(
|
|
model: ModuleType,
|
|
*,
|
|
firmware_one: str | None = "9.60",
|
|
firmware_two: str | None = "9.60",
|
|
requested: int = 0b11,
|
|
) -> object:
|
|
return model.ResultRequest(
|
|
execution_nonce=b"N" * 16,
|
|
request_id=b"R" * 16,
|
|
firmware_source_one=firmware_one,
|
|
firmware_source_two=firmware_two,
|
|
observer_version=1,
|
|
requested_capabilities=requested,
|
|
artifact_sha256=bytes.fromhex("11" * 32),
|
|
deadline_monotonic_ns=10_000,
|
|
)
|
|
|
|
|
|
def mutate(record: bytes, offset: int, value: int) -> bytes:
|
|
changed = bytearray(record)
|
|
changed[offset] ^= value
|
|
return bytes(changed)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
root = args.root.resolve()
|
|
model = load_module(root / "tests/phase09c_feasibility_model.py")
|
|
cases: list[tuple[str, Callable[[], None]]] = []
|
|
|
|
def case(name: str) -> Callable[[Callable[[], None]], Callable[[], None]]:
|
|
def register(function: Callable[[], None]) -> Callable[[], None]:
|
|
cases.append((name, function))
|
|
return function
|
|
|
|
return register
|
|
|
|
@case("startup callgraph")
|
|
def _() -> None:
|
|
require(model.validate_startup_model() == [], "startup model invalid")
|
|
|
|
@case("prohibited startup effects")
|
|
def _() -> None:
|
|
require(
|
|
model.PROHIBITED_STARTUP_EFFECTS
|
|
<= model.NORMAL_SDK_REACHABLE_PROHIBITED_EFFECTS,
|
|
"normal SDK effects hidden",
|
|
)
|
|
|
|
@case("freestanding dependency closure")
|
|
def _() -> None:
|
|
blockers = set(model.freestanding_blockers())
|
|
require(
|
|
{
|
|
"stack_alignment",
|
|
"callable_read_abi",
|
|
"callable_monotonic_time_abi",
|
|
"process_exit_abi",
|
|
"return_cleanup",
|
|
"bounded_result_copyout",
|
|
}
|
|
<= blockers,
|
|
"freestanding blockers missing",
|
|
)
|
|
|
|
@case("exit state machine")
|
|
def _() -> None:
|
|
for path in model.EXIT_PATHS:
|
|
require(not model.exit_path_is_safe(path), f"{path} became safe")
|
|
require("SIGKILL" in model.EXIT_PATHS["timeout"], "timeout kill hidden")
|
|
|
|
@case("output framing")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"ok", observed_capabilities=0b11
|
|
)
|
|
require(len(record) == 4096, "record is not fixed-size")
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "VALID_COMPLETE_RESULT",
|
|
"valid result rejected",
|
|
)
|
|
|
|
@case("maximum lengths")
|
|
def _() -> None:
|
|
expected = request(model, requested=0)
|
|
record = model.build_result(expected, b"x" * model.MAX_BODY_SIZE)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "VALID_COMPLETE_RESULT",
|
|
"maximum body rejected",
|
|
)
|
|
try:
|
|
model.build_result(expected, b"x" * (model.MAX_BODY_SIZE + 1))
|
|
except ValueError:
|
|
return
|
|
raise RuntimeError("oversized body accepted")
|
|
|
|
@case("truncation")
|
|
def _() -> None:
|
|
expected = request(model, requested=0)
|
|
record = model.build_result(
|
|
expected, b"x" * (model.MAX_BODY_SIZE + 1), truncate=True
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_TRUNCATED",
|
|
"truncation accepted",
|
|
)
|
|
|
|
@case("stale nonce")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11
|
|
)
|
|
stale = model.ResultRequest(
|
|
execution_nonce=b"S" * 16,
|
|
request_id=expected.request_id,
|
|
firmware_source_one="9.60",
|
|
firmware_source_two="9.60",
|
|
observer_version=1,
|
|
requested_capabilities=0b11,
|
|
artifact_sha256=expected.artifact_sha256,
|
|
deadline_monotonic_ns=10_000,
|
|
)
|
|
require(
|
|
model.validate_result(record, stale, now_monotonic_ns=1)
|
|
== "BLOCKED_STALE_NONCE",
|
|
"stale nonce accepted",
|
|
)
|
|
|
|
@case("duplicate result")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11
|
|
)
|
|
consumer = model.ResultConsumer()
|
|
require(
|
|
consumer.consume(record, expected, now_monotonic_ns=1)
|
|
== "VALID_COMPLETE_RESULT",
|
|
"first result rejected",
|
|
)
|
|
require(
|
|
consumer.consume(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_DUPLICATE_RESULT",
|
|
"duplicate accepted",
|
|
)
|
|
|
|
@case("result checksum failure")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"body", observed_capabilities=0b11
|
|
)
|
|
require(
|
|
model.validate_result(
|
|
mutate(record, model.HEADER_SIZE, 1),
|
|
expected,
|
|
now_monotonic_ns=1,
|
|
)
|
|
== "BLOCKED_RESULT_CHECKSUM",
|
|
"checksum corruption accepted",
|
|
)
|
|
|
|
@case("timeout")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=10_001)
|
|
== "BLOCKED_TIMEOUT",
|
|
"expired result accepted",
|
|
)
|
|
|
|
@case("incomplete completion marker")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11, complete=False
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_INCOMPLETE",
|
|
"incomplete record accepted",
|
|
)
|
|
|
|
@case("unsupported capability")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected,
|
|
b"",
|
|
observed_capabilities=0b01,
|
|
unsupported_capabilities=0b10,
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "VALID_RECORD_WITH_UNSUPPORTED_CAPABILITIES",
|
|
"explicit unsupported result lost",
|
|
)
|
|
|
|
@case("error versus empty success")
|
|
def _() -> None:
|
|
expected = request(model, requested=0)
|
|
error = model.build_result(
|
|
expected, b"", status=model.STATUS_OBSERVER_ERROR
|
|
)
|
|
empty = model.build_result(expected, b"")
|
|
require(
|
|
model.validate_result(error, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_OBSERVER_FAILURE",
|
|
"empty failure became success",
|
|
)
|
|
require(
|
|
model.validate_result(empty, expected, now_monotonic_ns=1)
|
|
== "VALID_COMPLETE_RESULT",
|
|
"valid no-capability empty result rejected",
|
|
)
|
|
|
|
@case("cleanup status")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected,
|
|
b"",
|
|
observed_capabilities=0b11,
|
|
cleanup_status=model.CLEANUP_FAILED,
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_CLEANUP_NOT_PROVEN",
|
|
"failed cleanup accepted",
|
|
)
|
|
|
|
@case("conflicting firmware sources")
|
|
def _() -> None:
|
|
expected = request(
|
|
model, firmware_one="9.60", firmware_two="9.61"
|
|
)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_FIRMWARE_CONFLICT",
|
|
"firmware conflict accepted",
|
|
)
|
|
|
|
@case("absent firmware source two")
|
|
def _() -> None:
|
|
expected = request(model, firmware_two=None)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b11
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_FIRMWARE_SOURCE_2_ABSENT",
|
|
"missing firmware source accepted",
|
|
)
|
|
|
|
@case("unknown protocol version")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected,
|
|
b"",
|
|
observed_capabilities=0b11,
|
|
protocol_version=2,
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_UNKNOWN_VERSION",
|
|
"unknown version accepted",
|
|
)
|
|
|
|
@case("capability completeness")
|
|
def _() -> None:
|
|
expected = request(model)
|
|
record = model.build_result(
|
|
expected, b"", observed_capabilities=0b01
|
|
)
|
|
require(
|
|
model.validate_result(record, expected, now_monotonic_ns=1)
|
|
== "BLOCKED_INCOMPLETE_CAPABILITY_RESULT",
|
|
"partial capability bitmap accepted",
|
|
)
|
|
|
|
@case("side-effect classification")
|
|
def _() -> None:
|
|
effects = model.classify_side_effects("filesystem_content_hash")
|
|
require("semantic_readonly" in effects, "semantic read missing")
|
|
require("atime_effect_possible" in effects, "atime risk hidden")
|
|
require("audit_effect_possible" in effects, "audit risk hidden")
|
|
require("cache_effect_possible" in effects, "cache risk hidden")
|
|
require("object_race_possible" in effects, "race risk hidden")
|
|
require(
|
|
not model.is_proven_side_effect_free("filesystem_content_hash"),
|
|
"read was promoted to side-effect-free",
|
|
)
|
|
|
|
@case("output architecture ordering")
|
|
def _() -> None:
|
|
require(
|
|
list(model.OUTPUT_ARCHITECTURES)
|
|
== [
|
|
"D1_CALLER_OWNED_BOUNDED_BUFFER",
|
|
"D2_EXISTING_REQUEST_RESPONSE",
|
|
"D3_LOADER_OWNED_STATUS_RECORD",
|
|
"D4_PROCESS_EXIT_STATUS",
|
|
],
|
|
"output architectures reordered",
|
|
)
|
|
|
|
for name, function in cases:
|
|
try:
|
|
function()
|
|
except Exception as error:
|
|
raise RuntimeError(f"Phase-0.9C protocol case failed: {name}: {error}") from error
|
|
|
|
print(f"Phase-0.9C protocol host tests: {len(cases)}/{len(cases)} PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|