#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Validate the offline Phase-1.0D diagnostic ladder evidence."""

from __future__ import annotations

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


PHASE = "PHASE_1_0D_LOADER_TO_ENTRY_DIAGNOSIS"
STATUS = "RETROARCH_PS5_ENTRY_DIAGNOSTIC_LADDER_BUILT_OFFLINE"
RETROARCH_BRANCH = "codex/ps5-loader-entry-diagnosis"
RETROARCH_COMMIT = "69b65858ffaee826d70f5c0df61013cd1b0e2048"
DENYLIST_SHA256 = "e9603b0e3792781ad5b511afb22ef61e3d4fd4c5a16bf928f8609193bcd97783"
CANARY_IMPORTS = {
    "_Exit",
    "memset",
    "nanosleep",
    "sceKernelSendNotificationRequest",
}
CANARY_NEEDED = {"libSceLibcInternal.sprx", "libkernel_web.sprx"}
REAL_MODEL_ARTIFACTS = {
    "retroarch_ps5_software_smoke.elf",
    "chimera_ps5_crt_entry_canary.elf",
    "retroarch_ps5_early_diag.elf",
    "chimera-gfx-lifecycle-probe.elf",
}
AUTHORIZATION_FIELDS = (
    "ps5_connection_authorized",
    "device_transfer_authorized",
    "device_execution_authorized",
    "installation_authorized",
    "autoload_authorized",
    "device_write_authorized",
    "automatic_retry",
)
ACTION_FIELDS = (
    "ps5_connected",
    "device_request_performed",
    "files_transferred",
    "device_write_performed",
    "target_execution_performed",
)
DELIVERABLES = (
    "docs/retroarch/phase-1.0d-loader-to-entry-analysis.md",
    "docs/retroarch/phase-1.0d-crt-entry-canary.md",
    "docs/retroarch/phase-1.0d-early-diagnostic-design.md",
    "docs/retroarch/phase-1.0d-startup-import-closure.md",
    "docs/retroarch/phase-1.0d-loader-static-model.md",
    "docs/retroarch/phase-1.0d-next-device-test-ladder.md",
    "manifests/retroarch/phase-1.0d-canary-artifact.json",
    "manifests/retroarch/phase-1.0d-early-diag-artifact.json",
    "manifests/retroarch/phase-1.0d-loader-model-results.json",
    "tools/validate_retroarch_phase10d.py",
    "tests/test_retroarch_phase10d.py",
    "packaging/retroarch/phase10d/SHA256SUMS.txt",
)


def load_json(path: Path) -> dict[str, Any]:
    value = json.loads(path.read_text(encoding="utf-8"))
    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 all_false(record: dict[str, Any], fields: tuple[str, ...]) -> bool:
    return all(record.get(field) is False for field in fields)


def reproducible_artifact(record: dict[str, Any]) -> bool:
    artifact = record.get("artifact", {})
    elf_hashes = artifact.get("clean_build_sha256", [])
    map_hashes = artifact.get("clean_map_sha256", [])
    return (
        artifact.get("size", 0) > 0
        and len(artifact.get("sha256", "")) == 64
        and len(elf_hashes) == 2
        and len(set(elf_hashes)) == 1
        and elf_hashes[0] == artifact.get("sha256")
        and len(map_hashes) == 2
        and len(set(map_hashes)) == 1
        and map_hashes[0] == artifact.get("linker_map_sha256")
    )


def wx_closed(headers: list[dict[str, Any]]) -> bool:
    loads = [item for item in headers if item.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 canary_is_minimal(record: dict[str, Any]) -> bool:
    behavior = record.get("behavior", {})
    elf = record.get("elf", {})
    return (
        set(elf.get("undefined_symbols", [])) == CANARY_IMPORTS
        and set(elf.get("dt_needed", [])) == CANARY_NEEDED
        and record.get("notification_abi", {}).get("maximum_attempts") == 1
        and record.get("notification_abi", {}).get("retry") is False
        and behavior.get("sleep_attempts") == 1
        and behavior.get("interrupted_sleep_retry") is False
        and all(
            behavior.get(field) is False
            for field in (
                "retroarch", "sdl", "videoout", "pad", "audioout",
                "filesystem", "networking", "threads", "autoload", "installation",
            )
        )
    )


def early_ladder_is_closed(record: dict[str, Any]) -> bool:
    diagnostic = record.get("diagnostic", {})
    frame = record.get("first_frame", {})
    policy = record.get("profile_policy", {})
    return (
        diagnostic.get("stages") == [f"D{index:02d}" for index in range(13)]
        and diagnostic.get("notification_maximum_attempts_per_stage") == 1
        and diagnostic.get("notification_retry") is False
        and diagnostic.get("notification_failure_blocks_primary_path") is False
        and frame.get("background") == "MAGENTA"
        and frame.get("fixed_rectangle") == "WHITE"
        and frame.get("embedded_pattern") == "BLACK"
        and frame.get("submit_attempts") == 1
        and frame.get("retry") is False
        and frame.get("second_buffer_initialization") is False
        and policy.get("write_firewall") is True
        and all(
            policy.get(field) is False
            for field in (
                "filesystem_writes_allowed", "networking", "dynamic_cores",
                "gnm", "autoload", "installation", "payload_launch",
                "automatic_retry",
            )
        )
    )


def model_result_set_is_complete(record: dict[str, Any]) -> bool:
    artifacts = record.get("real_artifacts", [])
    names = {item.get("name") for item in artifacts}
    return (
        names == REAL_MODEL_ARTIFACTS
        and all(
            item.get("classification")
            in {
                "ACCEPTED_BY_STATIC_MODEL",
                "REJECTED_BY_STATIC_MODEL",
                "MODEL_INCOMPLETE",
            }
            for item in artifacts
        )
        and all(len(item.get("sha256", "")) == 64 for item in artifacts)
        and record.get("model", {}).get("hardware_evidence") is False
    )


def sender_trace_is_bounded(trace: dict[str, Any]) -> bool:
    return (
        trace.get("connections") == 1
        and trace.get("sendall_calls") == 1
        and trace.get("hash_before_connect") is True
        and trace.get("shutdown_called") is False
        and trace.get("response_read") is False
        and trace.get("retry") is False
        and trace.get("reconnect") is False
        and trace.get("probe") is False
        and "REMOTE_EXECUTION" in trace.get("cannot_prove", [])
    )


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 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:
        canary = load_json(
            root / "manifests/retroarch/phase-1.0d-canary-artifact.json"
        )
        early = load_json(
            root / "manifests/retroarch/phase-1.0d-early-diag-artifact.json"
        )
        model = load_json(
            root / "manifests/retroarch/phase-1.0d-loader-model-results.json"
        )
    except (OSError, ValueError, json.JSONDecodeError) as error:
        return errors + [str(error)]

    for label, record in (("canary", canary), ("early", early), ("model", model)):
        if record.get("phase") != PHASE or record.get("status") != STATUS:
            errors.append(f"{label} phase/status mismatch")
        if not all_false(record.get("authorizations", {}), AUTHORIZATION_FIELDS):
            errors.append(f"{label} authorization must remain false")
        if not all_false(record.get("actions", {}), ACTION_FIELDS):
            errors.append(f"{label} device action must remain false")

    for label, record in (("canary", canary), ("early", early)):
        artifact = record.get("artifact", {})
        elf = record.get("elf", {})
        if not reproducible_artifact(record):
            errors.append(f"{label} is not reproducible")
        if artifact.get("execution_eligible") is not False:
            errors.append(f"{label} execution eligibility must be false")
        if elf.get("rwx_load_segment_count") != 0:
            errors.append(f"{label} RWX count is not zero")
        if not wx_closed(elf.get("program_headers", [])):
            errors.append(f"{label} headers are not W^X closed")
        if elf.get("init_array_size") != 0 or elf.get("fini_array_size") != 0:
            errors.append(f"{label} constructor arrays are not empty")
        if elf.get("tls") is not False:
            errors.append(f"{label} TLS must be absent")
    if not canary_is_minimal(canary):
        errors.append("canary closure is not minimal")
    if not early_ladder_is_closed(early):
        errors.append("early diagnostic ladder is not closed")
    if not model_result_set_is_complete(model):
        errors.append("loader model real-artifact set is incomplete")
    if not sender_trace_is_bounded(model.get("sender_trace_contract", {})):
        errors.append("sender trace contract is not bounded")

    negative = model.get("negative_host_cases", [])
    if len(negative) != 5 or {
        item.get("classification") for item in negative
    } != {
        "ACCEPTED_BY_STATIC_MODEL",
        "REJECTED_BY_STATIC_MODEL",
        "MODEL_INCOMPLETE",
    }:
        errors.append("negative loader cases are incomplete")

    denylist = root / "manifests/artifact-denylist.json"
    if not denylist.is_file() or sha256_file(denylist) != DENYLIST_SHA256:
        errors.append("permanent denylist changed")
    tracked = git(root, "ls-files").splitlines()
    for relative in tracked:
        if relative.lower().endswith((".elf", ".self", ".sprx", ".pkg")):
            errors.append(f"tracked target artifact: {relative}")
    forbidden_address = "192.168.10." + "105"
    for relative in tracked:
        path = root / relative
        try:
            text = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        if forbidden_address in text:
            errors.append(f"tracked device address in {relative}")

    if retroarch_root is not None:
        if git(retroarch_root, "rev-parse", "HEAD") != RETROARCH_COMMIT:
            errors.append("chimera-retroarch HEAD mismatch")
        if git(retroarch_root, "branch", "--show-current") != RETROARCH_BRANCH:
            errors.append("chimera-retroarch branch mismatch")
        for label, record in (("canary", canary), ("early", early)):
            artifact = record["artifact"]
            elf_path = retroarch_root / artifact["local_relative_path"]
            map_path = retroarch_root / artifact["linker_map_relative_path"]
            if not elf_path.is_file():
                errors.append(f"{label} local ELF missing")
            elif (
                elf_path.stat().st_size != artifact["size"]
                or sha256_file(elf_path) != artifact["sha256"]
            ):
                errors.append(f"{label} local ELF identity mismatch")
            if not map_path.is_file():
                errors.append(f"{label} local map missing")
            elif sha256_file(map_path) != artifact["linker_map_sha256"]:
                errors.append(f"{label} local map identity mismatch")
    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:
        print("\n".join(errors), file=sys.stderr)
        return 1
    print("Phase 1.0D offline diagnostic ladder evidence validated")
    return 0


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