#!/usr/bin/env python3
"""s09 - grab a Ghidra address: did the firmware actually execute it?

Real scenario (docs/05-GHIDRA-BRIDGE.md, docs/13, docs/14 5; and
Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24, 2026-10-01): a static
hypothesis (the GPFIFO GET encodes entry VA ``0x101000``) was only accepted
after the decoded status-queue fault showed the engine actually fetched
``0x40000`` because the GET field dropped the high VA bits (fix ``6f9cbfa``).
The static/dynamic join is exactly this: Ghidra owns addresses and
decompilation, risc-v-gsp-rm-debugger answers "did it execute here".

Analog: export a Ghidra-style ``names.tsv`` from the demo ELF *with a base
offset* (the classic program-address vs runtime-address trap), load it with
the right bias, hook the raw Ghidra address, trigger the path, and write the
resulting ``greeneye_hits.tsv`` annotation back for Ghidra.
"""
from __future__ import annotations

import os

try:  # package import (tests) or direct script execution (run_all)
    from . import _common as C
except ImportError:
    import sys as _sys

    _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import _common as C

#: Pretend the Ghidra project was based at runtime - 0x1000 (the bias pitfall).
BIAS = 0x1000


def _ghidra_tsv(ge, out_path: str) -> dict:
    image = ge.load_elf(C.DEMO_ELF)
    rows = []
    for symbol in image.symbols:
        if not getattr(symbol, "is_function", False) or not symbol.name:
            continue
        program = symbol.value - BIAS
        rows.append((program, symbol.name, symbol.size))
    rows.sort()
    with open(out_path, "w", encoding="utf-8") as fh:
        for program, name, size in rows:
            fh.write(f"{program:#x}\t{name}\tsize={size}\n")
    return {"functions": len(rows), "bias": BIAS}


def run() -> dict:
    import greeneye as ge
    from greeneye.ghidra import export_hits

    backend = C.choose_backend("qemu")
    out = C.artifact_dir("s09_grabbing_a_ghidra_address")
    session_dir = C.session_dir(out)
    names_tsv = os.path.join(session_dir, "ghidra_names.tsv")
    facts = _ghidra_tsv(ge, names_tsv)
    runtime_rpc_fault = ge.load_elf(C.DEMO_ELF)
    addr_rpc_fault = None
    for symbol in runtime_rpc_fault.symbols:
        if symbol.name == "rpc_fault":
            addr_rpc_fault = symbol.value
    if addr_rpc_fault is None:
        raise C.DemoError("rpc_fault not found in demo ELF")
    ghidra_addr = addr_rpc_fault - BIAS

    commands = [
        f'write Ghidra names.tsv from ELF (program = runtime - {BIAS:#x})',
        f'SymbolDB.load_names_tsv(names, bias={BIAS:#x})',
        f'hook(ghidra_addr {ghidra_addr:#x} + bias)   # raw static address',
        "trigger mailbox cmd=6; hits recorded; export_hits -> greeneye_hits.tsv",
    ]
    scenario = C.scenario(
        ids=[
            "docs/05-GHIDRA-BRIDGE.md / docs/13: static/dynamic address join",
            "Handoff-vGPU-IO.md PART 91 ADDENDUM 9 cycle 24 (2026-10-01): GET fault at 0x40000 vs hypothesis 0x101000",
            "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 5: program vs runtime addresses and bias",
        ],
        baseline=(
            "cycle 24's decoded status-queue fault ended a run of eight "
            "supervised cycles (18-25, ~1 h 41 m); the static GET-encoding "
            "hypothesis was unproven until a live PC/fault address existed"
        ),
        projection=(
            "for the real image the same names.tsv + bias + hook loop runs once "
            "H2.2 exposes execution of the .fwimage; the bias arithmetic demoed "
            "here is exactly what docs/05 warns about"
        ),
        blocked=True,
        device="H2.2 device model to execute the real .fwimage",
    )
    C.render_banner(
        "s09_grabbing_a_ghidra_address",
        "Ghidra gave an address: does the firmware execute exactly there, and can the hit be annotated back?",
        "static/dynamic join for the GPFIFO GET fault (0x40000 vs 0x101000)",
        scenario["static_baseline"],
        commands,
    )

    session_dir = C.session_dir(out)
    session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=session_dir, name="vd-s09")
    try:
        session.symbols.load_names_tsv(names_tsv, bias=BIAS)
        ctrl = session.symbols.lookup("gsp_ctrl")
        resolved = session.resolve(ghidra_addr + BIAS)
        hits = []
        session.hook(ghidra_addr + BIAS, on_enter=lambda hit: hits.append(hit.addr))
        session.run(duration=0.25)
        session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 6)
        session.run(duration=0.25)
        hits_tsv = export_hits(session_dir, os.path.join(session_dir, "greeneye_hits.tsv"), symbols=session.symbols)
        payload = {
            "success": len(hits) >= 1 and resolved == addr_rpc_fault,
            "backend": backend,
            "ghidra_functions": facts["functions"],
            "ghidra_names_tsv": os.path.relpath(names_tsv, C.TOOL_ROOT),
            "bias": BIAS,
            "ghidra_addr": f"{ghidra_addr:#x}",
            "runtime_addr": f"{resolved:#x}",
            "program_addr_used_without_bias": f"{ghidra_addr:#x}",
            "hits": hits,
            "hits_tsv": os.path.relpath(hits_tsv, C.TOOL_ROOT),
            "question": "join a static Ghidra address to a runtime hit, with the bias handled explicitly",
            "answer": (
                f"Ghidra addr {ghidra_addr:#x} + bias {BIAS:#x} = runtime {resolved:#x}; "
                f"the hook fired {len(hits)}x at that exact address; export_hits wrote "
                "the annotated greeneye_hits.tsv - the static hypothesis is now a "
                "recorded dynamic fact (and using the program address raw would miss by 0x1000)"
            ),
            "success_criteria": {
                "pass": "the biased address resolves to the real function and the hook records the hit; TSV annotates it",
                "fail": "the hit lands at a different address or the TSV is empty",
            },
            "real_case": scenario,
        }
        with open(hits_tsv, "r", encoding="utf-8") as fh:
            payload["greeneye_hits_tsv"] = fh.read()
        C.write_json(os.path.join(out, "ghidra-join.json"), payload)
    finally:
        session.close()
    return payload


def main() -> int:
    return 0 if C.timed_demo(run, C.artifact_dir("s09_grabbing_a_ghidra_address"))["success"] else 1


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