#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Validate Phase-1.0B bounded RetroArch smoke evidence."""

from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import struct
import subprocess
import sys
from typing import Any


PHASE = "PHASE_1_0B_FIRST_DEVICE_SMOKE_CANDIDATE_HARDENING"
STATUS = "RETROARCH_PS5_DEVICE_SMOKE_CANDIDATE_BUILT_WITH_DECLARED_RISKS"
GFX_BRANCH = "codex/chimera-gfx-phase10b-device-smoke-hardening"
RETROARCH_BRANCH = "codex/ps5-device-smoke-hardening"
RETROARCH_BASE = "ca1b45680577befc743e1c92fa40687e1b1745e7"
RETROARCH_RELEASE = "69a4f0ea1e8aaf442ae4858f2e7f2b31a1776576"
RETROARCH_TREE = "33babf9eb7699b5d571a3063ea21c3e488c159fe"
SDK_COMMIT = "d2e2e585740362976a39fdd5ccf390f199a7bc37"
PACBREW_COMMIT = "c2abcfcb60f569128abd0e8e70ad03a67bee5ea7"
SDL_COMMIT = "0baf4ac49382b537ba449901b5b6d0d189bb1fbb"
DENYLIST_SHA256 = "e9603b0e3792781ad5b511afb22ef61e3d4fd4c5a16bf928f8609193bcd97783"

DEVICE_AUTHORIZATION_FIELDS = (
    "ps5_connection_authorized",
    "device_transfer_authorized",
    "device_execution_authorized",
    "installation_authorized",
    "lifecycle_authorized",
    "autoload_authorized",
    "device_write_authorized",
    "automatic_retry",
)
DEVICE_ACTION_FIELDS = (
    "ps5_connected",
    "device_request_performed",
    "files_transferred",
    "device_write_performed",
    "target_execution_performed",
    "transfer_package_created",
    "execution_package_created",
    "installation_package_created",
)
DISABLED_FEATURES = {
    "networking",
    "online_updater",
    "dynamic_cores",
    "content_browser_data",
    "playlists",
    "history",
    "screenshots",
    "recording",
    "achievements",
    "shaders",
    "databases",
    "compressed_content",
    "sram",
    "savestates",
    "config_save",
    "remap_save",
    "autoconfig_save",
    "log_file",
    "temporary_files",
    "keyboard_ime",
    "haptics",
    "rumble",
    "lightbar",
    "gnm",
    "retroarch_dynamic_core_loading",
    "installation",
    "payload_launch",
    "autoload",
}
FORBIDDEN_IMPORT_MARKERS = (
    "scenet",
    "scehttp",
    "scessl",
    "scegnm",
    "scekernelloadstartmodule",
    "scepadsetvibration",
    "scepadsetlightbar",
    "sceime",
    "scekeyboard",
    "dlopen",
    "dlsym",
    "socket",
    "connect",
    "listen",
    "accept",
    "send",
    "recv",
)
REQUIRED_SYMBOLS = {
    "rarch_main",
    "retroarch_main_init",
    "runloop_iterate",
    "retro_init",
    "retro_deinit",
    "retro_run",
    "frontend_ctx_ps5",
    "chimera_ps5_smoke_tick",
    "chimera_ps5_smoke_block_write",
    "__wrap_open",
    "__wrap_fopen",
    "__wrap_fwrite",
}
DELIVERABLES = (
    "docs/retroarch/phase-1.0b-smoke-candidate-design.md",
    "docs/retroarch/phase-1.0b-runtime-and-exit-contract.md",
    "docs/retroarch/phase-1.0b-persistent-write-audit.md",
    "docs/retroarch/phase-1.0b-linker-and-wx-analysis.md",
    "docs/retroarch/phase-1.0b-device-risk-assessment.md",
    "docs/retroarch/phase-1.0b-proposed-one-shot-test.md",
    "docs/approvals/phase-1.0b-device-smoke-template.md",
    "manifests/retroarch/phase-1.0b-build.json",
    "manifests/retroarch/phase-1.0b-artifact.json",
    "manifests/retroarch/phase-1.0b-runtime-contract.json",
    "tests/test_retroarch_phase10b.py",
    "tools/validate_retroarch_phase10b.py",
    "packaging/retroarch/phase10b/SHA256SUMS.txt",
)


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 profile_is_closed(profile: dict[str, Any]) -> bool:
    return (
        DISABLED_FEATURES.issubset(set(profile.get("disabled_features", [])))
        and profile.get("persistent_writes_allowed") is False
        and profile.get("content_required") is False
        and profile.get("config_required") is False
        and profile.get("networking") is False
        and profile.get("autoload") is False
        and profile.get("automatic_retry") is False
    )


def route_is_read_only(route: dict[str, Any]) -> bool:
    return (
        route.get("write") is False
        and route.get("create") is False
        and route.get("append") is False
        and route.get("truncate") is False
        and route.get("rename") is False
        and route.get("unlink") is False
        and route.get("mkdir") is False
        and route.get("retry") is False
    )


def program_headers_are_wx_closed(headers: list[dict[str, Any]]) -> bool:
    loads = [header for header in headers if header.get("type") == "LOAD"]
    return (
        len(loads) >= 3
        and any("E" in str(item.get("flags", "")) for item in loads)
        and all(
            not (
                "W" in str(item.get("flags", ""))
                and "E" in str(item.get("flags", ""))
            )
            for item in loads
        )
    )


def imports_are_closed(imports: list[str]) -> bool:
    lowered = [item.lower() for item in imports]
    return not any(
        marker in item for marker in FORBIDDEN_IMPORT_MARKERS for item in lowered
    )


def parse_elf_program_headers(path: Path) -> list[dict[str, Any]]:
    with path.open("rb") as stream:
        header = stream.read(64)
        if len(header) != 64 or header[:5] != b"\x7fELF\x02":
            raise ValueError(f"{path} is not ELF64")
        byte_order = "<" if header[5] == 1 else ">"
        phoff = struct.unpack_from(f"{byte_order}Q", header, 32)[0]
        phentsize = struct.unpack_from(f"{byte_order}H", header, 54)[0]
        phnum = struct.unpack_from(f"{byte_order}H", header, 56)[0]
        if phentsize < 56 or phnum > 64:
            raise ValueError("invalid ELF program-header table")
        result: list[dict[str, Any]] = []
        stream.seek(phoff)
        for _ in range(phnum):
            raw = stream.read(phentsize)
            if len(raw) != phentsize:
                raise ValueError("truncated ELF program-header table")
            kind, flags = struct.unpack_from(f"{byte_order}II", raw)
            names = {1: "LOAD", 2: "DYNAMIC"}
            text = (
                ("R" if flags & 4 else "")
                + ("W" if flags & 2 else "")
                + ("E" if flags & 1 else "")
            )
            result.append(
                {
                    "type": names.get(kind, f"0x{kind:x}"),
                    "flags": text,
                    "offset": struct.unpack_from(f"{byte_order}Q", raw, 8)[0],
                    "virtual_address": struct.unpack_from(
                        f"{byte_order}Q", raw, 16
                    )[0],
                    "file_size": struct.unpack_from(f"{byte_order}Q", raw, 32)[0],
                    "memory_size": struct.unpack_from(
                        f"{byte_order}Q", raw, 40
                    )[0],
                    "alignment": struct.unpack_from(f"{byte_order}Q", raw, 48)[0],
                }
            )
        return result


def parse_elf_sections(path: Path) -> tuple[bytes, list[tuple[int, ...]]]:
    data = path.read_bytes()
    if len(data) < 64 or data[:6] != b"\x7fELF\x02\x01":
        raise ValueError(f"{path} is not little-endian ELF64")
    section_offset = struct.unpack_from("<Q", data, 40)[0]
    section_size, section_count = struct.unpack_from("<HH", data, 58)
    if section_size < 64 or section_count > 4096:
        raise ValueError("invalid ELF section-header table")
    sections: list[tuple[int, ...]] = []
    for index in range(section_count):
        offset = section_offset + index * section_size
        if offset + 64 > len(data):
            raise ValueError("truncated ELF section-header table")
        sections.append(struct.unpack_from("<IIQQQQIIQQ", data, offset))
    return data, sections


def parse_elf_dynamic_symbols(path: Path) -> list[str]:
    data, sections = parse_elf_sections(path)
    undefined: set[str] = set()
    for section in sections:
        if section[1] != 11:  # SHT_DYNSYM
            continue
        symbol_offset, symbol_size = section[4], section[5]
        string_index, symbol_entry_size = section[6], section[9]
        if string_index >= len(sections) or symbol_entry_size < 24:
            raise ValueError("invalid ELF dynamic-symbol table")
        strings_section = sections[string_index]
        strings = data[
            strings_section[4] : strings_section[4] + strings_section[5]
        ]
        for offset in range(
            symbol_offset, symbol_offset + symbol_size, symbol_entry_size
        ):
            if offset + 24 > len(data):
                raise ValueError("truncated ELF dynamic-symbol table")
            name_offset, _, _, section_index, _, _ = struct.unpack_from(
                "<IBBHQQ", data, offset
            )
            if section_index != 0 or name_offset >= len(strings):
                continue
            name_end = strings.find(b"\0", name_offset)
            if name_end < 0:
                raise ValueError("unterminated ELF dynamic-symbol name")
            name = strings[name_offset:name_end].decode("utf-8")
            if name:
                undefined.add(name)
    return sorted(undefined)


def parse_elf_needed(path: Path) -> list[str]:
    data, sections = parse_elf_sections(path)
    needed: list[str] = []
    for section in sections:
        if section[1] != 6:  # SHT_DYNAMIC
            continue
        dynamic_offset, dynamic_size = section[4], section[5]
        string_index, dynamic_entry_size = section[6], section[9]
        if string_index >= len(sections) or dynamic_entry_size < 16:
            raise ValueError("invalid ELF dynamic table")
        strings_section = sections[string_index]
        strings = data[
            strings_section[4] : strings_section[4] + strings_section[5]
        ]
        for offset in range(
            dynamic_offset, dynamic_offset + dynamic_size, dynamic_entry_size
        ):
            if offset + 16 > len(data):
                raise ValueError("truncated ELF dynamic table")
            tag, value = struct.unpack_from("<QQ", data, offset)
            if tag != 1:  # DT_NEEDED
                continue
            if value >= len(strings):
                raise ValueError("invalid ELF DT_NEEDED string offset")
            name_end = strings.find(b"\0", value)
            if name_end < 0:
                raise ValueError("unterminated ELF DT_NEEDED name")
            needed.append(strings[value:name_end].decode("utf-8"))
    return needed


def parse_elf_relocations(path: Path) -> dict[str, Any]:
    data, sections = parse_elf_sections(path)
    writable_ranges = [
        (
            header["virtual_address"],
            header["virtual_address"] + header["memory_size"],
        )
        for header in parse_elf_program_headers(path)
        if header["type"] == "LOAD" and "W" in header["flags"]
    ]
    total = 0
    relative = 0
    outside = 0
    relative_outside = 0
    by_type: dict[str, int] = {}
    for section in sections:
        if section[1] != 4:  # SHT_RELA
            continue
        relocation_offset, relocation_size = section[4], section[5]
        relocation_entry_size = section[9]
        if relocation_entry_size < 24:
            raise ValueError("invalid ELF relocation table")
        for offset in range(
            relocation_offset,
            relocation_offset + relocation_size,
            relocation_entry_size,
        ):
            if offset + 24 > len(data):
                raise ValueError("truncated ELF relocation table")
            target, info, _ = struct.unpack_from("<QQq", data, offset)
            relocation_type = info & 0xFFFFFFFF
            key = str(relocation_type)
            by_type[key] = by_type.get(key, 0) + 1
            total += 1
            in_writable_load = any(
                start <= target and target + 8 <= end
                for start, end in writable_ranges
            )
            if not in_writable_load:
                outside += 1
            if relocation_type == 8:  # R_X86_64_RELATIVE
                relative += 1
                if not in_writable_load:
                    relative_outside += 1
    return {
        "total": total,
        "relative": relative,
        "by_type": by_type,
        "outside_rw_load": outside,
        "relative_outside_rw_load": relative_outside,
    }


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 command failed")
    return result.stdout.strip()


def validate(root: Path, retroarch_root: Path | None = None) -> list[str]:
    errors: list[str] = []
    for relative in DELIVERABLES:
        if not (root / relative).is_file():
            errors.append(f"missing deliverable: {relative}")
    try:
        build = load_json(root / "manifests/retroarch/phase-1.0b-build.json")
        artifact = load_json(root / "manifests/retroarch/phase-1.0b-artifact.json")
        runtime = load_json(
            root / "manifests/retroarch/phase-1.0b-runtime-contract.json"
        )
    except (OSError, ValueError, json.JSONDecodeError) as error:
        return errors + [str(error)]

    if build.get("phase") != PHASE or build.get("status") != STATUS:
        errors.append("phase/status mismatch")
    if build.get("branch") != GFX_BRANCH:
        errors.append("GFX branch mismatch")
    for field in DEVICE_AUTHORIZATION_FIELDS:
        if build.get(field) is not False:
            errors.append(f"{field} must remain false")
    for field in DEVICE_ACTION_FIELDS:
        if build.get(field) is not False:
            errors.append(f"{field} must remain false")

    source = build.get("sources", {})
    expected_sources = {
        "retroarch_release_commit": RETROARCH_RELEASE,
        "retroarch_release_tree": RETROARCH_TREE,
        "retroarch_fork_base_commit": RETROARCH_BASE,
        "ps5_payload_sdk_commit": SDK_COMMIT,
        "pacbrew_commit": PACBREW_COMMIT,
        "sdl_commit": SDL_COMMIT,
    }
    for field, expected in expected_sources.items():
        if source.get(field) != expected:
            errors.append(f"source mismatch: {field}")
    if (
        build.get("chimera_retroarch", {}).get("branch") != RETROARCH_BRANCH
        or len(build.get("chimera_retroarch", {}).get("source_commit", "")) != 40
    ):
        errors.append("RetroArch fork identity is incomplete")
    if not profile_is_closed(build.get("profile", {})):
        errors.append("smoke profile is not closed")

    record = artifact.get("artifact", {})
    if record.get("name") != "retroarch_ps5_software_smoke.elf":
        errors.append("artifact label mismatch")
    if record.get("size", 0) <= 0 or len(record.get("sha256", "")) != 64:
        errors.append("artifact identity is incomplete")
    if len(record.get("linker_map_sha256", "")) != 64:
        errors.append("linker-map identity is incomplete")
    hashes = record.get("clean_build_sha256", [])
    if len(hashes) != 2 or len(set(hashes)) != 1 or hashes[0] != record.get("sha256"):
        errors.append("two clean builds are not byte-identical")
    if record.get("selected_exit_method") != "PROCESS__EXIT_AFTER_TEARDOWN":
        errors.append("exit method mismatch")
    if record.get("entrypoint") is None:
        errors.append("entrypoint missing")
    if not program_headers_are_wx_closed(record.get("program_headers", [])):
        errors.append("program headers are not W^X-closed")
    if not imports_are_closed(record.get("imports", [])):
        errors.append("forbidden import present")
    if not REQUIRED_SYMBOLS.issubset(set(record.get("defined_symbols", []))):
        errors.append("required real/smoke symbols are incomplete")
    if record.get("relative_relocations_outside_rw_load") != 0:
        errors.append("relative relocation target outside RW load")
    for field in (
        "persistent_writes_allowed",
        "content_required",
        "config_required",
        "networking",
        "autoload",
        "automatic_retry",
        "installation_eligible",
        "device_write_eligible",
        "transfer_eligible",
        "execution_eligible",
        "execution_authorized",
    ):
        if record.get(field) is not False:
            errors.append(f"artifact {field} must be false")

    if runtime.get("runtime_limit_ms") != 60000:
        errors.append("runtime limit mismatch")
    if runtime.get("frame_limit") != 3600:
        errors.append("frame limit mismatch")
    if runtime.get("shutdown_hold_ms") != 2000:
        errors.append("shutdown hold mismatch")
    if runtime.get("shutdown_request_max") != 1:
        errors.append("shutdown request is not single-shot")
    if runtime.get("exit_reachable_call_count") != 1:
        errors.append("exit is not exactly once reachable")
    if runtime.get("audio_submit_bounded") is not False:
        errors.append("blocking AudioOut risk was hidden")
    if not all(route_is_read_only(route) for route in runtime.get("write_routes", [])):
        errors.append("write route is not fail-closed")

    denylist = root / "manifests/artifact-denylist.json"
    if not denylist.is_file() or sha256_file(denylist) != DENYLIST_SHA256:
        errors.append("permanent denylist changed")
    for relative in git(root, "ls-files").splitlines():
        if relative.lower().endswith((".elf", ".self", ".sprx", ".pkg")):
            errors.append(f"tracked target artifact: {relative}")

    approval = (root / "docs/approvals/phase-1.0b-device-smoke-template.md").read_text(
        encoding="utf-8"
    )
    for value in (
        "authorized: false",
        "transfer_authorized: false",
        "execution_authorized: false",
        "installation_authorized: false",
        "automatic_retry: false",
        "autoload_authorized: false",
    ):
        if value not in approval:
            errors.append(f"approval template missing {value}")

    if retroarch_root is not None:
        path = retroarch_root / record.get("local_relative_path", "")
        map_path = retroarch_root / record.get("linker_map_relative_path", "")
        if not path.is_file():
            errors.append("local smoke ELF missing")
        elif path.stat().st_size != record.get("size") or sha256_file(path) != record.get(
            "sha256"
        ):
            errors.append("local smoke ELF identity mismatch")
        else:
            try:
                actual_headers = parse_elf_program_headers(path)
                if not program_headers_are_wx_closed(actual_headers):
                    errors.append("local smoke ELF has a W+X load segment")
                if actual_headers != record.get("program_headers"):
                    errors.append("local program headers differ from manifest")
                actual_needed = parse_elf_needed(path)
                if actual_needed != record.get("dt_needed"):
                    errors.append("local DT_NEEDED differs from manifest")
                actual_undefined = parse_elf_dynamic_symbols(path)
                if actual_undefined != record.get("undefined_symbols"):
                    errors.append("local undefined symbols differ from manifest")
                if not imports_are_closed(actual_undefined):
                    errors.append("local smoke ELF has a forbidden import")
                actual_relocations = parse_elf_relocations(path)
                if actual_relocations != record.get("relocation_audit"):
                    errors.append("local relocation audit differs from manifest")
            except (OSError, ValueError, struct.error) as error:
                errors.append(str(error))
        if not map_path.is_file():
            errors.append("local linker map missing")
        elif sha256_file(map_path) != record.get("linker_map_sha256"):
            errors.append("local linker-map identity mismatch")
        if git(retroarch_root, "rev-parse", "HEAD") != build.get(
            "chimera_retroarch", {}
        ).get("source_commit"):
            errors.append("RetroArch HEAD does not match manifest source")
    return errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--retroarch-root", type=Path)
    args = parser.parse_args()
    errors = validate(
        args.root.resolve(),
        args.retroarch_root.resolve() if args.retroarch_root else None,
    )
    if errors:
        for error in errors:
            print(error, file=sys.stderr)
        return 1
    print("Phase 1.0B bounded RetroArch smoke evidence validated")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())