OpenFUT/fifa17-recon/tools/gdb_game_setup_context_trace.py

"""Hardware-only trace of engine game-setup context selection.

Captures the function that requests team/side, selector indices 1/0, selected
transient context objects, and the typed value getter. No client writes.
"""
from __future__ import annotations

import json
import os
import struct
import time
import traceback

import gdb

DISPATCH = 0x147060D00
SELECT_VALUE = 0x147572C50
CONTEXT_SELECTED = 0x1477C845D
_STATE = None


def _reg(name: str) -> int:
    return int(gdb.parse_and_eval(f"${name}"))


def _read(address: int, size: int) -> bytes | None:
    if not address or address < 0:
        return None
    try:
        return bytes(gdb.selected_inferior().read_memory(address, size))
    except gdb.error:
        return None


def _u64(address: int) -> int | None:
    data = _read(address, 8)
    return struct.unpack("<Q", data)[0] if data else None


def _i32(address: int) -> int | None:
    data = _read(address, 4)
    return struct.unpack("<i", data)[0] if data else None


def _thread() -> dict:
    thread = gdb.selected_thread()
    if thread is None:
        return {}
    return {"name": thread.name, "ptid": list(thread.ptid), "global_num": thread.global_num}


def _printable_pointers(address: int, data: bytes) -> dict:
    found = {}
    for offset in range(0, len(data) - 7, 8):
        pointer = struct.unpack_from("<Q", data, offset)[0]
        raw = _read(pointer, 128)
        if not raw:
            continue
        value = raw.split(b"\0", 1)[0]
        try:
            text = value.decode("utf-8")
        except UnicodeDecodeError:
            continue
        if len(text) >= 3 and all(char.isprintable() for char in text):
            found[hex(offset)] = {"pointer": pointer, "text": text}
    return found


class State:
    def __init__(self, path: str):
        self.path = path
        self.index = 0
        self.requested_indices = {}

    def log(self, kind: str, **payload):
        self.index += 1
        event = {"event": kind, "event_index": self.index, "time_unix": time.time(),
                 "thread": _thread(), **payload}
        with open(self.path, "a", encoding="utf-8") as handle:
            handle.write(json.dumps(event, sort_keys=True) + "\n")
            handle.flush(); os.fsync(handle.fileno())

    def key(self):
        return tuple(_thread().get("ptid", ()))


class HardwareBreakpoint(gdb.Breakpoint):
    def __init__(self, state: State, address: int):
        self.state = state
        self.address = address
        super().__init__(f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False)
        self.silent = True


class DispatchBreakpoint(HardwareBreakpoint):
    def stop(self):
        try:
            rsp = _reg("rsp")
            caller = _u64(rsp)
            self.state.log(
                "game_setup_dispatch_entry",
                caller_return_address=caller,
                caller_disassembly=(gdb.execute(f"x/12i 0x{caller-32:x}", to_string=True)
                                    if caller else None),
                backtrace=gdb.execute("bt 24", to_string=True),
            )
        except Exception as exc:
            self.state.log("trace_error", where="dispatch", error=str(exc), traceback=traceback.format_exc())
        return False


class SelectValueBreakpoint(HardwareBreakpoint):
    def stop(self):
        try:
            index = _reg("rcx") & 0xFFFFFFFF
            self.state.requested_indices[self.state.key()] = index
            self.state.log("context_value_request", index=index)
        except Exception as exc:
            self.state.log("trace_error", where="select_value", error=str(exc), traceback=traceback.format_exc())
        return False


class ContextSelectedBreakpoint(HardwareBreakpoint):
    def stop(self):
        try:
            index = _reg("rdi") & 0xFFFFFFFF
            context = _reg("rbx")
            data = _read(context, 0x80) or b""
            self.state.log(
                "context_selected",
                requested_index=self.state.requested_indices.get(self.state.key()),
                selector_index=index,
                context=context,
                type_flags=_i32(context + 8),
                value_i32=_i32(context + 0x10),
                value_qword=_u64(context + 0x10),
                context_hex=data.hex(),
                printable_pointers=_printable_pointers(context, data),
            )
        except Exception as exc:
            self.state.log("trace_error", where="context_selected", error=str(exc), traceback=traceback.format_exc())
        return False




def _on_exit(event):
    if _STATE is not None:
        _STATE.log("inferior_exited", detail=str(event))


def start_trace(log_path: str, _cards_base: int):
    global _STATE
    open(log_path, "w", encoding="utf-8").close()
    _STATE = State(log_path)
    points = {
        "dispatch": DispatchBreakpoint(_STATE, DISPATCH),
        "select_value": SelectValueBreakpoint(_STATE, SELECT_VALUE),
        "context_selected": ContextSelectedBreakpoint(_STATE, CONTEXT_SELECTED),
    }
    gdb.events.exited.connect(_on_exit)
    _STATE.log(
        "trace_armed",
        breakpoints={name: {"number": bp.number, "va": bp.address} for name, bp in points.items()},
        hardware_only=True,
        client_memory_writes=False,
    )