#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Generate deterministic Phase-0.7 static ELF evidence without execution."""

from __future__ import annotations

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


EXPECTED_COMMITS = {
    "lifecycle": "fe08300339a13f899fb78ea404ada381a5cba87c",
    "loader": "197623058f509eddde18868dafcb92fdcac66464",
    "manager": "e23d94ff91233aa770e2342800c1467875bdef44",
}
EXPECTED_HASHES = {
    "lifecycle": "bfb4a5cc768e162fe4c2fddf41c3978e152722918a39085277fd172cb95a7182",
    "loader": "63e810982471eb40cae3a20aa9df9a0a02892f420e429874fae4e99aa400b561",
    "manager": "8fecf8241a46246eddbd21e8bb4d875f5d76f1f4f4c6a11384df1f131aa5e5b1",
}
EXPECTED_SIZES = {"lifecycle": 112680, "loader": 397000, "manager": 99560}
LIFECYCLE_IMPORTS = {"_exit", "sceKernelSendNotificationRequest"}
LIFECYCLE_NEEDED = {"libSceLibcInternal.sprx", "libkernel_web.sprx"}
LIFECYCLE_FORBIDDEN = re.compile(
    r"(sceGnm|VideoOut|SDL_|sceNet|socket|connect|dlopen|dlsym|"
    r"kernel_copyin|kernel_copyout|kernel_set_ucred|__patch_init)"
)
SENSITIVE_CATEGORIES = {
    "dynamic_loading": re.compile(
        r"(^|_)(dlopen|dlsym)$|^__(dlopen|dlsym)$|dynlib|"
        r"LoadStartModule|StopUnloadModule"
    ),
    "graphics_or_display": re.compile(
        r"sceGnm|[Vv]ideoOut|SDL_|[Gg]pu|[Mm][Mm][Ii][Oo]"
    ),
    "kernel_runtime_write": re.compile(
        r"__patch_init|kernel_copy(in|out)|kernel_set_ucred|"
        r"kernel_overlap_sockets|syscall.*(patch|set)|"
        r"(patch|set).*syscall"
    ),
    "network": re.compile(r"(^|_)(socket|connect|listen|accept)$|sceNet"),
    "ptrace_or_jit": re.compile(r"ptrace|PT_[A-Z]|[Jj][Ii][Tt]"),
}


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


def sha256_file(path: Path) -> str:
    return sha256_bytes(path.read_bytes())


def run(command: list[str]) -> str:
    result = subprocess.run(
        command,
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    return result.stdout.replace("\r\n", "\n")


def git_head(path: Path, expected: str, lifecycle_source: bool = False) -> str:
    git = [
        "git",
        "-c",
        "core.autocrlf=true",
        "-c",
        "core.fileMode=false",
        "-C",
        str(path),
    ]
    head = run([*git, "rev-parse", "HEAD"]).strip()
    if lifecycle_source:
        subprocess.run(
            [*git, "merge-base", "--is-ancestor", expected, head],
            check=True,
        )
        changed = subprocess.run(
            [
                *git,
                "diff",
                "--quiet",
                expected,
                "--",
                "samples/lifecycle_probe/main.c",
            ],
            check=False,
        )
        if changed.returncode != 0:
            raise ValueError(f"{path}: lifecycle source differs from {expected}")
        return expected
    if head != expected:
        raise ValueError(f"{path}: expected {expected}, got {head}")
    if run([*git, "status", "--porcelain"]):
        raise ValueError(f"{path}: source tree is dirty")
    return head


def parse_undefined(symbols: str) -> list[str]:
    imports: set[str] = set()
    for line in symbols.splitlines():
        match = re.search(r"\bU\s+(\S+)\s*$", line)
        if match:
            imports.add(match.group(1))
    return sorted(imports)


def parse_needed(dynamic: str) -> list[str]:
    return sorted(set(re.findall(r"Shared library: \[([^\]]+)\]", dynamic)))


def parse_callgraph(disassembly: str) -> list[dict[str, Any]]:
    current = "<outside-symbol>"
    edges: list[dict[str, Any]] = []
    for line in disassembly.splitlines():
        function = re.match(r"^[0-9a-fA-F]+\s+<([^>]+)>:$", line)
        if function:
            current = function.group(1)
            continue
        call = re.match(
            r"^\s*([0-9a-fA-F]+):.*\bcallq?\b\s+(.+?)\s*$", line
        )
        if not call:
            continue
        operand = call.group(2).strip()
        target_match = re.search(r"<([^>]+)>", operand)
        edges.append(
            {
                "address": f"0x{call.group(1).lower()}",
                "from": current,
                "indirect": "*" in operand or target_match is None,
                "target": target_match.group(1) if target_match else operand,
            }
        )
    return edges


def function_name(name: str) -> str:
    return re.sub(r"\+0x[0-9a-fA-F]+$", "", name)


def defined_functions(disassembly: str) -> set[str]:
    return {
        match.group(1)
        for match in re.finditer(
            r"^[0-9a-fA-F]+\s+<([^>]+)>:$", disassembly, re.MULTILINE
        )
    }


def map_symbols(linker_map: str) -> set[str]:
    symbols = set()
    pattern = re.compile(
        r"^\s*[0-9a-fA-F]+\s+[0-9a-fA-F]+\s+[0-9a-fA-F]+"
        r"\s+\d+\s+([A-Za-z_][A-Za-z0-9_.$@]*)\s*$"
    )
    for line in linker_map.splitlines():
        match = pattern.match(line)
        if match:
            symbols.add(match.group(1))
    return symbols


def direct_reachable(
    callgraph: list[dict[str, Any]], entrypoint: str
) -> set[str]:
    adjacency: dict[str, set[str]] = {}
    for edge in callgraph:
        if edge["indirect"]:
            continue
        source = function_name(edge["from"])
        target = function_name(edge["target"])
        adjacency.setdefault(source, set()).add(target)
    reachable = {entrypoint}
    pending = [entrypoint]
    while pending:
        source = pending.pop()
        for target in adjacency.get(source, set()):
            if target not in reachable:
                reachable.add(target)
                pending.append(target)
    return reachable


def sensitive_inventory(
    disassembly: str,
    linker_map: str,
    callgraph: list[dict[str, Any]],
    entrypoint: str,
) -> dict[str, Any]:
    defined = defined_functions(disassembly)
    mapped = map_symbols(linker_map)
    reachable = direct_reachable(callgraph, entrypoint)
    categories = {}
    for category, pattern in SENSITIVE_CATEGORIES.items():
        disassembly_linked = sorted(
            name for name in defined if pattern.search(name)
        )
        map_linked = sorted(name for name in mapped if pattern.search(name))
        categories[category] = {
            "directly_reachable_from_entrypoint": sorted(
                name for name in disassembly_linked if name in reachable
            ),
            "linked": sorted(set(disassembly_linked) | set(map_linked)),
            "linked_in_disassembly": disassembly_linked,
            "linked_in_linker_map": map_linked,
        }
    return {
        "categories": categories,
        "direct_call_reachability_available": entrypoint in defined,
        "direct_call_reachability_only": True,
        "entrypoint": entrypoint,
        "indirect_call_edges_retained_but_not_resolved": sum(
            1 for edge in callgraph if edge["indirect"]
        ),
    }


def normalize_map(data: str) -> str:
    return re.sub(
        r"/tmp/([A-Za-z0-9_]+)-[0-9a-f]{6}\.o",
        r"/tmp/\1-<deterministic-temp>.o",
        data.replace("\r\n", "\n"),
    )


def analyze(
    *,
    name: str,
    first: Path,
    second: Path,
    map_path: Path,
    nm: Path,
    readelf: Path,
    objdump: Path,
    output: Path,
) -> dict[str, Any]:
    first_bytes = first.read_bytes()
    second_bytes = second.read_bytes()
    if first_bytes != second_bytes:
        raise ValueError(f"{name}: the two clean builds are not byte-identical")
    digest = sha256_bytes(first_bytes)
    if digest != EXPECTED_HASHES[name] or len(first_bytes) != EXPECTED_SIZES[name]:
        raise ValueError(f"{name}: identity differs from the reviewed profile")

    dynamic = run([str(readelf), "-d", str(first)])
    symbols = run([str(nm), "-u", str(first)])
    headers = run([str(readelf), "-h", "-l", "-S", str(first)])
    relocations = run([str(readelf), "-r", str(first)])
    all_symbols = run([str(readelf), "-Ws", str(first)])
    disassembly = run([str(objdump), "-d", str(first)])
    callgraph = parse_callgraph(disassembly)
    normalized_map = normalize_map(map_path.read_text(encoding="utf-8"))
    sensitive = sensitive_inventory(
        disassembly, normalized_map, callgraph, "_start"
    )

    reports = {
        "callgraph": json.dumps(callgraph, indent=2, sort_keys=True) + "\n",
        "disassembly": disassembly,
        "dynamic": dynamic,
        "headers_sections": headers,
        "linker_map_normalized": normalized_map,
        "relocations": relocations,
        "symbols": all_symbols,
        "undefined": symbols,
    }
    report_hashes: dict[str, str] = {}
    for report_name, content in reports.items():
        report_path = output / f"{name}.{report_name}.txt"
        report_path.write_text(content, encoding="utf-8", newline="\n")
        report_hashes[report_name] = sha256_file(report_path)

    imports = parse_undefined(symbols)
    needed = parse_needed(dynamic)
    if name == "lifecycle":
        if set(imports) != LIFECYCLE_IMPORTS:
            raise ValueError(f"lifecycle: unexpected imports {imports}")
        if set(needed) != LIFECYCLE_NEEDED:
            raise ValueError(f"lifecycle: unexpected DT_NEEDED {needed}")
        forbidden_matches = sorted(set(LIFECYCLE_FORBIDDEN.findall(symbols)))
        if forbidden_matches:
            raise ValueError(f"lifecycle: forbidden import {forbidden_matches}")
        if b"phase07-fw960-v1" not in first_bytes or b"9.60" not in first_bytes:
            raise ValueError("lifecycle: firmware/build gates are absent")

    return {
        "byte_identical_clean_builds": True,
        "call_edges": callgraph,
        "dt_needed": needed,
        "filename": first.name,
        "imports": imports,
        "report_sha256": report_hashes,
        "sensitive_static_inventory": sensitive,
        "sha256": digest,
        "size": len(first_bytes),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--loader-source", type=Path, required=True)
    parser.add_argument("--manager-source", type=Path, required=True)
    parser.add_argument("--artifact-dir", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--manifest-output", type=Path, required=True)
    parser.add_argument("--nm", type=Path, required=True)
    parser.add_argument("--readelf", type=Path, required=True)
    parser.add_argument("--objdump", type=Path, required=True)
    args = parser.parse_args()

    root = args.root.resolve()
    sources = {
        "lifecycle": root,
        "loader": args.loader_source.resolve(),
        "manager": args.manager_source.resolve(),
    }
    source_commits = {
        name: git_head(
            path, EXPECTED_COMMITS[name], lifecycle_source=name == "lifecycle"
        )
        for name, path in sources.items()
    }
    artifact_dir = args.artifact_dir.resolve()
    output = args.output.resolve()
    output.mkdir(parents=True, exist_ok=True)
    artifacts: dict[str, Any] = {}
    for name, stem in {
        "lifecycle": "chimera-gfx-lifecycle-probe-phase07",
        "loader": "chimera-elfldr-phase07",
        "manager": "chimera-pldmgr-phase07",
    }.items():
        artifacts[name] = analyze(
            name=name,
            first=artifact_dir / f"{stem}-a.elf",
            second=artifact_dir / f"{stem}-b.elf",
            map_path=artifact_dir / f"{stem}-b.map",
            nm=args.nm.resolve(strict=True),
            readelf=args.readelf.resolve(strict=True),
            objdump=args.objdump.resolve(strict=True),
            output=output,
        )

    document = {
        "schema_version": 1,
        "decision": "READY_FOR_HARDENED_RUNTIME_DEPLOYMENT",
        "firmware": "9.60",
        "offline_only": True,
        "ps5_actions": {
            "connected": False,
            "installed": False,
            "transferred": False,
            "executed": False,
        },
        "source_commits": source_commits,
        "artifacts": artifacts,
        "forbidden_lifecycle_imports": [],
        "permanent_denylist_sha256": (
            "4be1c17b4964f2b68c39b5145bc4af4619c32512d60269ecf5c39728b390fa63"
        ),
    }
    machine_output = output / "phase-0.7-offline-audit.json"
    content = json.dumps(document, indent=2, sort_keys=True) + "\n"
    machine_output.write_text(content, encoding="utf-8", newline="\n")
    args.manifest_output.resolve().write_text(
        content, encoding="utf-8", newline="\n"
    )
    print(
        "Phase-0.7 audit passed: three byte-identical ELF pairs, exact "
        "firmware 9.60 lifecycle imports, and no PS5 action"
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, subprocess.CalledProcessError, ValueError) as error:
        print(f"Phase-0.7 artifact audit failed: {error}")
        raise SystemExit(1) from error