funman300
433a9b22dd
Add a fail-closed, fresh-process native trace workflow for the Offline Seasons
fixture-to-match-team boundary. The supervisor ignores UMU's short-lived
FIFA17.exe process, requires CardsDLL, verifies TracerPid and hardware arming,
rejects pre-existing records, structurally locates fixtures/final records, and
detaches cleanly after capture.
The four payloads reproduce the measured chain without client writes:
FUN_1800fc500
-> actual season vector, fixture index 0 / team 73
-> temporary [73,130000] pair (not the final record)
CardsDLL service -> engine 0x147c652ce
-> live final +0x14 writes at the 0x45c side stride
-> correct [73,130000], then local overwrite [130000,130000]
engine wrapper 0x147ce47e0
<- CardsDLL 0x180031861
<- CardsGameSetupAdapter local `teams` query result already 130000
All execute breakpoints and watchpoints are hardware-only. /proc/PID/mem is
opened rb. No INT3, write_memory, patch, game input, server behavior, or Rust
code. Locator uses zero-based --fixture-index (the live selector is 0 when
season/user.round is 1) and never filters on transient +0x18 handles.
389 lines
13 KiB
Python
389 lines
13 KiB
Python
"""GDB Python payload for read-only FIFA17 match-team writer tracing.
|
|
|
|
Loaded by trace_match_team_writer.py. Uses hardware execute breakpoints and a
|
|
4-byte hardware WRITE watchpoint only; never inserts INT3 and never writes game
|
|
memory.
|
|
|
|
Breakpoints (CardsDLL image VAs):
|
|
|
|
* FUN_1800fc500 entry -- derives output pair from RDX and arms *(int*)(rdx+4).
|
|
* 0x1800fc595 -- pre-write opponent lookup into pair[1].
|
|
* 0x1800fc5b8 -- mirrored pre-write opponent lookup into pair[0].
|
|
|
|
The dynamic watchpoint catches the exact write establishing pair[1], whether it
|
|
is the opponent lookup at 0x1800fc595 or the own-club store at 0x1800fc5a0.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import struct
|
|
import time
|
|
import traceback
|
|
|
|
import gdb
|
|
|
|
CARDS_IMAGE_BASE = 0x180000000
|
|
ENTRY_RVA = 0x0FC500
|
|
LOOKUP_TO_TEAM1_RVA = 0x0FC595
|
|
LOOKUP_TO_TEAM0_RVA = 0x0FC5B8
|
|
TEAM1_POST_PC_TO_WRITER = {
|
|
0x1800FC599: 0x1800FC595, # mov [r14+4],ecx
|
|
0x1800FC5A4: 0x1800FC5A0, # mov [r14+4],eax
|
|
}
|
|
|
|
_STATE = None
|
|
|
|
|
|
def _reg(name: str) -> int:
|
|
return int(gdb.parse_and_eval(f"${name}"))
|
|
|
|
|
|
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 _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 _u8(address: int) -> int | None:
|
|
data = _read(address, 1)
|
|
return data[0] if data else None
|
|
|
|
|
|
def _u32(address: int) -> int | None:
|
|
data = _read(address, 4)
|
|
return struct.unpack("<I", 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 _u64(address: int) -> int | None:
|
|
data = _read(address, 8)
|
|
return struct.unpack("<Q", data)[0] if data else None
|
|
|
|
|
|
def _cstring(address: int, maximum: int = 256) -> str | None:
|
|
data = _read(address, maximum)
|
|
if not data:
|
|
return None
|
|
return data.split(b"\0", 1)[0].decode("utf-8", "replace")
|
|
|
|
|
|
def _rtti_name(vtable: int, cards_base: int) -> str | None:
|
|
"""MSVC x64 RTTI name from vtable[-1] CompleteObjectLocator.
|
|
|
|
PE RVAs in the locator are module-relative. Failure is evidence-free and is
|
|
logged as null; no pointer is named from an offset coincidence.
|
|
"""
|
|
locator = _u64(vtable - 8) if vtable else None
|
|
if not locator:
|
|
return None
|
|
raw = _read(locator, 24)
|
|
if not raw:
|
|
return None
|
|
_signature, _offset, _cd_offset, type_rva, _hier_rva, self_rva = struct.unpack(
|
|
"<IIIiii", raw
|
|
)
|
|
if not (0 <= type_rva < 0x10000000 and 0 <= self_rva < 0x10000000):
|
|
return None
|
|
image_base = locator - self_rva
|
|
if abs(image_base - cards_base) > 0x100000:
|
|
return None
|
|
return _cstring(image_base + type_rva + 16)
|
|
|
|
|
|
def _object(address: int, cards_base: int) -> dict:
|
|
vtable = _u64(address) if address else None
|
|
return {
|
|
"address": address,
|
|
"vtable": vtable,
|
|
"vtable_image_va": (
|
|
CARDS_IMAGE_BASE + (vtable - cards_base)
|
|
if vtable and cards_base <= vtable < cards_base + 0x400000
|
|
else None
|
|
),
|
|
"rtti": _rtti_name(vtable, cards_base) if vtable else None,
|
|
}
|
|
|
|
|
|
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 _provenance(state, destination: int | None = None) -> dict:
|
|
"""Recover the candidate's live input chain without naming the objects."""
|
|
regs = _registers()
|
|
context = regs["rbx"]
|
|
output_pair = regs["r14"]
|
|
obj = regs["rbp"]
|
|
nested = _u64(obj + 0xB0) if obj else None
|
|
field_2e8 = nested + 0x2E8 if nested else None
|
|
source_base = _u64(field_2e8) if field_2e8 else None
|
|
participant_holder = regs["r12"]
|
|
participant = _u64(participant_holder) if participant_holder else None
|
|
index_70 = _u8(participant + 0x70) if participant else None
|
|
source_address = (
|
|
source_base + index_70 * 16
|
|
if source_base is not None and index_70 is not None
|
|
else None
|
|
)
|
|
source_bytes = _read(source_address, 16) if source_address else None
|
|
decoded = None
|
|
if source_bytes and len(source_bytes) == 16:
|
|
team_id, byte4, byte5, pad, word8, wordc = struct.unpack("<iBBHii", source_bytes)
|
|
decoded = {
|
|
"team_id": team_id,
|
|
"byte_4": byte4,
|
|
"byte_5": byte5,
|
|
"pad_6": pad,
|
|
"word_8": word8,
|
|
"word_c": wordc,
|
|
}
|
|
pair_bytes = _read(output_pair, 8) if output_pair else None
|
|
return {
|
|
"destination": destination,
|
|
"context": _object(context, state.cards_base),
|
|
"entry_context": _object(state.current_entry.get("context", 0), state.cards_base),
|
|
"output_pair": output_pair,
|
|
"entry_output_pair": state.current_entry.get("output_pair"),
|
|
"output_pair_bytes": pair_bytes.hex() if pair_bytes else None,
|
|
"output_team_id_0": _i32(output_pair) if output_pair else None,
|
|
"output_team_id_1": _i32(output_pair + 4) if output_pair else None,
|
|
"obj": _object(obj, state.cards_base),
|
|
"nested_at_obj_plus_b0": _object(nested or 0, state.cards_base),
|
|
"field_plus_2e8_address": field_2e8,
|
|
"source_array_base": source_base,
|
|
"participant_holder": participant_holder,
|
|
"participant": _object(participant or 0, state.cards_base),
|
|
"participant_plus_70": index_70,
|
|
"source_record_address": source_address,
|
|
"source_record_hex": source_bytes.hex() if source_bytes else None,
|
|
"source_record": decoded,
|
|
"registers": regs,
|
|
}
|
|
|
|
|
|
class State:
|
|
def __init__(self, log_path: str, cards_base: int):
|
|
self.log_path = log_path
|
|
self.cards_base = cards_base
|
|
self.current_entry: dict = {}
|
|
self.watchpoint = None
|
|
self.event_index = 0
|
|
|
|
def log(self, kind: str, **payload):
|
|
self.event_index += 1
|
|
event = {
|
|
"event": kind,
|
|
"event_index": self.event_index,
|
|
"time_unix": time.time(),
|
|
"thread": _thread(),
|
|
**payload,
|
|
}
|
|
with open(self.log_path, "a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(event, sort_keys=True) + "\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
|
|
|
|
class Team1Watchpoint(gdb.Breakpoint):
|
|
def __init__(self, state: State, address: int):
|
|
self.state = state
|
|
self.address = address
|
|
super().__init__(
|
|
f"*(int*)0x{address:x}",
|
|
type=gdb.BP_WATCHPOINT,
|
|
wp_class=gdb.WP_WRITE,
|
|
internal=False,
|
|
)
|
|
self.silent = True
|
|
|
|
def stop(self):
|
|
try:
|
|
pc = _reg("rip")
|
|
image_pc = CARDS_IMAGE_BASE + (pc - self.state.cards_base)
|
|
writer = TEAM1_POST_PC_TO_WRITER.get(image_pc)
|
|
source_value = None
|
|
if writer == 0x1800FC595:
|
|
source_value = _reg("rcx") & 0xFFFFFFFF
|
|
elif writer == 0x1800FC5A0:
|
|
source_value = _reg("rax") & 0xFFFFFFFF
|
|
self.state.log(
|
|
"team1_write_post",
|
|
watch_address=self.address,
|
|
value=_i32(self.address),
|
|
stopped_pc=pc,
|
|
stopped_image_va=image_pc,
|
|
writer_image_va=writer,
|
|
source_value=source_value,
|
|
disassembly=gdb.execute("x/10i $pc-32", to_string=True),
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
provenance=_provenance(self.state, self.address),
|
|
)
|
|
if writer is not None:
|
|
# The output pair is a short-lived stack buffer. Leaving the
|
|
# watchpoint active after the candidate's exact write produced
|
|
# 114k unrelated events when that stack memory was reused.
|
|
# The two hardware lookup breakpoints remain armed, so disabling
|
|
# only this completed one-shot watch loses no provenance.
|
|
self.enabled = False
|
|
self.state.log(
|
|
"team1_watchpoint_disabled",
|
|
watch_address=self.address,
|
|
reason="candidate exact write captured",
|
|
)
|
|
except Exception as exc: # GDB must continue even if evidence rendering fails.
|
|
self.state.log("trace_error", where="team1_watchpoint", error=str(exc),
|
|
traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
class EntryBreakpoint(gdb.Breakpoint):
|
|
def __init__(self, state: State, address: int):
|
|
self.state = state
|
|
super().__init__(
|
|
f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False
|
|
)
|
|
self.silent = True
|
|
|
|
def stop(self):
|
|
try:
|
|
context, output_pair = _reg("rcx"), _reg("rdx")
|
|
self.state.current_entry = {
|
|
"context": context,
|
|
"output_pair": output_pair,
|
|
"entry_thread": _thread(),
|
|
}
|
|
if self.state.watchpoint is not None:
|
|
try:
|
|
self.state.watchpoint.delete()
|
|
except gdb.error:
|
|
pass
|
|
initial = _i32(output_pair + 4)
|
|
self.state.watchpoint = Team1Watchpoint(self.state, output_pair + 4)
|
|
self.state.log(
|
|
"candidate_entry",
|
|
entry_image_va=0x1800FC500,
|
|
context=_object(context, self.state.cards_base),
|
|
output_pair=output_pair,
|
|
team_id_1_address=output_pair + 4,
|
|
team_id_1_initial=initial,
|
|
watchpoint_number=self.state.watchpoint.number,
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
registers=_registers(),
|
|
)
|
|
self.state.log(
|
|
"team1_watchpoint_armed",
|
|
watch_address=output_pair + 4,
|
|
watchpoint_number=self.state.watchpoint.number,
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="candidate_entry", error=str(exc),
|
|
traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
class LookupStoreBreakpoint(gdb.Breakpoint):
|
|
def __init__(self, state: State, address: int, image_va: int, destination_offset: int):
|
|
self.state = state
|
|
self.image_va = image_va
|
|
self.destination_offset = destination_offset
|
|
super().__init__(
|
|
f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False
|
|
)
|
|
self.silent = True
|
|
|
|
def stop(self):
|
|
try:
|
|
destination = _reg("r14") + self.destination_offset
|
|
self.state.log(
|
|
"opponent_lookup_store_pre",
|
|
writer_image_va=self.image_va,
|
|
destination=destination,
|
|
destination_offset=self.destination_offset,
|
|
source_register="ecx",
|
|
source_value=_reg("rcx") & 0xFFFFFFFF,
|
|
disassembly=gdb.execute("x/5i $pc", to_string=True),
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
provenance=_provenance(self.state, destination),
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="lookup_store", 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):
|
|
"""Called from the supervisor's gdb command file after attach."""
|
|
global _STATE
|
|
open(log_path, "w", encoding="utf-8").close()
|
|
_STATE = State(log_path, cards_base)
|
|
entry = EntryBreakpoint(_STATE, cards_base + ENTRY_RVA)
|
|
lookup_team1 = LookupStoreBreakpoint(
|
|
_STATE,
|
|
cards_base + LOOKUP_TO_TEAM1_RVA,
|
|
0x1800FC595,
|
|
4,
|
|
)
|
|
lookup_team0 = LookupStoreBreakpoint(
|
|
_STATE,
|
|
cards_base + LOOKUP_TO_TEAM0_RVA,
|
|
0x1800FC5B8,
|
|
0,
|
|
)
|
|
gdb.events.exited.connect(_on_exit)
|
|
_STATE.log(
|
|
"trace_armed",
|
|
cards_base=cards_base,
|
|
breakpoints={
|
|
"candidate_entry": {"number": entry.number, "image_va": 0x1800FC500},
|
|
"lookup_to_team1": {
|
|
"number": lookup_team1.number,
|
|
"image_va": 0x1800FC595,
|
|
},
|
|
"lookup_to_team0": {
|
|
"number": lookup_team0.number,
|
|
"image_va": 0x1800FC5B8,
|
|
},
|
|
},
|
|
hardware_only=True,
|
|
client_memory_writes=False,
|
|
)
|