269 lines
9.9 KiB
Python
269 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Validate the inactive Phase-1.0V one-shot collector model."""
|
|
|
|
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_0V_INACTIVE_ONE_SHOT_SHSRV_COLLECTOR"
|
|
STATUS = "INACTIVE_OFFLINE_COLLECTOR_MODEL_COMPLETE_NO_LIVE_CLIENT"
|
|
START_COMMIT = "50e042a96cd75925aa39a579c6242f3613caca03"
|
|
MODEL_SIZE = 7429
|
|
MODEL_SHA256 = "f8a306dafee5d135919bec5afda789dd741e57f39803b7683fb8747c186db25c"
|
|
PARSER_SHA256 = "4701a057a98b4874e49e1bcf11db9a9a3a105e48f2c25e42796bff10f238f7c2"
|
|
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",
|
|
)
|
|
|
|
|
|
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.0V 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
|
|
|
|
|
|
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,
|
|
"target_address": None,
|
|
"target_port": None,
|
|
"window": None,
|
|
"exact_literal_path": None,
|
|
"commands": [],
|
|
"deadline_seconds": None,
|
|
"run_id": None,
|
|
"expires_at": None,
|
|
}
|
|
|
|
|
|
def model_is_offline(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"relative_path": "tools/phase10v_shsrv_collector_model.py",
|
|
"size": MODEL_SIZE,
|
|
"sha256": MODEL_SHA256,
|
|
"phase10t_parser_sha256": PARSER_SHA256,
|
|
"offline_only": True,
|
|
"stdin_only": True,
|
|
"network_transport_present": False,
|
|
"socket_import_present": False,
|
|
"address_argument_present": False,
|
|
"port_argument_present": False,
|
|
"command_argument_present": False,
|
|
"command_send_present": False,
|
|
"filesystem_output_present": False,
|
|
}
|
|
|
|
|
|
def bounds_are_fail_closed(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"max_raw_bytes": 65536,
|
|
"max_sanitized_bytes": 65536,
|
|
"max_chunks": 256,
|
|
"stdin_chunk_bytes": 4096,
|
|
"strict_utf8": True,
|
|
"one_shot": True,
|
|
"automatic_retry": False,
|
|
"reconnect": False,
|
|
"resume": False,
|
|
"fallback": False,
|
|
}
|
|
|
|
|
|
def telnet_is_offline_partial(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"incremental": True,
|
|
"fragmented_negotiation_supported": True,
|
|
"subnegotiation_removed": True,
|
|
"incomplete_control_rejected": True,
|
|
"negotiation_replies_sent": False,
|
|
"live_prompt_contract_proven": False,
|
|
}
|
|
|
|
|
|
def sanitization_is_strict(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"raw_transcript_output": False,
|
|
"serial_value_output": False,
|
|
"model_value_output": False,
|
|
"temperature_value_output": False,
|
|
"cpu_frequency_value_output": False,
|
|
"unknown_path_output": False,
|
|
"approved_literal_path_metadata_output": True,
|
|
"exact_identity_output": False,
|
|
"logical_buffer_discard": True,
|
|
"physical_memory_erasure_proven": False,
|
|
}
|
|
|
|
|
|
def review_remediation_is_complete(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"telnet_doubled_iac_state_fixed": True,
|
|
"empty_chunks_ignored": True,
|
|
"expected_paths_absolute_normalized_and_character_allowlisted": True,
|
|
"firmware_metadata_gated_to_exact_9_60": True,
|
|
"compile_metadata_format_validated": True,
|
|
"parser_numeric_failure_normalized": True,
|
|
"physical_memory_erasure_proven": False,
|
|
}
|
|
|
|
|
|
def decision_requires_review(record: dict[str, Any]) -> bool:
|
|
return record == {
|
|
"offline_collector_model_complete": True,
|
|
"exact_deployed_shsrv_identity": "UNPROVEN",
|
|
"live_network_client_created": False,
|
|
"live_client_implementation_allowed": False,
|
|
"live_collection_allowed": False,
|
|
"launch_context_experiment_allowed": False,
|
|
"device_action_allowed": False,
|
|
"next_step": "HUMAN_REVIEW_OF_OFFLINE_COLLECTOR_MODEL",
|
|
}
|
|
|
|
|
|
def source_has_no_network_capability(path: Path) -> bool:
|
|
source = path.read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
imports = {
|
|
alias.name.split(".", 1)[0]
|
|
for node in ast.walk(tree)
|
|
if isinstance(node, (ast.Import, ast.ImportFrom))
|
|
for alias in node.names
|
|
}
|
|
forbidden_args = (
|
|
'"--target"', '"--host"', '"--address"', '"--port"',
|
|
'"--command"', '"--connect"', '"--send"',
|
|
)
|
|
return (
|
|
not imports.intersection({
|
|
"socket", "telnetlib", "urllib", "requests", "http", "ftplib"})
|
|
and all(token not in source for token in forbidden_args)
|
|
and '"--offline-transcript"' in source
|
|
and '"--expected-path"' in source
|
|
and "sys.stdin.buffer.read(4096)" in source
|
|
and "open(" not in source
|
|
and "write_text(" not in source
|
|
and "write_bytes(" not in source
|
|
)
|
|
|
|
|
|
def exact_file(path: Path, size: int, digest: str) -> bool:
|
|
return path.stat().st_size == size and sha256(path) == digest
|
|
|
|
|
|
def validate(root: Path) -> list[str]:
|
|
errors: list[str] = []
|
|
try:
|
|
record = load_json(
|
|
root / "manifests/retroarch/phase-1.0v-inactive-shsrv-collector.json")
|
|
if record.get("phase") != PHASE or record.get("status") != STATUS:
|
|
errors.append("phase/status mismatch")
|
|
if record.get("start_commit") != START_COMMIT:
|
|
errors.append("start commit mismatch")
|
|
if not activation_is_inactive(record.get("activation", {})):
|
|
errors.append("activation record is not inert")
|
|
if not model_is_offline(record.get("model", {})):
|
|
errors.append("model identity or offline boundary mismatch")
|
|
if not bounds_are_fail_closed(record.get("bounds", {})):
|
|
errors.append("collector bounds were relaxed")
|
|
if not telnet_is_offline_partial(record.get("telnet_model", {})):
|
|
errors.append("Telnet model was promoted to a live contract")
|
|
if not sanitization_is_strict(record.get("sanitization", {})):
|
|
errors.append("sanitization or memory claim was relaxed")
|
|
if not review_remediation_is_complete(
|
|
record.get("review_remediation", {})):
|
|
errors.append("self-review remediation is incomplete")
|
|
if not all_authorizations_false(record.get("authorizations", {})):
|
|
errors.append("authorization remains active or missing")
|
|
if not decision_requires_review(record.get("decision", {})):
|
|
errors.append("live implementation or device decision is enabled")
|
|
performed = record.get("performed_actions", {})
|
|
if not performed or not all(value is False for value in performed.values()):
|
|
errors.append("performed action is present")
|
|
tests = record.get("tests", {})
|
|
if not (
|
|
tests.get("chimera_gfx_ctest") == "77_OF_77_PASS"
|
|
and tests.get("phase10v_guardrails") == 18
|
|
and tests.get("phase10v_model_tests") == 21
|
|
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")
|
|
|
|
model_path = root / "tools/phase10v_shsrv_collector_model.py"
|
|
parser_path = root / "tools/phase10t_shsrv_transcript.py"
|
|
if not exact_file(model_path, MODEL_SIZE, MODEL_SHA256):
|
|
errors.append("collector model identity mismatch")
|
|
if sha256(parser_path) != PARSER_SHA256:
|
|
errors.append("Phase-1.0T parser identity mismatch")
|
|
if not source_has_no_network_capability(model_path):
|
|
errors.append("collector model exposes network, persistence or live arguments")
|
|
approval = (
|
|
root / "docs/approvals/phase-1.0v-shsrv-collector.md").read_text(
|
|
encoding="utf-8")
|
|
for token in (
|
|
"active=false", "ps5_connection_authorized=false",
|
|
"device_request_authorized=false", "automatic_retry=false",
|
|
"target_address=null", "target_port=null", "commands=[]",
|
|
):
|
|
if token not in approval:
|
|
errors.append(f"inactive approval token missing: {token}")
|
|
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)
|
|
args = parser.parse_args()
|
|
errors = validate(args.root.resolve())
|
|
if errors:
|
|
for error in errors:
|
|
print(f"ERROR: {error}")
|
|
return 1
|
|
print("Phase-1.0V inactive collector-model validation passed")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|