303 lines
10 KiB
Python
303 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Host guardrails for the blocked Phase-0.9B observer audit."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from types import ModuleType
|
|
from typing import Any
|
|
|
|
|
|
def load_module(name: str, path: Path) -> ModuleType:
|
|
spec = importlib.util.spec_from_file_location(name, 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[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise RuntimeError(message)
|
|
|
|
|
|
def require_invalid(errors: list[str], scenario: str) -> None:
|
|
require(bool(errors), f"unsafe Phase-0.9B mutation passed: {scenario}")
|
|
|
|
|
|
def synthetic_artifact_findings(
|
|
*,
|
|
imports: list[str],
|
|
strings: list[str],
|
|
open_flags: list[str],
|
|
reachable_calls: list[str],
|
|
) -> list[str]:
|
|
denied = (
|
|
"kernel_copyin",
|
|
"kernel_copyout",
|
|
"kernel_set_ucred",
|
|
"ptrace",
|
|
"dlopen",
|
|
"dlsym",
|
|
"sceKernelLoadStartModule",
|
|
"socket",
|
|
"connect",
|
|
"bind",
|
|
"listen",
|
|
"accept",
|
|
"kill",
|
|
"exec",
|
|
"spawn",
|
|
"rename",
|
|
"unlink",
|
|
"chmod",
|
|
"chown",
|
|
"mkdir",
|
|
"mount",
|
|
"reboot",
|
|
"videoout",
|
|
"gnm",
|
|
"sdl",
|
|
"retroarch",
|
|
)
|
|
findings: list[str] = []
|
|
for category, values in (
|
|
("import", imports),
|
|
("string", strings),
|
|
("reachable", reachable_calls),
|
|
):
|
|
for value in values:
|
|
lowered = value.lower()
|
|
if any(token.lower() in lowered for token in denied):
|
|
findings.append(f"{category}:{value}")
|
|
for value in open_flags:
|
|
if value in {"O_WRONLY", "O_RDWR", "O_CREAT", "O_TRUNC", "O_APPEND"}:
|
|
findings.append(f"open_flag:{value}")
|
|
return findings
|
|
|
|
|
|
def test_host_model(model: ModuleType) -> None:
|
|
require(model.evaluate_firmware("9.60", "9.60") == "OBSERVED", "equal firmware")
|
|
require(model.evaluate_firmware("9.60", "9.61") == "CONFLICT", "firmware conflict")
|
|
require(model.evaluate_firmware("9.60", None) == "UNPROVEN", "missing firmware")
|
|
|
|
empty_digest = hashlib.sha256(b"").hexdigest()
|
|
regular = model.MockObject(expected_sha256=empty_digest)
|
|
require(model.evaluate_object(regular) == ("OBSERVED", None), "regular object")
|
|
require(
|
|
model.evaluate_object(model.MockObject(symlink=True))[1]
|
|
== "PATH_SYMLINK_SAFETY_UNPROVEN",
|
|
"symlink",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(object_id_after="dev:1/ino:2"))[1]
|
|
== "OBJECT_ID_CHANGED",
|
|
"object ID change",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(size_after=1))[1] == "SIZE_CHANGED",
|
|
"size change",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(size_before=1, size_after=1))[1]
|
|
== "SHORT_READ",
|
|
"short read",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(read_error=True))[1] == "READ_ERROR",
|
|
"read error",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(expected_sha256="0" * 64))[1]
|
|
== "HASH_MISMATCH",
|
|
"hash mismatch",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(path_known=False))[1] == "UNKNOWN_PATH",
|
|
"unknown path",
|
|
)
|
|
require(
|
|
model.evaluate_object(model.MockObject(path_conflict=True))[1] == "PATH_CONFLICT",
|
|
"path conflict",
|
|
)
|
|
|
|
backup = model.MockObject(object_id_before="dev:1/ino:2")
|
|
require(model.evaluate_live_backup(regular, backup) == "SEPARATE_OBJECTS", "backup")
|
|
require(
|
|
model.evaluate_live_backup(regular, regular) == "SAME_OBJECT",
|
|
"same live and backup object",
|
|
)
|
|
require(
|
|
model.evaluate_live_backup(regular, None) == "BACKUP_MISSING",
|
|
"backup missing",
|
|
)
|
|
|
|
for category in ("mount", "process", "listener"):
|
|
require(
|
|
model.unsupported_query(False) == "UNSUPPORTED_OR_UNPROVEN",
|
|
f"unsupported {category} query",
|
|
)
|
|
require(model.evaluate_autoload(False, False) == "UNPROVEN", "autoload missing")
|
|
require(model.evaluate_autoload(True, False) == "ERROR", "autoload parse")
|
|
|
|
output_limited = model.run_terminal_scenario(record_count=100)
|
|
require(len(output_limited.emitted) == 64, "output limit")
|
|
require(output_limited.exit_reached, "exit after output limit")
|
|
|
|
error_limited = model.run_terminal_scenario(error_count=20)
|
|
require(error_limited.errors == 9, "error limit")
|
|
require(error_limited.exit_reached, "exit after error limit")
|
|
|
|
deadline = model.run_terminal_scenario(deadline_reached=True)
|
|
require(deadline.emitted[0]["raw_error"] == "DEADLINE_REACHED", "deadline")
|
|
require(deadline.exit_reached, "exit after deadline")
|
|
|
|
output_failure = model.run_terminal_scenario(output_channel_ok=False)
|
|
require(
|
|
output_failure.emitted[0]["raw_error"] == "OUTPUT_CHANNEL_FAILED",
|
|
"output failure",
|
|
)
|
|
require(output_failure.exit_reached, "exit after output failure")
|
|
|
|
for observer in (output_limited, error_limited, deadline, output_failure):
|
|
require(observer.retry_count == 0, "retry occurred")
|
|
require(observer.persistent_write_count == 0, "persistent write occurred")
|
|
require(
|
|
observer.service_or_process_mutation_count == 0,
|
|
"service/process mutation occurred",
|
|
)
|
|
require(observer.listener_count == 0, "listener occurred")
|
|
require(observer.lifecycle_call_count == 0, "lifecycle call occurred")
|
|
require(observer.installer_call_count == 0, "installer call occurred")
|
|
require(
|
|
observer.graphics_or_retroarch_call_count == 0,
|
|
"graphics/RetroArch call occurred",
|
|
)
|
|
|
|
|
|
def test_negative_policy(
|
|
validator: ModuleType, root: Path, manifest: dict[str, Any]
|
|
) -> None:
|
|
ready = copy.deepcopy(manifest)
|
|
ready["status"] = "READY"
|
|
require_invalid(validator.validate_manifest(root, ready), "READY status")
|
|
|
|
for field in validator.AUTHORIZATION_FIELDS:
|
|
changed = copy.deepcopy(manifest)
|
|
changed["authorization"][field] = True
|
|
require_invalid(
|
|
validator.validate_manifest(root, changed), f"authorization {field}=true"
|
|
)
|
|
|
|
built = copy.deepcopy(manifest)
|
|
built["build_gate"]["observer_source_created"] = True
|
|
built["build_gate"]["observer_target_declared"] = True
|
|
built["build_gate"]["target_build_performed"] = True
|
|
built["artifact"]["present"] = True
|
|
built["artifact"]["path"] = "observer.elf"
|
|
built["artifact"]["sha256"] = "1" * 64
|
|
built["artifact"]["size"] = 1
|
|
require_invalid(validator.validate_manifest(root, built), "artifact appeared")
|
|
|
|
startup_claim = copy.deepcopy(manifest)
|
|
startup_claim["build_gate"]["startup_and_exit_abi_proven"] = True
|
|
require_invalid(validator.validate_manifest(root, startup_claim), "startup proof")
|
|
|
|
output_claim = copy.deepcopy(manifest)
|
|
output_claim["build_gate"]["non_persistent_output_channel_proven"] = True
|
|
require_invalid(validator.validate_manifest(root, output_claim), "output proof")
|
|
|
|
implementation = copy.deepcopy(manifest)
|
|
implementation["implementation"]["observer_logic_implemented"] = True
|
|
implementation["implementation"]["observations_implemented"] = ["firmware"]
|
|
require_invalid(
|
|
validator.validate_manifest(root, implementation), "target implementation"
|
|
)
|
|
|
|
reproducible = copy.deepcopy(manifest)
|
|
reproducible["reproducibility"]["status"] = "PASSED"
|
|
reproducible["reproducibility"]["build_1_sha256"] = "1" * 64
|
|
reproducible["reproducibility"]["build_2_sha256"] = "1" * 64
|
|
reproducible["reproducibility"]["byte_identical"] = True
|
|
require_invalid(
|
|
validator.validate_manifest(root, reproducible),
|
|
"unperformed build became reproducible",
|
|
)
|
|
|
|
|
|
def test_artifact_audit_guardrails(manifest: dict[str, Any]) -> None:
|
|
require(manifest["artifact"]["present"] is False, "blocked artifact exists")
|
|
require(manifest["artifact"]["path"] is None, "blocked artifact path exists")
|
|
require(
|
|
manifest["static_artifact_audit"]["status"]
|
|
== "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
|
|
"missing-artifact audit was promoted",
|
|
)
|
|
|
|
findings = synthetic_artifact_findings(
|
|
imports=["kernel_copyin", "bind", "sceKernelLoadStartModule"],
|
|
strings=["/autoload_status", "RetroArch", "VideoOut"],
|
|
open_flags=["O_RDONLY", "O_WRONLY", "O_CREAT"],
|
|
reachable_calls=["_start->ptrace", "observer->rename", "observer->kill"],
|
|
)
|
|
require(len(findings) >= 10, "denied-capability scanner missed synthetic cases")
|
|
require(
|
|
synthetic_artifact_findings(
|
|
imports=[],
|
|
strings=[],
|
|
open_flags=["O_RDONLY", "O_NOFOLLOW", "O_CLOEXEC"],
|
|
reachable_calls=[],
|
|
)
|
|
== [],
|
|
"read-only synthetic audit produced a false positive",
|
|
)
|
|
|
|
|
|
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(
|
|
"phase09b_observer_model", root / "tests/phase09b_observer_model.py"
|
|
)
|
|
validator = load_module(
|
|
"phase09b_validator", root / "tools/validate_phase09b_observer_audit.py"
|
|
)
|
|
manifest = validator.load_json(
|
|
root / "manifests/runtime/phase-0.9b-observer.json"
|
|
)
|
|
|
|
errors = validator.collect_errors(root)
|
|
if errors:
|
|
raise RuntimeError("; ".join(errors))
|
|
test_host_model(model)
|
|
test_negative_policy(validator, root, manifest)
|
|
test_artifact_audit_guardrails(manifest)
|
|
|
|
schema = validator.load_json(
|
|
root / "manifests/runtime/phase-0.9b-observation-plan.schema.json"
|
|
)
|
|
default = schema["x-chimera-default-plan"]
|
|
require(default["read_paths"] == [], "default plan contains paths")
|
|
require(default["device_address"] is None, "default plan contains address")
|
|
require(default["maximum_execution_count"] == 0, "default plan permits execution")
|
|
for field in validator.AUTHORIZATION_FIELDS:
|
|
require(default[field] is False, f"default plan {field} is not false")
|
|
|
|
print("Phase-0.9B observer host and artifact guardrails: PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|