292 lines
12 KiB
Python
292 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate Phase-1.0X inactive injected-transport evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
|
|
PHASE = "PHASE_1_0X_INACTIVE_INJECTED_TRANSPORT"
|
|
STATUS = "INACTIVE_INJECTED_TRANSPORT_COMPLETE_LIVE_PROTOCOL_BLOCKED"
|
|
START_COMMIT = "45228225e3a0e8811b4a5bcaa79e6f125db3a7f8"
|
|
SHSRV_COMMIT = "6f320637d56d344a0e7797753099e33238bbf146"
|
|
PARSER_SHA256 = "4701a057a98b4874e49e1bcf11db9a9a3a105e48f2c25e42796bff10f238f7c2"
|
|
COLLECTOR_SIZE = 7429
|
|
COLLECTOR_SHA256 = "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c"
|
|
POLICY_SIZE = 6997
|
|
POLICY_SHA256 = "747d23c88f2722e8e8846599c3ac1dae3826eb3fca881caaad36b251f30f3592"
|
|
TRANSPORT_SIZE = 8040
|
|
TRANSPORT_SHA256 = "568d7578482ecf2fcd9e29085b2eb9d8705fc699611508acdf22afd30f2ddd23"
|
|
TRANSPORT_TEST_SIZE = 13087
|
|
TRANSPORT_TEST_SHA256 = "19709ed6ab456be428d262b3f0afb4f6f577b34e4db80e66a73ed63d9bf2cd43"
|
|
AUTHORIZATION_FIELDS = (
|
|
"target_build_authorized", "ps5_connection_authorized",
|
|
"device_request_authorized", "result_receive_authorized",
|
|
"device_transfer_authorized", "device_execution_authorized",
|
|
"installation_authorized", "autoload_authorized",
|
|
"device_write_authorized", "automatic_retry", "reconnect_authorized",
|
|
)
|
|
NETWORK_MODULES = {
|
|
"socket", "telnetlib", "urllib", "requests", "http", "ftplib",
|
|
"asyncio", "selectors",
|
|
}
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError("Phase-1.0X manifest is not an object")
|
|
return value
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def git(root: Path, *args: str) -> str:
|
|
result = subprocess.run(
|
|
["git", *args], cwd=root, capture_output=True, text=True, check=False)
|
|
if result.returncode:
|
|
raise RuntimeError(result.stderr.strip() or "git failed")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def exact_file(path: Path, size: int, digest: str) -> bool:
|
|
return path.stat().st_size == size and sha256(path) == digest
|
|
|
|
|
|
def all_authorizations_false(record: dict[str, Any]) -> bool:
|
|
return all(record.get(field) is False for field in AUTHORIZATION_FIELDS)
|
|
|
|
|
|
def activation_is_inactive(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"active": False, "policy_sha256": None, "collector_sha256": None,
|
|
"transport_sha256": None, "run_id": None, "target_address": None,
|
|
"target_port": None, "window": None, "exact_literal_path": None,
|
|
"commands": [], "deadline_seconds": None, "expires_at": None,
|
|
}
|
|
|
|
|
|
def architecture_is_injected_only(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"injected_adapter_only": True,
|
|
"injected_monotonic_clock_only": True,
|
|
"policy_cli_present": False,
|
|
"live_cli_present": False,
|
|
"network_transport_present": False,
|
|
"socket_import_present": False,
|
|
"dns_lookup_present": False,
|
|
"connect_present": False,
|
|
"target_address_present": False,
|
|
"command_byte_formatter_present": False,
|
|
"telnet_reply_generator_present": False,
|
|
"live_prompt_detector_present": False,
|
|
"fake_boundary_labels_only": True,
|
|
}
|
|
|
|
|
|
def evidence_is_bounded(record: dict[str, Any]) -> bool:
|
|
required_true = {
|
|
"receipt_before_adapter_open", "exclusive_leaf_create", "file_fsync",
|
|
"close_and_reopen", "reopened_bytes_verified",
|
|
"sanitized_output_receipt_hash_binding", "trusted_caller_owned_root_required",
|
|
}
|
|
required_false = {
|
|
"overwrite_supported", "delete_or_cleanup_supported",
|
|
"raw_transcript_persisted", "target_address_persisted",
|
|
"partial_file_cleanup", "partial_file_is_valid_evidence",
|
|
"directory_fsync",
|
|
}
|
|
return (
|
|
set(record) == required_true | required_false |
|
|
{"directory_entry_durability"}
|
|
and all(record.get(field) is True for field in required_true)
|
|
and all(record.get(field) is False for field in required_false)
|
|
and record.get("directory_entry_durability") == "UNPROVEN"
|
|
)
|
|
|
|
|
|
def deadline_is_partial(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"one_absolute_monotonic_deadline": True,
|
|
"pre_and_post_boundary_checks": True,
|
|
"remaining_budget_passed_to_adapter": True,
|
|
"adapter_close_after_open_attempt": True,
|
|
"retry_loop_present": False,
|
|
"second_open_present": False,
|
|
"blocking_adapter_call_preemption": False,
|
|
"real_socket_timeout_present": False,
|
|
"live_cleanup_proven": False,
|
|
}
|
|
|
|
|
|
def decision_is_offline_only(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"offline_injected_transport_complete": True,
|
|
"offline_local_evidence_complete_with_limitations": True,
|
|
"exact_deployed_shsrv_identity": "UNPROVEN",
|
|
"live_protocol_framing": "BLOCKED_UNPROVEN",
|
|
"live_network_client_created": False,
|
|
"live_client_implementation_allowed": False,
|
|
"live_collection_allowed": False,
|
|
"device_action_allowed": False,
|
|
"next_step": "OFFLINE_EXACT_PROMPT_AND_TELNET_FRAMING_AUDIT",
|
|
}
|
|
|
|
|
|
def source_has_no_network_import(path: Path) -> bool:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
imports = {
|
|
alias.name.split(".", 1)[0]
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, (ast.Import, ast.ImportFrom))
|
|
for alias in node.names
|
|
}
|
|
return not imports.intersection(NETWORK_MODULES)
|
|
|
|
|
|
def validate(root: Path, shsrv_root: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
manifest = load_json(
|
|
root / "manifests/retroarch/phase-1.0x-inactive-transport.json")
|
|
if manifest.get("phase") != PHASE or manifest.get("status") != STATUS:
|
|
errors.append("phase/status mismatch")
|
|
if manifest.get("start_commit") != START_COMMIT:
|
|
errors.append("start commit mismatch")
|
|
if not activation_is_inactive(manifest.get("activation", {})):
|
|
errors.append("tracked activation is not inert")
|
|
if not all_authorizations_false(manifest.get("authorizations", {})):
|
|
errors.append("authorization remains active or missing")
|
|
if not architecture_is_injected_only(manifest.get("architecture", {})):
|
|
errors.append("architecture contains live transport capability")
|
|
if not evidence_is_bounded(manifest.get("local_evidence", {})):
|
|
errors.append("local evidence contract mismatch")
|
|
if not deadline_is_partial(manifest.get("deadline_and_cleanup", {})):
|
|
errors.append("deadline limitations were promoted or changed")
|
|
if not decision_is_offline_only(manifest.get("decision", {})):
|
|
errors.append("offline-only decision mismatch")
|
|
performed = manifest.get("performed_actions", {})
|
|
if not performed or not all(value is False for value in performed.values()):
|
|
errors.append("performed device or network action is present")
|
|
missing = manifest.get("missing_live_components", {})
|
|
if len(missing) != 8 or not all(value is True for value in missing.values()):
|
|
errors.append("missing live component was promoted")
|
|
|
|
bindings = manifest.get("source_bindings", {})
|
|
expected_bindings = {
|
|
"shsrv_reference_commit": SHSRV_COMMIT,
|
|
"phase10t_parser_sha256": PARSER_SHA256,
|
|
"phase10v_collector_size": COLLECTOR_SIZE,
|
|
"phase10v_collector_sha256": COLLECTOR_SHA256,
|
|
"phase10w_policy_size": POLICY_SIZE,
|
|
"phase10w_policy_sha256": POLICY_SHA256,
|
|
"phase10x_transport_size": TRANSPORT_SIZE,
|
|
"phase10x_transport_sha256": TRANSPORT_SHA256,
|
|
"phase10x_transport_tests_size": TRANSPORT_TEST_SIZE,
|
|
"phase10x_transport_tests_sha256": TRANSPORT_TEST_SHA256,
|
|
}
|
|
if bindings != expected_bindings:
|
|
errors.append("source bindings mismatch")
|
|
identities = (
|
|
(root / "tools/phase10t_shsrv_transcript.py", None, PARSER_SHA256),
|
|
(root / "tools/phase10v_shsrv_collector_model.py",
|
|
COLLECTOR_SIZE, COLLECTOR_SHA256),
|
|
(root / "tools/phase10w_shsrv_client_policy.py",
|
|
POLICY_SIZE, POLICY_SHA256),
|
|
(root / "tools/phase10x_inactive_transport.py",
|
|
TRANSPORT_SIZE, TRANSPORT_SHA256),
|
|
(root / "tests/test_phase10x_inactive_transport.py",
|
|
TRANSPORT_TEST_SIZE, TRANSPORT_TEST_SHA256),
|
|
)
|
|
for path, size, digest in identities:
|
|
if (size is not None and path.stat().st_size != size) or \
|
|
sha256(path) != digest:
|
|
errors.append(f"source identity mismatch: {path.name}")
|
|
|
|
transport_path = root / "tools/phase10x_inactive_transport.py"
|
|
if not source_has_no_network_import(transport_path):
|
|
errors.append("inactive transport imports networking")
|
|
source = transport_path.read_text(encoding="utf-8")
|
|
for forbidden in ("def main(", "argparse", "target_address", "2323"):
|
|
if forbidden in source:
|
|
errors.append(f"inactive transport contains forbidden token: {forbidden}")
|
|
for required in (
|
|
"os.O_EXCL", "os.fsync", "path.read_bytes()",
|
|
"adapter.close_once()", "receipt = evidence.create_consumed_receipt",
|
|
):
|
|
if required not in source:
|
|
errors.append(f"transport control missing: {required}")
|
|
|
|
if git(shsrv_root, "rev-parse", "HEAD") != SHSRV_COMMIT:
|
|
errors.append("shsrv reference commit mismatch")
|
|
if git(shsrv_root, "status", "--short"):
|
|
errors.append("shsrv reference tree is dirty")
|
|
sh_source = (shsrv_root / "sh.c").read_text(encoding="utf-8")
|
|
for token in (
|
|
"sh_prompt(void)", 'getenv("PWD")', 'setenv("PWD", "/", 0)',
|
|
'fprintf(stdout, "%s$ ", cwd ? cwd : "(null)")', "sh_prompt();",
|
|
):
|
|
if token not in sh_source:
|
|
errors.append(f"shsrv prompt evidence missing: {token}")
|
|
|
|
approval = (
|
|
root / "docs/approvals/phase-1.0x-inactive-transport.md"
|
|
).read_text(encoding="utf-8")
|
|
for token in (
|
|
"active=false", "target_address=null", "target_port=null",
|
|
"commands=[]", "ps5_connection_authorized=false",
|
|
"device_request_authorized=false", "automatic_retry=false",
|
|
):
|
|
if token not in approval:
|
|
errors.append(f"inactive approval token missing: {token}")
|
|
tests = manifest.get("tests", {})
|
|
if not (
|
|
tests.get("chimera_gfx_ctest") == "80_OF_80_PASS"
|
|
and tests.get("phase10x_guardrails") == 18
|
|
and tests.get("phase10x_transport_tests") == 18
|
|
and tests.get("safety_audit") == "PASS"
|
|
and tests.get("secret_scan") == "PASS"
|
|
and tests.get("network_required_by_tests") is False
|
|
and tests.get("hardware_claim_from_host_test") is False
|
|
):
|
|
errors.append("test evidence mismatch")
|
|
tracked = git(root, "ls-files").splitlines()
|
|
if any(path.lower().endswith((".elf", ".self", ".sprx", ".pkg"))
|
|
for path in tracked):
|
|
errors.append("target artifact is tracked")
|
|
except (OSError, RuntimeError, ValueError, json.JSONDecodeError,
|
|
SyntaxError) as error:
|
|
errors.append(str(error))
|
|
return errors
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--root", type=Path, required=True)
|
|
parser.add_argument("--shsrv-root", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
errors = validate(args.root.resolve(), args.shsrv_root.resolve())
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}")
|
|
return 1
|
|
print("Phase-1.0X inactive injected-transport validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|