Add FIFA17 game-setup context tracers

This commit is contained in:
funman300
2026-08-25 18:29:55 +00:00
parent 433a9b22dd
commit 5181e103dc
3 changed files with 656 additions and 0 deletions
@@ -0,0 +1,225 @@
"""Hardware-only origin trace for the exact SetTeam team context.
Matches the typed integer context pointer selected by SetTeam to the constructor
invocation that produced it. No client memory writes.
"""
from __future__ import annotations
from collections import deque
import json
import os
import struct
import time
import traceback
import gdb
CONTEXT_REUSE = 0x1477C17FC
CONTEXT_ALLOCATED = 0x1477C18C1
SET_TEAM_STUB = 0x147060A80
LOCKED_SETTER_RETURN = 0x1477C2415
CONTEXT_STACK_COUNT = 0x144BCEDA0
CONTEXT_STACK_ARRAY = 0x144BCEDA8
INTERESTING = {73, 130000, 130001}
_STATE = None
def _reg(name):
return int(gdb.parse_and_eval(f"${name}"))
def _read(address, size):
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):
data = _read(address, 8)
return struct.unpack("<Q", data)[0] if data else None
def _i32(address):
data = _read(address, 4)
return struct.unpack("<i", data)[0] if data else None
def _thread():
thread = gdb.selected_thread()
if thread is None:
return {}
return {"name": thread.name, "ptid": list(thread.ptid), "global_num": thread.global_num}
class State:
def __init__(self, path):
self.path = path
self.index = 0
self.total_constructor_hits = 0
self.interesting_constructor_hits = 0
self.pending_allocations = {}
self.origins = deque(maxlen=4096)
def log(self, kind, **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 thread_key(self):
return tuple(_thread().get("ptid", ()))
def remember_origin(self, context, origin):
if context:
self.origins.append({**origin, "context": context})
def find_origin(self, context):
return next((origin for origin in reversed(self.origins)
if origin["context"] == context), None)
class HardwareBreakpoint(gdb.Breakpoint):
def __init__(self, state, address):
self.state = state
self.address = address
super().__init__(f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False)
self.silent = True
class ContextReuseBreakpoint(HardwareBreakpoint):
def stop(self):
self.state.total_constructor_hits += 1
try:
value = _reg("rcx") & 0xFFFFFFFF
if value not in INTERESTING:
return False
self.state.interesting_constructor_hits += 1
rsp = _reg("rsp")
direct_return = _u64(rsp + 0x28)
origin = {
"value": value,
"direct_return_address": direct_return,
"upstream_return_address": (
_u64(rsp + 0x68)
if direct_return == LOCKED_SETTER_RETURN
else direct_return
),
"constructor_stack_hex": (_read(rsp, 0x100) or b"").hex(),
"constructor_hit": self.state.total_constructor_hits,
}
context = _reg("rax")
if context:
self.state.remember_origin(context, origin)
else:
self.state.pending_allocations[self.state.thread_key()] = origin
except Exception as exc:
self.state.log(
"trace_error",
where="context_reuse",
error=str(exc),
traceback=traceback.format_exc(),
)
return False
class ContextAllocatedBreakpoint(HardwareBreakpoint):
def stop(self):
try:
origin = self.state.pending_allocations.pop(self.state.thread_key(), None)
if origin is not None:
self.state.remember_origin(_reg("rdx"), origin)
except Exception as exc:
self.state.log(
"trace_error",
where="context_allocated",
error=str(exc),
traceback=traceback.format_exc(),
)
return False
class SetTeamStubBreakpoint(HardwareBreakpoint):
def stop(self):
try:
count = _i32(CONTEXT_STACK_COUNT)
array = _u64(CONTEXT_STACK_ARRAY)
team_context = (
_u64(array + (count - 2) * 8)
if array and count is not None and count >= 2
else None
)
side_context = (
_u64(array + (count - 1) * 8)
if array and count is not None and count >= 1
else None
)
rsp = _reg("rsp")
self.state.log(
"set_team_stub_entry",
context_stack_count=count,
team_context=team_context,
team_context_hex=(_read(team_context, 0x40) or b"").hex(),
team_value=_i32(team_context + 0x10) if team_context else None,
side_context=side_context,
side_value=_i32(side_context + 0x10) if side_context else None,
matched_origin=self.state.find_origin(team_context),
caller_return_address=_u64(rsp),
entry_registers={
name: _reg(name)
for name in ("rcx", "rdx", "r8", "r9")
},
backtrace=gdb.execute("bt 32", to_string=True),
total_constructor_hits=self.state.total_constructor_hits,
interesting_constructor_hits=self.state.interesting_constructor_hits,
)
except Exception as exc:
self.state.log(
"trace_error",
where="set_team_stub",
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),
total_constructor_hits=_STATE.total_constructor_hits,
interesting_constructor_hits=_STATE.interesting_constructor_hits,
)
def start_trace(log_path, _cards_base):
global _STATE
open(log_path, "w", encoding="utf-8").close()
_STATE = State(log_path)
points = {
"context_reuse": ContextReuseBreakpoint(_STATE, CONTEXT_REUSE),
"context_allocated": ContextAllocatedBreakpoint(_STATE, CONTEXT_ALLOCATED),
"set_team_stub": SetTeamStubBreakpoint(_STATE, SET_TEAM_STUB),
}
gdb.events.exited.connect(_on_exit)
_STATE.log(
"trace_armed",
breakpoints={
name: {"number": point.number, "va": point.address}
for name, point in points.items()
},
hardware_only=True,
client_memory_writes=False,
matching="exact_context_pointer",
)
@@ -0,0 +1,167 @@
"""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,
)
@@ -0,0 +1,264 @@
"""Hardware-only trace of CardsGameSetupAdapter query 13 and overwrite input.
Breakpoints:
FUN_180031340 entry incoming teamId/side/context
0x18003148f pre-call query id, selector, output/count pointers
0x180031495 post-call complete 48-byte records and count
0x180031861 submit original incoming teamId sent to engine
This proves whether query 13 influences the overwrite. No INT3/software
breakpoints, client writes, or game input.
"""
from __future__ import annotations
import json
import os
import struct
import time
import traceback
import gdb
CARDS_IMAGE_BASE = 0x180000000
ENTRY = 0x180031340
QUERY_PRE = 0x18003148F
QUERY_POST = 0x180031495
SUBMIT = 0x180031861
MAX_RECORDS = 100
RECORD_SIZE = 48
_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 or size < 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 _registers() -> dict:
names = (
"rax", "rbx", "rcx", "rdx", "rsi", "rdi", "rbp", "rsp",
"r8", "r9", "r10", "r11", "r12", "r13", "r14", "r15", "rip",
)
return {name: _reg(name) for name in names}
def _printable_pointer(pointer: int) -> str | None:
data = _read(pointer, 96)
if not data:
return None
raw = data.split(b"\0", 1)[0]
if len(raw) < 3:
return None
try:
text = raw.decode("utf-8")
except UnicodeDecodeError:
return None
return text if all(char.isprintable() for char in text) else None
def _decode_record(data: bytes, address: int) -> dict:
words = list(struct.unpack("<12i", data))
qwords = list(struct.unpack("<6Q", data))
strings = {}
for index, pointer in enumerate(qwords):
text = _printable_pointer(pointer)
if text:
strings[f"qword_{index}"] = {"pointer": pointer, "text": text}
interesting = {
str(value): [index * 4 for index, word in enumerate(words) if word == value]
for value in (73, 240, 241, 243, 130000, 130001)
if value in words
}
return {
"address": address,
"hex": data.hex(),
"i32": words,
"u32": [value & 0xFFFFFFFF for value in words],
"f32": list(struct.unpack("<12f", data)),
"qwords": qwords,
"strings": strings,
"interesting_values": interesting,
}
class State:
def __init__(self, path: str, cards_base: int):
self.path = path
self.cards_base = cards_base
self.index = 0
self.calls = {}
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 thread_key(self):
return tuple(_thread().get("ptid", ()))
class HardwareBreakpoint(gdb.Breakpoint):
def __init__(self, state: State, image_va: int):
self.state = state
self.image_va = image_va
address = state.cards_base + (image_va - CARDS_IMAGE_BASE)
super().__init__(f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False)
self.silent = True
class EntryBreakpoint(HardwareBreakpoint):
def stop(self):
try:
self.state.log(
"game_setup_entry",
incoming_context=_reg("rcx"),
incoming_side=_reg("rdx") & 0xFFFFFFFF,
incoming_team_id=_reg("r8") & 0xFFFFFFFF,
incoming_r9=_reg("r9"),
registers=_registers(),
backtrace=gdb.execute("bt 24", to_string=True),
)
except Exception as exc:
self.state.log("trace_error", where="entry", error=str(exc), traceback=traceback.format_exc())
return False
class QueryPreBreakpoint(HardwareBreakpoint):
def stop(self):
try:
rsp = _reg("rsp")
adapter = _reg("rcx")
vtable = _u64(adapter)
count_pointer = _u64(rsp + 0x20)
state = {
"adapter": adapter,
"adapter_vtable": vtable,
"query_target": _u64(vtable + 0xE0) if vtable else None,
"query_id": _reg("rdx") & 0xFFFFFFFF,
"selector": _reg("r8") & 0xFFFFFFFF,
"output_buffer": _reg("r9"),
"count_pointer": count_pointer,
"sixth_argument": _u64(rsp + 0x28),
"count_before": _i32(count_pointer) if count_pointer else None,
"saved_incoming_team_id": _i32(rsp + 0x34),
"saved_side": _i32(rsp + 0x50),
"saved_engine_context": _u64(rsp + 0x68),
"adapter_prefix_hex": (_read(adapter, 0x100) or b"").hex(),
}
self.state.calls[self.state.thread_key()] = state
self.state.log(
"query13_pre",
**state,
registers=_registers(),
backtrace=gdb.execute("bt 24", to_string=True),
)
except Exception as exc:
self.state.log("trace_error", where="query_pre", error=str(exc), traceback=traceback.format_exc())
return False
class QueryPostBreakpoint(HardwareBreakpoint):
def stop(self):
try:
state = self.state.calls.get(self.state.thread_key(), {})
count_pointer = state.get("count_pointer")
output = state.get("output_buffer")
count = _i32(count_pointer) if count_pointer else None
safe_count = min(max(count or 0, 0), MAX_RECORDS)
records = []
for index in range(safe_count):
address = output + index * RECORD_SIZE
data = _read(address, RECORD_SIZE)
if data and len(data) == RECORD_SIZE:
records.append(_decode_record(data, address))
self.state.log(
"query13_post",
query_state=state,
count_after=count,
records=records,
saved_incoming_team_id_after=_i32(_reg("rsp") + 0x34),
saved_side_after=_i32(_reg("rsp") + 0x50),
registers=_registers(),
)
except Exception as exc:
self.state.log("trace_error", where="query_post", error=str(exc), traceback=traceback.format_exc())
return False
class SubmitBreakpoint(HardwareBreakpoint):
def stop(self):
try:
rsp = _reg("rsp")
self.state.log(
"game_setup_submit",
submitted_team_id=_reg("r8") & 0xFFFFFFFF,
submitted_side=_reg("rdx") & 0xFFFFFFFF,
engine_context=_reg("rcx"),
saved_incoming_team_id=_i32(rsp + 0x34),
saved_side=_i32(rsp + 0x50),
registers=_registers(),
backtrace=gdb.execute("bt 24", to_string=True),
)
except Exception as exc:
self.state.log("trace_error", where="submit", 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, cards_base)
points = {
"entry": EntryBreakpoint(_STATE, ENTRY),
"query_pre": QueryPreBreakpoint(_STATE, QUERY_PRE),
"query_post": QueryPostBreakpoint(_STATE, QUERY_POST),
"submit": SubmitBreakpoint(_STATE, SUBMIT),
}
gdb.events.exited.connect(_on_exit)
_STATE.log(
"trace_armed",
breakpoints={name: {"number": bp.number, "image_va": bp.image_va} for name, bp in points.items()},
record_size=RECORD_SIZE,
hardware_only=True,
client_memory_writes=False,
)