#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Audit SDK v0.41 startup and decide whether a minimal ELF may be built."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import subprocess
import tempfile
from collections import deque
from pathlib import Path
from typing import Any


EXPECTED_SDK_COMMIT = "d2e2e585740362976a39fdd5ccf390f199a7bc37"
CRT_OBJECTS = {
    "crt.o": "crt/crt.c",
    "syscall.o": "crt/syscall.c",
    "klog.o": "crt/klog.c",
    "nid.o": "crt/nid.c",
    "kernel.o": "crt/kernel.c",
    "rtld.o": "crt/rtld.c",
    "rtld_so.o": "crt/rtld_so.c",
    "rtld_sprx.o": "crt/rtld_sprx.c",
    "rtld_payload.o": "crt/rtld_payload.c",
    "rtld_dlfcn.o": "crt/rtld_dlfcn.c",
    "mdbg.o": "crt/mdbg.c",
    "patch.o": "crt/patch.c",
}
EMPTY_CRT_ARCHIVES = (
    "crti.o",
    "crtn.o",
    "crtbegin.o",
    "crtend.o",
    "crtbeginS.o",
    "crtendS.o",
)
DANGEROUS_REACHABLE = {
    "__patch_init",
    "kernel_copyin",
    "kernel_copyout",
    "kernel_set_ucred_attrs",
    "kernel_set_ucred_caps",
}
LINKED_PROHIBITED = DANGEROUS_REACHABLE | {
    "__dlopen",
    "__dlsym",
    "kernel_mprotect",
    "kernel_overlap_sockets",
    "kernel_set_vmem_protection",
}


def run(command: list[str], allowed: tuple[int, ...] = (0,)) -> str:
    result = subprocess.run(command, check=False, capture_output=True, text=True)
    if result.returncode not in allowed:
        raise ValueError(
            f"command failed ({result.returncode}): {' '.join(command)}\n"
            f"{result.stdout}{result.stderr}"
        )
    return result.stdout + result.stderr


def sha256_bytes(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


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 require(text: str, snippets: tuple[str, ...], source: str) -> None:
    for snippet in snippets:
        if snippet not in text:
            raise ValueError(f"{source}: missing pinned evidence: {snippet}")


def normalize_target(symbol: str) -> str:
    symbol = re.sub(r"[-+]0x[0-9a-f]+$", "", symbol)
    symbol = re.sub(r"[-+][0-9a-f]+$", "", symbol)
    return symbol


def parse_symbols(nm_output: str) -> tuple[set[str], set[str]]:
    functions: set[str] = set()
    undefined: set[str] = set()
    for line in nm_output.splitlines():
        parts = line.split()
        if len(parts) < 2:
            continue
        name, kind = parts[0], parts[1]
        if kind in {"T", "t"}:
            functions.add(name)
        elif kind.lower() == "u" or kind == "w":
            undefined.add(name)
    return functions, undefined


def parse_callgraph(disassembly: str) -> list[dict[str, str]]:
    label = re.compile(r"^[0-9a-f]+ <([^>]+)>:$")
    instruction = re.compile(
        r"^\s*([0-9a-f]+):\s+([a-z][a-z0-9.]*)\s*(.*?)\s*$"
    )
    relocation = re.compile(r"^\s*[0-9a-f]+:\s+R_X86_64_\S+\s+(\S+)")
    current: str | None = None
    pending: dict[str, str] | None = None
    edges: list[dict[str, str]] = []

    def flush() -> None:
        nonlocal pending
        if pending is None:
            return
        operand = pending.pop("operand")
        direct = re.search(r"<([^>]+)>", operand)
        if "target" not in pending:
            if direct is not None and "+0x" not in direct.group(1):
                pending["target"] = direct.group(1)
            else:
                pending["target"] = f"INDIRECT:{operand}"
        edges.append(pending)
        pending = None

    for raw_line in disassembly.splitlines():
        line = raw_line.rstrip()
        match = label.match(line)
        if match is not None:
            flush()
            current = match.group(1)
            continue
        match = relocation.match(line)
        if match is not None and pending is not None:
            pending["target"] = normalize_target(match.group(1))
            flush()
            continue
        match = instruction.match(line)
        if match is None:
            continue
        flush()
        if current is None:
            continue
        address, mnemonic, operand = match.groups()
        if mnemonic in {"call", "callq"}:
            pending = {
                "address": address,
                "caller": current,
                "kind": "call",
                "operand": operand,
            }
        elif mnemonic in {"jmp", "jmpq"} and (
            operand.startswith("*") or ("<" in operand and "+0x" not in operand)
        ):
            pending = {
                "address": address,
                "caller": current,
                "kind": "tail_call",
                "operand": operand,
            }
    flush()
    return edges


def reachable_callgraph(
    edges: list[dict[str, str]], functions: set[str]
) -> tuple[list[str], list[dict[str, str]]]:
    by_caller: dict[str, list[dict[str, str]]] = {}
    for edge in edges:
        by_caller.setdefault(edge["caller"], []).append(edge)
    visited: set[str] = set()
    selected: list[dict[str, str]] = []
    queue: deque[str] = deque(["_start"])
    while queue:
        caller = queue.popleft()
        if caller in visited:
            continue
        visited.add(caller)
        for edge in by_caller.get(caller, []):
            selected.append(edge)
            target = edge["target"]
            if target in functions and target not in visited:
                queue.append(target)
    return sorted(visited), sorted(
        selected, key=lambda item: (item["caller"], item["address"], item["target"])
    )


def driver_trace(compiler: Path, source: Path, freestanding: bool) -> str:
    command = [str(compiler), "-###", "-Qunused-arguments", "-Werror"]
    if freestanding:
        command.extend(["-nostartfiles", "-nodefaultlibs"])
    command.extend([str(source), "-o", str(source.with_suffix(".never.elf"))])
    trace = run(command)
    if source.with_suffix(".never.elf").exists():
        raise ValueError("compiler -### unexpectedly produced an ELF")
    trace = trace.replace(str(source.parent), "<TEMP>")
    return re.sub(r"trace-[0-9a-f]+\.o", "trace-<ID>.o", trace)


def git_grep_loader_contract(checkout: Path) -> list[str]:
    if not (checkout / ".git").exists():
        return []
    output = run(
        [
            "git",
            "-C",
            str(checkout),
            "grep",
            "-n",
            "-e",
            "payload_args_t",
            "-e",
            "payloadout",
            "--",
        ],
        allowed=(0, 1),
    )
    return sorted(line for line in output.splitlines() if line.strip())


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--sdk-source", type=Path, required=True)
    parser.add_argument("--sdk-install", type=Path, required=True)
    parser.add_argument("--objdump", type=Path, required=True)
    parser.add_argument("--readelf", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()

    root = args.root.resolve(strict=True)
    sdk_source = args.sdk_source.resolve(strict=True)
    sdk_install = args.sdk_install.resolve(strict=True)
    sdk_head = run(["git", "-C", str(sdk_source), "rev-parse", "HEAD"]).strip()
    if sdk_head != EXPECTED_SDK_COMMIT:
        raise ValueError("SDK source does not match the pinned v0.41 commit")

    source_files = set(CRT_OBJECTS.values()) | {
        "Makefile.inc",
        "crt/Makefile",
        "crt/payload.h",
        "include/ps5/payload.h",
        "host/bin/prospero-clang",
        "host/bin/prospero-lld",
        "host/elf_x86_64.x",
        "host/Makefile",
        "host/toolchain/prospero.cmake",
        "host/toolchain/prospero.mk",
        "host/toolchain/prospero.sh",
    }
    source_files.update(
        str(path.relative_to(sdk_source)).replace("\\", "/")
        for path in (sdk_source / "crt").glob("*.h")
    )
    source_hashes = {
        relative: sha256_file(sdk_source / relative)
        for relative in sorted(source_files)
    }

    makefile = (sdk_source / "crt/Makefile").read_text(encoding="utf-8")
    object_text_match = re.search(r"OBJ := (.+?)\n\n", makefile, re.DOTALL)
    if object_text_match is None:
        raise ValueError("could not parse CRT object inventory")
    makefile_objects = object_text_match.group(1).replace("\\\n", " ").split()
    if makefile_objects != list(CRT_OBJECTS):
        raise ValueError(f"CRT object inventory changed: {makefile_objects}")
    require(
        makefile,
        (
            "-ffreestanding -fno-builtin -nostdlib -fPIC",
            "-target x86_64-sie-ps5",
            "crt1.o: $(OBJ)",
            "$(LD) -m elf_x86_64 -r -o $@ $^",
        ),
        "crt/Makefile",
    )

    crt = (sdk_source / "crt/crt.c").read_text(encoding="utf-8")
    patch = (sdk_source / "crt/patch.c").read_text(encoding="utf-8")
    payload_header = (sdk_source / "include/ps5/payload.h").read_text(
        encoding="utf-8"
    )
    linker_script = (sdk_source / "host/elf_x86_64.x").read_text(encoding="utf-8")
    compiler_wrapper = (sdk_source / "host/bin/prospero-clang").read_text(
        encoding="utf-8"
    )
    require(
        crt,
        (
            "_start(payload_args_t *args)",
            "for(unsigned char* bss=__bss_start; bss<__bss_end; bss++)",
            "__crt_syscall_init(args)",
            "__kernel_init(args)",
            "__klog_init()",
            "__patch_init()",
            "__rtld_init()",
            "return payload_terminate()",
        ),
        "crt/crt.c",
    )
    require(
        patch,
        (
            "kernel_set_ucred_caps(pid, caps)",
            "kernel_set_ucred_attrs(pid, attrs)",
            "caps[15] |= 0x40",
            "attrs[3] |= 0x80",
            "kernel_copyin(&uaddr, kaddr + 0xf0, sizeof(uaddr))",
            "kernel_copyin(&uaddr, kaddr + 0xf8, sizeof(uaddr))",
        ),
        "crt/patch.c",
    )
    require(
        payload_header,
        (
            "int (*sys_dynlib_dlsym)(int, const char*, void*)",
            "int* rwpipe",
            "int* rwpair",
            "intptr_t kpipe_addr",
            "intptr_t kdata_base_addr",
            "int* payloadout",
        ),
        "include/ps5/payload.h",
    )
    require(
        linker_script,
        (
            "OUTPUT_ARCH(i386:x86-64)",
            "PROVIDE_HIDDEN (__bss_start = .)",
            "PROVIDE_HIDDEN (__bss_end = .)",
            ".init_array",
            ".fini_array",
            "PT_DYNAMIC",
        ),
        "host/elf_x86_64.x",
    )
    require(
        compiler_wrapper,
        (
            'LIBS_CRT="${PS5_PAYLOAD_SDK}/target/lib/crt1.o"',
            'if [[ "$ARG" == "-nostdlibc" ||',
            '"$ARG" == "-nodefaultlibs"',
            'if [[ "$ARG" == "-nostartfiles" ||',
            "-target x86_64-sie-ps5",
        ),
        "host/bin/prospero-clang",
    )

    crt1 = sdk_install / "target/lib/crt1.o"
    nm = sdk_install / "bin/llvm-nm"
    ar = sdk_install / "bin/llvm-ar"
    compiler = sdk_install / "bin/prospero-clang"
    linker = sdk_install / "bin/prospero-lld"
    for path in (crt1, nm, ar, compiler, linker, args.objdump, args.readelf):
        path.resolve(strict=True)

    nm_output = run([str(nm), "--format=posix", str(crt1)])
    functions, undefined = parse_symbols(nm_output)
    expected_undefined = {
        "_DYNAMIC",
        "__bss_end",
        "__bss_start",
        "__fini_array_end",
        "__fini_array_start",
        "__image_end",
        "__image_start",
        "__init_array_end",
        "__init_array_start",
        "main",
    }
    if undefined != expected_undefined:
        raise ValueError(f"crt1.o undefined symbol inventory changed: {undefined}")
    if not DANGEROUS_REACHABLE.issubset(functions):
        raise ValueError("crt1.o no longer contains the reviewed dangerous functions")
    if not LINKED_PROHIBITED.issubset(functions):
        raise ValueError("crt1.o linked prohibited-function inventory changed")

    empty_archives: dict[str, list[str]] = {}
    for filename in EMPTY_CRT_ARCHIVES:
        members = run([str(ar), "t", str(sdk_install / "target/lib" / filename)])
        empty_archives[filename] = members.splitlines()
        if empty_archives[filename]:
            raise ValueError(f"{filename} unexpectedly contains members")

    disassembly = run(
        [str(args.objdump), "-dr", "--no-show-raw-insn", str(crt1)]
    )
    all_edges = parse_callgraph(disassembly)
    reachable_functions, reachable_edges = reachable_callgraph(all_edges, functions)
    reachable_set = set(reachable_functions)
    if not DANGEROUS_REACHABLE.issubset(reachable_set):
        missing = sorted(DANGEROUS_REACHABLE - reachable_set)
        raise ValueError(f"disassembly call graph lost expected reachable writes: {missing}")

    section_table = run([str(args.readelf), "-SW", str(crt1)])
    relocation_table = run([str(args.readelf), "-Wr", str(crt1)])
    elf_header = run([str(args.readelf), "-h", str(crt1)])
    if ".text._start" not in section_table or ".rela.text._start" not in section_table:
        raise ValueError("crt1.o lacks the audited _start sections")
    if "__patch_init" not in relocation_table or "kernel_copyin" not in relocation_table:
        raise ValueError("crt1.o relocation evidence no longer exposes the write path")
    tls_sections = sorted(
        match.group(1)
        for line in section_table.splitlines()
        if (match := re.search(r"\]\s+(\.\S+)\s+\S+.*\sT\s", line)) is not None
    )

    with tempfile.TemporaryDirectory() as directory:
        trace_source = Path(directory) / "trace.c"
        trace_source.write_text("int main(void) { return 0; }\n", encoding="utf-8")
        default_trace = driver_trace(compiler, trace_source, freestanding=False)
        free_trace = driver_trace(compiler, trace_source, freestanding=True)
    for expected in ("crt1.o", '"-lc"', '"-lkernel_web"', '"-lSceLibcInternal"', '"-lSceNet"'):
        if expected not in default_trace:
            raise ValueError(f"default driver trace lacks {expected}")
    for forbidden in ("crt1.o", '"-lc"', '"-lkernel_web"', '"-lSceLibcInternal"', '"-lSceNet"'):
        if forbidden in free_trace:
            raise ValueError(f"freestanding driver trace retains {forbidden}")

    upstream = root / "work/upstream"
    loader_search: dict[str, list[str]] = {}
    for name in ("SDL", "RetroArch", "pacbrew-repo", "ps5-linux-loader"):
        loader_search[name] = git_grep_loader_contract(upstream / name)
    if any(loader_search.values()):
        raise ValueError("an unreviewed local loader caller appeared in the evidence set")

    document: dict[str, Any] = {
        "schema_version": 1,
        "audit_date": "2026-07-17",
        "decision": "BLOCKED",
        "scope": "phase-0.5-kernelwrite-free-startup-offline-only",
        "artifact": {
            "built": False,
            "execution_eligible": False,
            "reason": "safe loader return and cleanup are not proven",
            "sha256": None,
        },
        "sdk": {
            "release": "v0.41",
            "commit": EXPECTED_SDK_COMMIT,
            "target": "x86_64-sie-ps5",
            "source_hashes": source_hashes,
        },
        "tool_versions": {
            "prospero_clang": run([str(compiler), "--version"]).splitlines()[0],
            "prospero_lld": run([str(linker), "--version"]).splitlines()[0],
            "objdump": run([str(args.objdump), "--version"]).splitlines()[0],
            "readelf": run([str(args.readelf), "--version"]).splitlines()[0],
        },
        "startup_linkage": {
            "default_driver_additions": [
                "target/lib/crt1.o",
                "libc",
                "libkernel_web",
                "libSceLibcInternal",
                "libSceNet",
            ],
            "default_driver_trace_sha256": sha256_bytes(default_trace.encode()),
            "freestanding_flags": ["-nostartfiles", "-nodefaultlibs"],
            "freestanding_driver_additions": [],
            "freestanding_driver_trace_sha256": sha256_bytes(free_trace.encode()),
            "crt1_sha256": sha256_file(crt1),
            "installed_linker_script_sha256": sha256_file(
                sdk_install / "ldscripts/elf_x86_64.x"
            ),
            "installed_main_script_sha256": sha256_file(
                sdk_install / "target/lib/main.script"
            ),
            "installed_prx_script_sha256": sha256_file(
                sdk_install / "target/lib/prx.script"
            ),
            "crt1_composition": [
                {"object": obj, "source": source}
                for obj, source in CRT_OBJECTS.items()
            ],
            "empty_startup_archives": empty_archives,
            "crt1_undefined_symbols": sorted(undefined),
        },
        "crt1_static_evidence": {
            "disassembly_sha256": sha256_bytes(disassembly.encode()),
            "elf_header_sha256": sha256_bytes(elf_header.encode()),
            "section_table_sha256": sha256_bytes(section_table.encode()),
            "relocation_table_sha256": sha256_bytes(relocation_table.encode()),
            "relocation_count": len(re.findall(r"R_X86_64_", relocation_table)),
            "tls_sections": tls_sections,
            "reachable_functions": reachable_functions,
            "reachable_callgraph": reachable_edges,
            "reachable_prohibited_functions": sorted(
                DANGEROUS_REACHABLE & reachable_set
            ),
            "linked_prohibited_functions": sorted(LINKED_PROHIBITED & functions),
            "linked_but_not_startup_reachable": sorted(
                LINKED_PROHIBITED - reachable_set
            ),
            "source_token_inventory": {
                "GNM": any(
                    "sceGnm" in (sdk_source / source).read_text(encoding="utf-8")
                    for source in CRT_OBJECTS.values()
                ),
                "SDL": any(
                    "SDL_" in (sdk_source / source).read_text(encoding="utf-8")
                    for source in CRT_OBJECTS.values()
                ),
                "VideoOut": any(
                    "sceVideoOut" in (sdk_source / source).read_text(encoding="utf-8")
                    for source in CRT_OBJECTS.values()
                ),
                "module_loading": True,
                "network_socket_helper": True,
            },
        },
        "loader_evidence": {
            "sdk_readme_loader_candidates": [
                "ps5-payload-dev/elfldr",
                "cryonumb/elfloader via ps5-jar-loader",
                "shahrilnet/remote_lua_loader",
            ],
            "local_non_sdk_contract_hits": loader_search,
            "exact_loader_used_for_firmware_9_60_identified": False,
            "caller_source_present": False,
            "entry_argument_layout": "SAFE_SOURCE_FACT",
            "callee_first_argument_register_rdi": "SAFE_DISASSEMBLY_FACT",
            "caller_stack_alignment": "UNPROVEN",
            "payloadout_lifetime_and_writability": "UNPROVEN",
            "return_address_and_return_consumption": "UNPROVEN",
            "post_return_cleanup": "UNPROVEN",
            "crash_cleanup": "UNPROVEN",
            "pre_entry_loader_process_changes": "UNPROVEN",
        },
        "review": [
            {
                "component": "payload_args_t field order and 48-byte x86_64 layout",
                "status": "SAFE",
                "basis": "public SDK header plus _start payloadout access at offset 0x28",
            },
            {
                "component": "SDK v0.41 default crt1.o",
                "status": "UNSAFE",
                "basis": "source, relocation, and reachable disassembly prove kernel-write paths",
            },
            {
                "component": "-nostartfiles -nodefaultlibs removes SDK CRT/default libs",
                "status": "SAFE",
                "basis": "compiler -### trace only; no output ELF was produced",
            },
            {
                "component": "custom BSS zeroing implementation",
                "status": "SAFE",
                "basis": "linker symbols and byte-zero loop can be freestanding",
            },
            {
                "component": "loader entry stack and ABI preconditions",
                "status": "UNPROVEN",
                "basis": "callee disassembly does not prove caller behavior",
            },
            {
                "component": "return, exit, crash, and cleanup behavior",
                "status": "UNPROVEN",
                "basis": "no exact pinned public loader caller source is locally available",
            },
            {
                "component": "loader changes before _start",
                "status": "UNPROVEN",
                "basis": "payload args expose pre-existing kernel access but not how it was established",
            },
        ],
        "custom_artifact_verification": {
            "linker_map": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "disassembly": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "undefined_symbols": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "dt_needed": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "relocations": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "init_fini_arrays": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "tls": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "double_clean_build": "NOT_PERFORMED_BLOCKED_BEFORE_BUILD",
            "artifact_sha256": None,
        },
        "minimal_missing_evidence": [
            "the exact loader and pinned source commit used on firmware 9.60",
            "the caller instruction sequence that establishes RDI, RSP alignment, and return address",
            "payload_args_t allocation, payloadout lifetime, writability, and ownership",
            "the caller path after _start returns, including restoration and cleanup",
            "crash/fault behavior before and during _start",
            "all process, credential, syscall, module, and memory changes made before entry",
        ],
        "safe_alternatives": [
            "continue libchimera-gfx through mock and software backends",
            "develop a separately scoped Linux-on-PS5 backend as a long-term track",
        ],
    }
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    print(
        "Phase-0.5 startup audit: BLOCKED; loader return contract is unproven; "
        "no PS5 ELF built"
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (KeyError, OSError, TypeError, ValueError) as error:
        print(f"startup feasibility audit failed: {error}")
        raise SystemExit(1) from error