#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-3.0-or-later """Validate the host-only Phase-0.9E-R2 inner-backup correlation.""" from __future__ import annotations import argparse import hashlib import importlib.util import json from pathlib import Path import re import subprocess import sys from types import ModuleType from typing import Any import zipfile BASELINE = "48714bb542f3ea5b9893d0461ba4c9d938cd6d8d" BRANCH = "codex/chimera-gfx-phase09e-r2-inner-correlation" PHASE = "PHASE_0_9E_R2_INNER_BACKUP_CORRELATION" OUTER_NAME = "Y2JB-Autoloader-403-1240.zip" OUTER_SIZE = 504159435 OUTER_SHA256 = "805e3f87f0c371223619ffc7d3a7b3c0d41a1fae8a8b1171d9e2f162659e8291" INNER_NAME = "PS5/EXPORT/BACKUP/202606102126_00/archive.dat" INNER_SIZE = 504365056 INNER_SHA256 = "6439834e8856d45b6d6fe699b74c35ca6985a199ea8ecf3e398c018d37be2d55" DENYLIST_SHA256 = "e9603b0e3792781ad5b511afb22ef61e3d4fd4c5a16bf928f8609193bcd97783" PRIMARY_CLASSIFICATION = "LOCAL_BACKUP_UNCORRELATED" AUTHORIZATION_FIELDS = ( "device_action_authorized", "target_build_authorized", "transfer_authorized", "execution_authorized", "installation_authorized", "lifecycle_authorized", "autoload_authorized", "device_write_authorized", "rescue_payload_design_authorized", "automatic_retry", ) IMMUTABLE_HASHES = { "manifests/runtime/phase-0.8-read-only-preflight.json": "47d7f452f8799979fe99b3e6d56859f03544112725bf7e5b349eba5ed81b3322", "manifests/runtime/phase-0.8-remediation.json": "a9dafed8c83722c43709dcf90ea117c21d6f996ff8fe233810ed08aa20cbe071", "manifests/runtime/phase-0.9-anti-brick-design.json": "39fd7c70cae998d9d74e7caf1ff3c19f9f76f7de2e5a50bc09baede4ba53e9e9", "manifests/runtime/phase-0.9b-observer.json": "104c4a667ad17f9827fc7276852c6faeef96effb8e3561a5ebc19a62c7d51634", "manifests/runtime/phase-0.9c-feasibility.json": "84eb737ff3486d0c5d8b5ecf06809f93bc573497da4919595b4c1817fd34247c", "manifests/runtime/phase-0.9d-existing-stack-readback.json": "86e5aaf034685dbe058b71ffeec645b682f0a8cc7d249e8ac0397155233991de", "manifests/runtime/phase-0.9e-bootstrap-provenance.json": "5dfa9bfe2ae751b2f0ea0e03c60c1a4471a35452389cf471456e6c54eb4601cf", "manifests/runtime/phase-0.9e-loader-protocol.json": "a7ca8b4e8072cb60ad4cd4869f6c508e8014059c8d81ad3b94a8d9e8cd0db4cb", "manifests/runtime/phase-0.9e-r-release-correlation.json": "09f916b20def51fd519690b6d940de5f93818a4d9bd5eebef462fe2a828b88be", "manifests/runtime/phase-0.9e-r-port9020-audit.json": "ea84f1885ad4286670a401cbf73e9a371cdcd03a1f247908b90ea27553ee820e", } DELIVERABLES = ( "docs/runtime/phase-0.9e-r2-local-download-provenance.md", "docs/runtime/phase-0.9e-r2-inner-archive-correlation.md", "docs/runtime/phase-0.9e-r2-siecaf-structural-analysis.md", "docs/runtime/phase-0.9e-r2-community-backup-correlation.md", "docs/runtime/phase-0.9e-r2-final-provenance-decision.md", "manifests/runtime/phase-0.9e-r2-inner-correlation.json", "manifests/runtime/phase-0.9e-r2-siecaf-fingerprints.json", "tools/inspect_siecaf_header.py", "tools/validate_phase09er2_correlation.py", "tests/test_phase09er2_correlation.py", "tests/test_siecaf_header_parser.py", "packaging/phase09er2/SHA256SUMS.txt", ) FORBIDDEN_SUFFIXES = { ".c", ".cc", ".cpp", ".s", ".asm", ".ld", ".elf", ".self", ".sprx", ".pkg", ".bin", ".wasm", ".sqlite", ".download", } FORBIDDEN_PREFIXES = ("include/", "src/", "adapters/", "samples/") MAX_TRACKED_FILE_SIZE = 10 * 1024 * 1024 def load_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as stream: value = json.load(stream) if not isinstance(value, dict): raise ValueError(f"{path} is not a JSON object") return value def sha256_file(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 sha256_zip_entry(path: Path, entry: str) -> tuple[int, str]: digest = hashlib.sha256() with zipfile.ZipFile(path) as archive: info = archive.getinfo(entry) with archive.open(info) as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return info.file_size, digest.hexdigest() def git(root: Path, *args: str) -> str: result = subprocess.run( ["git", *args], cwd=root, check=False, capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(result.stderr.strip() or "git command failed") return result.stdout.strip() 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 import {path}") module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module def candidate_archive_path(root: Path) -> Path: direct = Path.home() / "Downloads" / OUTER_NAME if direct.is_file(): return direct for ancestor in (root, *root.parents): if ancestor.parent.name.lower() == "users": return ancestor / "Downloads" / OUTER_NAME return direct def inner_byte_match( *, left_size: int, right_size: int, left_sha256: str, right_sha256: str, full_byte_equal: bool, ) -> bool: """Require count, digest, and complete comparison for an inner match.""" return left_size == right_size and left_sha256 == right_sha256 and full_byte_equal def classify_inner_comparison( *, download_valid: bool, inner_present: bool, left_size: int, right_size: int, left_sha256: str, right_sha256: str, full_byte_equal: bool, ) -> str: if not download_valid: return "OFFICIAL_ASSET_DOWNLOAD_INVALID" if not inner_present: return "EXPECTED_INNER_ARCHIVE_ABSENT" if inner_byte_match( left_size=left_size, right_size=right_size, left_sha256=left_sha256, right_sha256=right_sha256, full_byte_equal=full_byte_equal, ): return "INNER_ARCHIVE_BYTE_MATCH" if left_size == right_size: return "INNER_ARCHIVE_SIZE_ONLY_MATCH" return "INNER_ARCHIVE_HASH_MISMATCH" def phase09f_reconsideration_allowed( *, inner_source_bound: bool, mediafire_maker_source_bound: bool, auditable_bootstrap_closure: bool, ) -> bool: return auditable_bootstrap_closure and ( inner_source_bound or mediafire_maker_source_bound ) def _canonical_hash(records: list[dict[str, Any]]) -> str: encoded = json.dumps( records, sort_keys=True, separators=(",", ":"), ensure_ascii=True ).encode("ascii") return hashlib.sha256(encoded).hexdigest() def _normalized_records( archive: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: segment_fields = ( "section_id", "part_number", "data_offset", "aligned_length", "unaligned_length", "hash_key_id_or_algorithm_type", "encryption_key_id_or_algorithm_version", "iv_hex", ) layout_fields = segment_fields[:-1] hash_fields = ("section_id", "section_type", "section_hash_128_hex") segments = [ {field: item[field] for field in segment_fields} for item in sorted( archive["segments"], key=lambda value: ( value["section_id"], value["part_number"], value["data_offset"], ), ) ] hashes = [ {field: item[field] for field in hash_fields} for item in sorted( archive["section_hashes"], key=lambda value: (value["section_id"], value["section_type"]), ) ] layout = [{field: item[field] for field in layout_fields} for item in segments] return segments, hashes, layout def validate_fingerprint_archive(name: str, archive: dict[str, Any]) -> list[str]: errors: list[str] = [] if archive.get("classification") != "SIECAF_VALID_STRUCTURE": errors.append(f"{name}: structure is not valid") return errors header = archive.get("header", {}) if ( "key_or_unknown_16_hex" in header or "raw_hex" in header or header.get("key_or_unknown_16_redacted") is not True or len(header.get("key_or_unknown_16_sha256", "")) != 64 ): errors.append(f"{name}: cryptographic header material is not redacted") count = header.get("segment_count") if len(archive.get("segments", [])) != count: errors.append(f"{name}: segment count mismatch") if len(archive.get("section_hashes", [])) != count: errors.append(f"{name}: hash count mismatch") segments, hashes, layout = _normalized_records(archive) if _canonical_hash(segments) != archive.get("normalized_segment_table_sha256"): errors.append(f"{name}: normalized segment hash mismatch") if _canonical_hash(hashes) != archive.get("normalized_hash_blocks_sha256"): errors.append(f"{name}: normalized hash-block hash mismatch") layout_value = [ { "unknown_u64": header.get("unknown_u64"), "version_i32": header.get("version_i32"), "segment_count": count, "file_offset": header.get("file_offset"), "file_size": header.get("file_size"), }, *layout, ] if _canonical_hash(layout_value) != archive.get("normalized_layout_sha256"): errors.append(f"{name}: normalized layout hash mismatch") structural = _canonical_hash( [ { "header_raw_sha256": header.get("raw_sha256"), "normalized_segment_table_sha256": archive.get( "normalized_segment_table_sha256" ), "normalized_hash_blocks_sha256": archive.get( "normalized_hash_blocks_sha256" ), } ] ) if structural != archive.get("structural_fingerprint_sha256"): errors.append(f"{name}: structural fingerprint mismatch") for field in ( "duplicate_metadata_section_keys", "duplicate_hash_section_ids", "overlaps", "gaps", "errors", "warnings", ): if archive.get(field): errors.append(f"{name}: unexpected {field}") if archive.get("trailing_data_bytes") != 0: errors.append(f"{name}: trailing data is not zero") return errors def validate_fingerprints( root: Path, value: dict[str, Any], inspector: ModuleType, local_path: Path, ) -> list[str]: errors: list[str] = [] if value.get("phase") != PHASE or value.get("status") != "HOST_ONLY_EVIDENCE": errors.append("fingerprint manifest phase/status mismatch") evidence = value.get("parser_evidence", {}) if ( evidence.get("source_commit") != "36d014672bc87577a6e0d750c2cccadc3fae0854" or evidence.get("header_blob") != "fdbc368353a7797464873ada306f0297257eb95e" ): errors.append("public SIECAF source binding changed") for field in ( "decryption_attempted", "content_extraction_attempted", "ps5_bar_tool_executed", ): if evidence.get(field) is not False: errors.append(f"fingerprint manifest permits {field}") archives = value.get("archives", {}) expected_names = { "local", "official_y2jb_1_6_4_03", "community_owendswang_v1_4_autoloader_7_61", } if set(archives) != expected_names: errors.append("fingerprint archive set mismatch") return errors for name, archive in archives.items(): errors.extend(validate_fingerprint_archive(name, archive)) with zipfile.ZipFile(local_path) as local: info = local.getinfo(INNER_NAME) with local.open(info) as stream: observed = inspector.inspect_siecaf( stream, info.file_size, source_label="local-mediafire-inner" ) if observed != archives["local"]: errors.append("local SIECAF evidence does not reproduce") pairs = { "local_vs_official_y2jb_1_6_4_03": ( archives["local"], archives["official_y2jb_1_6_4_03"], ), "local_vs_community_owendswang_v1_4_autoloader_7_61": ( archives["local"], archives["community_owendswang_v1_4_autoloader_7_61"], ), "official_vs_community": ( archives["official_y2jb_1_6_4_03"], archives["community_owendswang_v1_4_autoloader_7_61"], ), } comparisons = value.get("comparisons", {}) for name, (left, right) in pairs.items(): expected = inspector.compare_structures(left, right) if comparisons.get(name) != expected: errors.append(f"{name}: comparison does not reproduce") if expected["classification"] != "SIECAF_LAYOUT_DIFFERENT": errors.append(f"{name}: expected fail-closed layout difference") authorization = value.get("authorization", {}) for field in AUTHORIZATION_FIELDS: if authorization.get(field) is not False: errors.append(f"fingerprints authorize {field}") return errors def validate_manifest(value: dict[str, Any]) -> list[str]: errors: list[str] = [] if value.get("phase") != PHASE or value.get("status") != "HOST_ONLY_COMPLETE": errors.append("main manifest phase/status mismatch") if value.get("baseline_commit") != BASELINE: errors.append("baseline commit mismatch") if value.get("prior_phase_classification") != "LOCAL_BACKUP_NOT_CORRELATED": errors.append("prior classification was not preserved") if value.get("primary_provenance_classification") != PRIMARY_CLASSIFICATION: errors.append("primary provenance classification mismatch") for field in ( "outer_zip_official", "runtime_deployment_verified", "current_device_contents_verified", "phase09f_offline_design_reconsideration_allowed", ): if value.get(field) is not False: errors.append(f"{field} must remain false") if value.get("runtime_firmware_9_60") != "UNPROVEN": errors.append("runtime firmware state was overclaimed") outer = value.get("local_candidate", {}).get("outer", {}) inner = value.get("local_candidate", {}).get("inner", {}) if ( outer.get("name"), outer.get("size"), outer.get("sha256"), ) != (OUTER_NAME, OUTER_SIZE, OUTER_SHA256): errors.append("local outer identity mismatch") if ( inner.get("entry"), inner.get("size"), inner.get("sha256"), ) != (INNER_NAME, INNER_SIZE, INNER_SHA256): errors.append("local inner identity mismatch") provenance = value.get("local_download_provenance", {}) zone = provenance.get("zone_identifier", {}) if provenance.get("classification") != "MEDIAFIRE_URL_EXACT": errors.append("local URL provenance classification mismatch") if zone.get("mediafire_file_key") != "jq3fcutuwbb1mrb": errors.append("MediaFire key mismatch") if "" not in zone.get("host_url", {}).get( "path_redacted", "" ): errors.append("signed MediaFire path is not redacted") browser = provenance.get("browser_history", {}) if ( browser.get("matching_records") != 0 or browser.get("full_history_exported") is not False or browser.get("original_databases_modified") is not False ): errors.append("browser-history boundary mismatch") mediafire = value.get("mediafire_object", {}) if mediafire.get("classification") != "MEDIAFIRE_OBJECT_METADATA_BOUND": errors.append("MediaFire object classification mismatch") if mediafire.get("redownload_performed") is not False: errors.append("MediaFire duplicate download was recorded") official = value.get("official_candidate", {}) if ( official.get("asset_id") != 442358421 or official.get("official_size") != 504395044 or official.get("official_sha256") != "b01b4f442327f9eca90ffc4506dfa58249e4ac70cb9d7488c856c9e8dfaf37b4" or official.get("download_attempts") != 1 or official.get("resume") is not False or official.get("automatic_retry") is not False or official.get("download_valid") is not True ): errors.append("official download evidence mismatch") comparison = official.get("inner_comparison", {}) observed_class = classify_inner_comparison( download_valid=official.get("download_valid") is True, inner_present=True, left_size=comparison.get("official_inner_size", -1), right_size=comparison.get("local_inner_size", -1), left_sha256=comparison.get("official_inner_sha256", ""), right_sha256=comparison.get("local_inner_sha256", ""), full_byte_equal=comparison.get("full_byte_equal") is True, ) if comparison.get("classification") != observed_class: errors.append("official inner classification does not reproduce") families = value.get("community_families", []) if len(families) != 3: errors.append("community family count mismatch") downloaded = [item for item in families if item.get("large_backup_downloaded")] if len(downloaded) != 1: errors.append("community large-download count mismatch") elif ( downloaded[0].get("downloaded_asset", {}).get("inner_comparison") != "INNER_ARCHIVE_HASH_MISMATCH" ): errors.append("community inner mismatch is not preserved") siecaf = value.get("siecaf", {}) for field in ( "local_vs_official", "local_vs_community", "official_vs_community", ): if siecaf.get(field) != "SIECAF_LAYOUT_DIFFERENT": errors.append(f"{field}: structural result overclaimed") for field in ( "decryption_attempted", "content_extraction_attempted", "ps5_bar_tool_executed", ): if siecaf.get(field) is not False: errors.append(f"main manifest permits {field}") authorization = value.get("authorization", {}) for field in AUTHORIZATION_FIELDS: if authorization.get(field) is not False: errors.append(f"main manifest authorizes {field}") actions = value.get("actions", {}) prohibited_true = ( "download_resume_used", "automatic_retry_used", "downloaded_file_executed", "downloaded_file_restored", "ps5_bar_tool_executed", "ps5_connected", "ps5_ip_used", "device_request_performed", "files_transferred_to_or_from_ps5", "target_build_performed", "target_code_created", "target_artifact_created", "payload_created", "device_client_created", ) for field in prohibited_true: if actions.get(field) is not False: errors.append(f"prohibited action recorded: {field}") storage = value.get("temporary_research_storage", {}) if ( storage.get("cleaned") is not True or storage.get("cleanup_verified") is not True or storage.get("large_files_tracked") is not False or storage.get("browser_database_copies_tracked") is not False ): errors.append("temporary-storage final state mismatch") if value.get("blocking_state", {}).get("payload_manager_backup") != ( "HARD_BLOCKER_FOR_INSTALLATION" ): errors.append("Payload Manager installation blocker changed") if phase09f_reconsideration_allowed( inner_source_bound=False, mediafire_maker_source_bound=False, auditable_bootstrap_closure=False, ): errors.append("Phase 0.9F fail-closed decision failed") return errors def validate_checksums(root: Path) -> list[str]: errors: list[str] = [] path = root / "packaging/phase09er2/SHA256SUMS.txt" if not path.is_file(): return ["Phase-0.9E-R2 checksum inventory is missing"] entries = 0 for line in path.read_text(encoding="utf-8").splitlines(): if not line or line.startswith("#"): continue match = re.fullmatch(r"([0-9a-f]{64}) (.+)", line) if not match: errors.append(f"malformed checksum line: {line}") continue expected, relative = match.groups() target = root / relative if not target.is_file(): errors.append(f"checksum target missing: {relative}") elif sha256_file(target) != expected: errors.append(f"checksum mismatch: {relative}") entries += 1 if entries < 12: errors.append("checksum inventory is unexpectedly small") return errors def validate_repository(root: Path) -> list[str]: errors: list[str] = [] if git(root, "branch", "--show-current") != BRANCH: errors.append("current branch mismatch") ancestors = git(root, "merge-base", "--is-ancestor", BASELINE, "HEAD") if ancestors: errors.append("unexpected merge-base output") for relative, expected in IMMUTABLE_HASHES.items(): path = root / relative if not path.is_file() or sha256_file(path) != expected: errors.append(f"immutable evidence changed: {relative}") denylist = root / "manifests/artifact-denylist.json" if sha256_file(denylist) != DENYLIST_SHA256: errors.append("permanent artifact denylist changed") for relative in DELIVERABLES: if not (root / relative).is_file(): errors.append(f"deliverable missing: {relative}") tracked = git(root, "ls-files").splitlines() for relative in tracked: path = root / relative if path.is_file() and path.stat().st_size > MAX_TRACKED_FILE_SIZE: errors.append(f"large tracked file: {relative}") suffix = Path(relative).suffix.lower() if relative.startswith("packaging/phase09er2/") and suffix in { ".zip", ".rar", ".7z", ".dat", }: errors.append(f"backup material tracked: {relative}") changed = git( root, "diff", "--name-only", "--diff-filter=ACMR", f"{BASELINE}...HEAD" ).splitlines() changed.extend(git(root, "diff", "--name-only", "--diff-filter=ACMR").splitlines()) untracked = git(root, "ls-files", "--others", "--exclude-standard").splitlines() changed.extend(untracked) for relative in sorted(set(filter(None, changed))): normalized = relative.replace("\\", "/") suffix = Path(normalized).suffix.lower() if normalized.startswith(FORBIDDEN_PREFIXES) or suffix in FORBIDDEN_SUFFIXES: errors.append(f"target/binary source forbidden in phase: {normalized}") path = root / relative if path.is_file() and path.stat().st_size <= 2 * 1024 * 1024: try: text = path.read_text(encoding="utf-8") except UnicodeDecodeError: errors.append(f"unexpected binary tracked or staged: {normalized}") continue if re.search(r"download2434\.mediafire\.com/[^<\s]+", text): errors.append(f"unredacted signed MediaFire URL: {normalized}") return errors def validate_local_candidate(root: Path) -> tuple[Path, list[str]]: errors: list[str] = [] path = candidate_archive_path(root) if not path.is_file(): return path, [f"local candidate missing: {path}"] if path.stat().st_size != OUTER_SIZE or sha256_file(path) != OUTER_SHA256: errors.append("local outer candidate identity changed") return path, errors try: size, digest = sha256_zip_entry(path, INNER_NAME) except (KeyError, OSError, zipfile.BadZipFile) as error: errors.append(f"local inner candidate cannot be read: {error}") else: if size != INNER_SIZE or digest != INNER_SHA256: errors.append("local inner candidate identity changed") return path, errors def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--root", type=Path, required=True) args = parser.parse_args() root = args.root.resolve() errors: list[str] = [] local_path, candidate_errors = validate_local_candidate(root) errors.extend(candidate_errors) errors.extend(validate_repository(root)) main_manifest = load_json( root / "manifests/runtime/phase-0.9e-r2-inner-correlation.json" ) fingerprints = load_json( root / "manifests/runtime/phase-0.9e-r2-siecaf-fingerprints.json" ) errors.extend(validate_manifest(main_manifest)) if local_path.is_file(): inspector = load_module( "phase09er2_siecaf_inspector", root / "tools/inspect_siecaf_header.py", ) errors.extend(validate_fingerprints(root, fingerprints, inspector, local_path)) errors.extend(validate_checksums(root)) if errors: for error in errors: print(f"ERROR: {error}") return 1 print( "Phase-0.9E-R2 validation PASS: " "LOCAL_BACKUP_UNCORRELATED; no device action authorized" ) return 0 if __name__ == "__main__": raise SystemExit(main())