diff --git a/fifa17-recon/tools/gdb_engine_overwrite_entry_trace.py b/fifa17-recon/tools/gdb_engine_overwrite_entry_trace.py new file mode 100644 index 0000000..486cd58 --- /dev/null +++ b/fifa17-recon/tools/gdb_engine_overwrite_entry_trace.py @@ -0,0 +1,126 @@ +"""Hardware-only trace of the engine-local overwrite wrapper entry. + +Breaks before the prologue of FUN_147ce47e0, where [rsp] is the exact direct +caller return address and R8D is the team ID later written to the final match +record. This closes the one frame Wine PE unwinding could not recover. + +No INT3/software breakpoints. No client memory writes. +""" +from __future__ import annotations + +import json +import os +import struct +import time +import traceback + +import gdb + +WRAPPER_VA = 0x147CE47E0 +_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(" 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} + + +class State: + def __init__(self, path: str): + self.path = path + self.index = 0 + + def log(self, kind: str, **payload): + self.index += 1 + thread = _thread() + 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()) + + +class WrapperBreakpoint(gdb.Breakpoint): + def __init__(self, state: State): + self.state = state + super().__init__( + f"*0x{WRAPPER_VA:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False + ) + self.silent = True + + def stop(self): + try: + stack = _reg("rsp") + caller_return = _u64(stack) + self.state.log( + "engine_overwrite_wrapper_entry", + wrapper_va=WRAPPER_VA, + caller_return_address=caller_return, + source_team_id=_reg("r8") & 0xFFFFFFFF, + side_argument=_reg("rdx") & 0xFFFFFFFF, + registers=_registers(), + caller_disassembly=( + gdb.execute(f"x/12i 0x{caller_return - 32:x}", to_string=True) + if caller_return else None + ), + backtrace=gdb.execute("bt 32", to_string=True), + ) + except Exception as exc: + self.state.log( + "trace_error", where="engine_overwrite_wrapper", 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) + breakpoint = WrapperBreakpoint(_STATE) + gdb.events.exited.connect(_on_exit) + _STATE.log( + "trace_armed", + breakpoints={"engine_overwrite_wrapper": {"number": breakpoint.number, "va": WRAPPER_VA}}, + hardware_only=True, + client_memory_writes=False, + ) diff --git a/fifa17-recon/tools/gdb_final_team_writer_trace.py b/fifa17-recon/tools/gdb_final_team_writer_trace.py new file mode 100644 index 0000000..d0908e4 --- /dev/null +++ b/fifa17-recon/tools/gdb_final_team_writer_trace.py @@ -0,0 +1,226 @@ +"""GDB payload for the LIVE-PROVEN engine match-team +0x14 writer. + +READ-ONLY hardware debug only: + + 0x147c652ce mov dword [rdx + rcx + 0x44], r8d + +At the first team-like source value, derives both fixed-stride record fields +from live RCX and arms 4-byte WRITE watchpoints on: + + teamId A = rcx + 0x44 + teamId B = rcx + 0x44 + 0x45c + +The execute breakpoint records the intended source value before every call. The +watchpoints then capture both the expected write and any later overwrite, even +if the overwrite comes from a different function. + +No INT3/software breakpoints. No client memory writes. +""" +from __future__ import annotations + +import json +import os +import struct +import time +import traceback + +import gdb + +WRITER_VA = 0x147C652CE +POST_WRITER_VA = 0x147C652D3 +SIDE_STRIDE = 0x45C +TEAM_FIELD_OFF = 0x44 +RECORD_FIELD_OFF = 0x14 +TEAM_LIKE = {73, 240, 241, 243, 130000, 130001} + +_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: + return None + try: + return bytes(gdb.selected_inferior().read_memory(address, size)) + except gdb.error: + return None + + +def _i32(address: int) -> int | None: + data = _read(address, 4) + return struct.unpack(" 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} + + +class State: + def __init__(self, log_path: str): + self.log_path = log_path + self.event_index = 0 + self.engine_base = None + self.watch_a = None + self.watch_b = None + + 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()) + + def arm_fields(self, engine_base: int): + if self.engine_base == engine_base and self.watch_a and self.watch_b: + return + for watchpoint in (self.watch_a, self.watch_b): + if watchpoint is not None: + try: + watchpoint.delete() + except gdb.error: + pass + self.engine_base = engine_base + self.watch_a = TeamFieldWatchpoint(self, 0, engine_base + TEAM_FIELD_OFF) + self.watch_b = TeamFieldWatchpoint( + self, 1, engine_base + TEAM_FIELD_OFF + SIDE_STRIDE + ) + self.log( + "team_field_watchpoints_armed", + engine_base=engine_base, + team_id_a_address=self.watch_a.address, + team_id_b_address=self.watch_b.address, + watchpoint_a=self.watch_a.number, + watchpoint_b=self.watch_b.number, + ) + + +class TeamFieldWatchpoint(gdb.Breakpoint): + def __init__(self, state: State, side: int, address: int): + self.state = state + self.side = side + 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") + writer = WRITER_VA if pc == POST_WRITER_VA else None + record_start = self.address - RECORD_FIELD_OFF + record = _read(record_start, 0x7C) + self.state.log( + "final_team_field_write_post", + side=self.side, + watch_address=self.address, + value=_i32(self.address), + stopped_pc=pc, + writer_va=writer, + record_start=record_start, + record_hex=record.hex() if record else None, + registers=_registers(), + disassembly=gdb.execute("x/12i $pc-32", to_string=True), + backtrace=gdb.execute("bt 32", to_string=True), + ) + except Exception as exc: + self.state.log( + "trace_error", + where="team_field_watchpoint", + error=str(exc), + traceback=traceback.format_exc(), + ) + return False + + +class FinalWriterBreakpoint(gdb.Breakpoint): + def __init__(self, state: State): + self.state = state + super().__init__( + f"*0x{WRITER_VA:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False + ) + self.silent = True + + def stop(self): + try: + engine_base = _reg("rcx") + side_offset = _reg("rdx") + source_value = _reg("r8") & 0xFFFFFFFF + if source_value in TEAM_LIKE: + self.state.arm_fields(engine_base) + destination = engine_base + side_offset + TEAM_FIELD_OFF + side = side_offset // SIDE_STRIDE if side_offset in (0, SIDE_STRIDE) else None + self.state.log( + "final_writer_pre", + instruction_va=WRITER_VA, + engine_base=engine_base, + side_offset=side_offset, + side=side, + destination=destination, + record_start=destination - RECORD_FIELD_OFF, + source_register="r8d", + source_value=source_value, + prior_value=_i32(destination), + team_id_a_address=engine_base + TEAM_FIELD_OFF, + team_id_b_address=engine_base + TEAM_FIELD_OFF + SIDE_STRIDE, + team_id_a_before=_i32(engine_base + TEAM_FIELD_OFF), + team_id_b_before=_i32(engine_base + TEAM_FIELD_OFF + SIDE_STRIDE), + registers=_registers(), + disassembly=gdb.execute("x/6i $pc", to_string=True), + backtrace=gdb.execute("bt 32", to_string=True), + ) + except Exception as exc: + self.state.log( + "trace_error", + where="final_writer", + 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) + writer = FinalWriterBreakpoint(_STATE) + gdb.events.exited.connect(_on_exit) + _STATE.log( + "trace_armed", + breakpoints={ + "final_writer": {"number": writer.number, "va": WRITER_VA}, + }, + side_stride=SIDE_STRIDE, + team_field_offset=TEAM_FIELD_OFF, + hardware_only=True, + client_memory_writes=False, + ) diff --git a/fifa17-recon/tools/gdb_match_team_writer_trace.py b/fifa17-recon/tools/gdb_match_team_writer_trace.py new file mode 100644 index 0000000..9db19e9 --- /dev/null +++ b/fifa17-recon/tools/gdb_match_team_writer_trace.py @@ -0,0 +1,388 @@ +"""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(" int | None: + data = _read(address, 4) + return struct.unpack(" int | None: + data = _read(address, 8) + return struct.unpack(" 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( + " 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(" 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: + return None + try: + return bytes(gdb.selected_inferior().read_memory(address, size)) + except gdb.error: + return None + + +def _pair(address: int) -> list[int] | None: + data = _read(address, 8) + return list(struct.unpack("<2i", data)) if data 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} + + +class State: + def __init__(self, log_path: str, cards_base: int): + self.log_path = log_path + self.cards_base = cards_base + 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 PairSubmitBreakpoint(gdb.Breakpoint): + def __init__(self, state: State, image_va: int, name: str, pointer_reg: str, index_reg: str): + self.state = state + self.image_va = image_va + self.name = name + self.pointer_reg = pointer_reg + self.index_reg = index_reg + 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 + + def stop(self): + try: + pointer = _reg(self.pointer_reg) + index = _reg(self.index_reg) & 0xFFFFFFFF + pair_base = pointer - index * 4 + self.state.log( + self.name, + instruction_image_va=self.image_va, + source_value=_reg("r8") & 0xFFFFFFFF, + side=_reg("rdx") & 0xFF, + engine_base=_reg("rcx"), + pair_pointer=pointer, + pair_index=index, + pair_base=pair_base, + pair=_pair(pair_base), + registers=_registers(), + disassembly=gdb.execute("x/5i $pc", to_string=True), + backtrace=gdb.execute("bt 24", to_string=True), + ) + except Exception as exc: + self.state.log( + "trace_error", where=self.name, error=str(exc), + traceback=traceback.format_exc() + ) + return False + + +class Mode76BuilderBreakpoint(gdb.Breakpoint): + def __init__(self, state: State): + self.state = state + address = state.cards_base + (MODE76_BUILDER - CARDS_IMAGE_BASE) + super().__init__(f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False) + self.silent = True + + def stop(self): + try: + self.state.log( + "mode76_builder_entry", + instruction_image_va=MODE76_BUILDER, + object=_reg("rcx"), + registers=_registers(), + backtrace=gdb.execute("bt 24", to_string=True), + ) + except Exception as exc: + self.state.log( + "trace_error", where="mode76_builder", 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) + breakpoints = {} + for image_va, (name, pointer_reg, index_reg) in SITES.items(): + bp = PairSubmitBreakpoint(_STATE, image_va, name, pointer_reg, index_reg) + breakpoints[name] = {"number": bp.number, "image_va": image_va} + builder = Mode76BuilderBreakpoint(_STATE) + breakpoints["mode76_builder"] = {"number": builder.number, "image_va": MODE76_BUILDER} + gdb.events.exited.connect(_on_exit) + _STATE.log( + "trace_armed", + breakpoints=breakpoints, + hardware_only=True, + client_memory_writes=False, + ) diff --git a/fifa17-recon/tools/offline_match_locator.py b/fifa17-recon/tools/offline_match_locator.py new file mode 100755 index 0000000..6aa6a5d --- /dev/null +++ b/fifa17-recon/tools/offline_match_locator.py @@ -0,0 +1,306 @@ +#!/usr/bin/env python3 +"""Read-only dynamic locator for FIFA17 Offline Seasons match state. + +Never relies on heap addresses or allocator handles. It identifies: + +* the 10 x 16-byte parsed fixture array from its complete wire-derived record + sequence (teamId/difficulty/roundId/rewardMult/coins), +* match-team records from the corrected invariant prefix (11,7,0,0,76), never + from the transient +0x18 handle, +* the match-config team pair from structural fields around it, not its team ids. + + offline_match_locator.py [pid] [--fixture-index 0] [--json] + offline_match_locator.py --selftest + +READ-ONLY: /proc//mem is opened 'rb'. No debugger and no game input. +""" +from __future__ import annotations + +import argparse +import glob +import json +import os +import re +import struct +import sys +from dataclasses import asdict, dataclass + +DEFAULT_TEAMS = (73, 240, 241, 243, 73, 240, 241, 243, 73, 240) +MATCH_HEADER = struct.pack("<5i", 11, 7, 0, 0, 76) +PARTICIPANT_PREFIX = struct.pack("<8i", -1, -2, -1, -2, -1, -2, -1, -2) +F01 = 0x3DCCCCCD + + +@dataclass +class Fixture: + address: int + selected_address: int + selected_index: int + selected_team_id: int + records: list[dict[str, int]] + + +@dataclass +class MatchTeam: + address: int + team_id: int + marker_18: int + marker_1c: int + xi: list[int] + substitutes: list[int] + + +@dataclass +class MatchConfig: + pair_address: int + team_id_0: int + team_id_1: int + player_count_0: int + player_count_1: int + + +def find_pids() -> list[int]: + """All live FIFA17.exe processes, largest resident set first. + + The UMU/Proton launch chain briefly creates a small process with the same + comm before the real game. Returning the first /proc glob match attached + the trace supervisor to that short-lived process and missed the match. + """ + found = [] + for directory in glob.glob("/proc/[0-9]*"): + try: + with open(os.path.join(directory, "comm")) as handle: + if handle.read().strip() != "FIFA17.exe": + continue + pid = int(os.path.basename(directory)) + with open(os.path.join(directory, "statm")) as handle: + resident_pages = int(handle.read().split()[1]) + found.append((resident_pages, pid)) + except (OSError, ValueError, IndexError): + continue + return [pid for _resident, pid in sorted(found, reverse=True)] + + +def find_pid() -> int | None: + pids = find_pids() + return pids[0] if pids else None + + +def fixture_bytes(teams: tuple[int, ...] = DEFAULT_TEAMS) -> bytes: + return b"".join( + struct.pack(" 512 * 1024 * 1024: + continue + if writable_anon_only and (perms[1] != "w" or path): + continue + yield lo, hi, perms, path + + +def _i32(buf: bytes, offset: int) -> int: + return struct.unpack_from(" list[Fixture]: + pattern = fixture_bytes() + found = [] + offset = buf.find(pattern) + while offset >= 0: + records = [] + for round_id in range(len(DEFAULT_TEAMS)): + at = offset + round_id * 16 + team_id, difficulty, parsed_round, _pad, reward_mult, coins = struct.unpack_from( + " list[MatchTeam]: + found = [] + offset = buf.find(MATCH_HEADER) + while offset >= 0: + if offset + 0x7C <= len(buf): + found.append( + MatchTeam( + address=base + offset, + team_id=_i32(buf, offset + 0x14), + marker_18=_i32(buf, offset + 0x18), + marker_1c=_i32(buf, offset + 0x1C), + xi=list(struct.unpack_from("<11i", buf, offset + 0x20)), + substitutes=list(struct.unpack_from("<12i", buf, offset + 0x4C)), + ) + ) + offset = buf.find(MATCH_HEADER, offset + 4) + return found + + +def _valid_config(buf: bytes, pair: int) -> bool: + required = pair + 0x50 + if pair < 0 or required > len(buf): + return False + return ( + tuple(struct.unpack_from("<4I", buf, pair + 0x1C)) == (F01, F01, F01, F01) + and _i32(buf, pair + 0x38) == 11 + and _i32(buf, pair + 0x3C) == 11 + and _i32(buf, pair + 0x40) == 0 + and _i32(buf, pair + 0x44) == 5 + ) + + +def scan_match_config_buffer(buf: bytes, base: int) -> list[MatchConfig]: + found = [] + offset = buf.find(PARTICIPANT_PREFIX) + while offset >= 0: + pair = offset + len(PARTICIPANT_PREFIX) + if _valid_config(buf, pair): + found.append( + MatchConfig( + pair_address=base + pair, + team_id_0=_i32(buf, pair), + team_id_1=_i32(buf, pair + 4), + player_count_0=_i32(buf, pair + 0x38), + player_count_1=_i32(buf, pair + 0x3C), + ) + ) + offset = buf.find(PARTICIPANT_PREFIX, offset + 4) + return found + + +def scan_process( + pid: int, + selected_index: int, + *, + include_fixture: bool = True, + writable_anon_only: bool = False, +) -> dict[str, list]: + result: dict[str, list] = {"fixtures": [], "match_teams": [], "match_configs": []} + with open(f"/proc/{pid}/mem", "rb", 0) as memory: + for lo, hi, _perms, _path in readable_regions( + pid, writable_anon_only=writable_anon_only + ): + try: + memory.seek(lo) + buf = memory.read(hi - lo) + except (OSError, ValueError, OverflowError): + continue + if include_fixture: + result["fixtures"].extend(scan_fixture_buffer(buf, lo, selected_index)) + result["match_teams"].extend(scan_match_team_buffer(buf, lo)) + result["match_configs"].extend(scan_match_config_buffer(buf, lo)) + return result + + +def selftest() -> None: + fixture = fixture_bytes() + team = bytearray(0x7C) + team[:20] = MATCH_HEADER + struct.pack_into(" int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("pid", nargs="?", type=int) + parser.add_argument("--fixture-index", type=int, default=0) + parser.add_argument("--json", action="store_true") + parser.add_argument("--selftest", action="store_true") + parser.add_argument("--writable-anon-only", action="store_true") + args = parser.parse_args() + if args.selftest: + selftest() + return 0 + pid = args.pid or find_pid() + if not pid: + print("FIFA17.exe not found", file=sys.stderr) + return 2 + if not 0 <= args.fixture_index < len(DEFAULT_TEAMS): + print("--fixture-index must be 0..9", file=sys.stderr) + return 2 + result = scan_process( + pid, + args.fixture_index, + writable_anon_only=args.writable_anon_only, + ) + serial = {key: [asdict(value) for value in values] for key, values in result.items()} + serial["pid"] = pid + if args.json: + print(json.dumps(serial, sort_keys=True)) + return 0 + print(f"pid={pid}") + for fixture in result["fixtures"]: + print( + f"fixture @0x{fixture.address:x}; selected index {fixture.selected_index} " + f"@0x{fixture.selected_address:x} teamId={fixture.selected_team_id}" + ) + for config in result["match_configs"]: + print( + f"match config pair @0x{config.pair_address:x}: " + f"[{config.team_id_0}, {config.team_id_1}]" + ) + for team in result["match_teams"]: + print( + f"match team @0x{team.address:x}: teamId={team.team_id} " + f"handles=[{team.marker_18}, {team.marker_1c}]" + ) + print( + f"counts: fixtures={len(result['fixtures'])} " + f"configs={len(result['match_configs'])} teams={len(result['match_teams'])}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/fifa17-recon/tools/trace_match_team_writer.py b/fifa17-recon/tools/trace_match_team_writer.py new file mode 100755 index 0000000..99646d9 --- /dev/null +++ b/fifa17-recon/tools/trace_match_team_writer.py @@ -0,0 +1,512 @@ +#!/usr/bin/env python3 +"""Supervise one hardware-only FIFA17 match-team writer capture. + +This is the robust fresh-client entry point. It waits for the largest-RSS +FIFA17.exe process that has CardsDLL loaded, attaches gdb before FUT navigation +can construct match teams, and loads a hardware-only GDB Python payload. + +The concurrent read-only structural locator proves when the fixture and final +match-team records exist. A zero-hit result is trusted only if gdb is still +alive, TracerPid is the gdb process, the payload reported `trace_armed`, no +records pre-existed the trace, and two final records then appeared. + +The default payload traces FUN_1800fc500 and derives a 4-byte teamId[1] +watchpoint from live RDX. Other payloads trace the final engine writer or its +caller; all expose the same `start_trace(log, cards_base)` entry point. + +No INT3/software breakpoints. No client memory writes. /proc//mem is opened +'rb'. The operator alone drives the game. + + trace_match_team_writer.py --status /tmp/mt-status.json \ + --trace /tmp/mt-trace.jsonl --gdb-log /tmp/mt-gdb.log --fixture-index 0 +""" +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess +import sys +import tempfile +import time +from dataclasses import asdict +from pathlib import Path + +from offline_match_locator import find_pids, scan_process + +CARDS_IMAGE_BASE = 0x180000000 +DEFAULT_TIMEOUT = 45 * 60 + + +def cards_base(pid: int) -> int | None: + try: + with open(f"/proc/{pid}/maps") as maps: + for line in maps: + if "CardsDLL_Win64_retail.dll" in line: + return int(line.split("-", 1)[0], 16) + except OSError: + pass + return None + + +def tracer_pid(pid: int) -> int | None: + try: + with open(f"/proc/{pid}/status") as status: + for line in status: + if line.startswith("TracerPid:"): + return int(line.split()[1]) + except OSError: + pass + return None + + +def target_state(pid: int) -> str | None: + try: + with open(f"/proc/{pid}/status") as status: + for line in status: + if line.startswith("State:"): + return line.split()[1] + except OSError: + pass + return None + + +def read_events(path: Path) -> list[dict]: + if not path.exists(): + return [] + events = [] + try: + with path.open(encoding="utf-8", errors="replace") as handle: + for line in handle: + try: + events.append(json.loads(line)) + except json.JSONDecodeError: + continue + except OSError: + return [] + return events + + +def event_counts(events: list[dict]) -> dict[str, int]: + counts: dict[str, int] = {} + for event in events: + kind = event.get("event", "unknown") + counts[kind] = counts.get(kind, 0) + 1 + return counts + + +class Status: + def __init__(self, path: Path, monitor_log: Path): + self.path = path + self.monitor_log = monitor_log + self.data: dict = {"started_unix": time.time(), "state": "starting"} + self.write() + + def write(self, **updates): + self.data.update(updates) + self.data["updated_unix"] = time.time() + temporary = self.path.with_suffix(self.path.suffix + ".tmp") + temporary.write_text(json.dumps(self.data, indent=2, sort_keys=True) + "\n") + os.replace(temporary, self.path) + + def log(self, message: str, **payload): + record = {"time_unix": time.time(), "message": message, **payload} + with self.monitor_log.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(record, sort_keys=True) + "\n") + handle.flush() + os.fsync(handle.fileno()) + print(message, flush=True) + + +def gdb_commands(pid: int, cards: int, payload: Path, trace: Path) -> str: + # Wine uses these signals for thread suspension/runtime plumbing. They must + # pass through, or batch gdb stops and silently detaches. + signals = ["SIGUSR1", "SIGUSR2", "SIGPIPE", "SIGCHLD"] + [ + f"SIG{number}" for number in range(32, 40) + ] + lines = [ + "set confirm off", + "set pagination off", + "set height 0", + "set width 0", + f"attach {pid}", + ] + lines.extend(f"handle {name} nostop noprint pass" for name in signals) + lines.extend( + [ + f"source {payload}", + f'python start_trace({json.dumps(str(trace))}, {cards})', + "continue", + ] + ) + return "\n".join(lines) + "\n" + + +def serialise_locations(locations: dict) -> dict: + return {key: [asdict(value) for value in values] for key, values in locations.items()} + + +def terminate_gdb(process: subprocess.Popen, status: Status, pid: int): + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=12) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + deadline = time.time() + 8 + while time.time() < deadline and tracer_pid(pid): + time.sleep(0.25) + status.log( + "gdb detached", + gdb_returncode=process.returncode, + tracer_pid=tracer_pid(pid), + target_state=target_state(pid), + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--status", type=Path, required=True) + parser.add_argument("--trace", type=Path, required=True) + parser.add_argument("--gdb-log", type=Path, required=True) + parser.add_argument("--monitor-log", type=Path, default=Path("/tmp/mt-monitor.jsonl")) + parser.add_argument("--fixture-index", type=int, default=0) + parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT) + parser.add_argument("--post-record-wait", type=int, default=12) + parser.add_argument( + "--arm-check-seconds", + type=int, + default=0, + help="attach, prove hardware breakpoints arm, then detach without claiming a capture", + ) + parser.add_argument( + "--wait-for-record-clear", + action="store_true", + help="keep tracing through abandon; accept creation only after old records disappear", + ) + parser.add_argument( + "--exclude-pid", + action="append", + type=int, + default=[], + help="ignore an existing FIFA process and attach only after process replacement", + ) + parser.add_argument( + "--payload", + default="gdb_match_team_writer_trace.py", + help="GDB Python payload in this tool directory; must expose start_trace(log, cards_base)", + ) + args = parser.parse_args() + + for path in (args.status, args.trace, args.gdb_log, args.monitor_log): + path.parent.mkdir(parents=True, exist_ok=True) + for path in (args.trace, args.gdb_log, args.monitor_log): + path.unlink(missing_ok=True) + status = Status(args.status, args.monitor_log) + payload = Path(__file__).with_name(args.payload).resolve() + if not payload.exists(): + status.write(state="failed", error=f"missing gdb payload: {payload}") + return 2 + + deadline = time.time() + args.timeout + status.write(state="waiting_for_ready_process", excluded_pids=args.exclude_pid) + status.log( + "waiting for FIFA17.exe with CardsDLL", + excluded_pids=args.exclude_pid, + ) + pid = None + cards = None + while time.time() < deadline: + # UMU/Proton creates a short-lived small FIFA17.exe before the real + # client. Never bind to the first comm match. Require CardsDLL and prefer + # the largest-RSS process (find_pids is ordered that way). + for candidate in find_pids(): + if candidate in args.exclude_pid: + continue + candidate_cards = cards_base(candidate) + if candidate_cards: + pid, cards = candidate, candidate_cards + break + if pid: + break + time.sleep(0.25) + if not pid or not cards: + status.write(state="timed_out", phase="ready_process") + return 3 + + status.write(state="ready_process_found", pid=pid, cards_base=cards) + status.log("real FIFA17.exe with CardsDLL found", pid=pid, cards_base=cards) + + command_path = Path(tempfile.gettempdir()) / f"mt-trace-{pid}.gdb" + command_path.write_text(gdb_commands(pid, cards, payload, args.trace)) + gdb_handle = args.gdb_log.open("w", encoding="utf-8") + process = subprocess.Popen( + ["gdb", "-q", "-nx", "-x", str(command_path)], + stdout=gdb_handle, + stderr=subprocess.STDOUT, + text=True, + ) + status.write( + state="attaching", + pid=pid, + cards_base=cards, + cards_image_base=CARDS_IMAGE_BASE, + gdb_pid=process.pid, + gdb_command_file=str(command_path), + payload=args.payload, + hardware_only=True, + client_memory_writes=False, + ) + status.log("gdb launched", pid=pid, gdb_pid=process.pid, cards_base=cards) + + armed = False + arm_deadline = min(deadline, time.time() + 60) + while time.time() < arm_deadline: + if process.poll() is not None: + break + events = read_events(args.trace) + if any(event.get("event") == "trace_armed" for event in events): + armed = True + break + time.sleep(0.25) + if not armed: + gdb_handle.close() + status.write( + state="failed", + phase="arm", + gdb_returncode=process.poll(), + tracer_pid=tracer_pid(pid), + trace_events=event_counts(read_events(args.trace)), + ) + if process.poll() is None: + terminate_gdb(process, status, pid) + return 4 + + attached = tracer_pid(pid) == process.pid + status.write( + state="armed", + tracer_pid=tracer_pid(pid), + target_state=target_state(pid), + trace_events=event_counts(read_events(args.trace)), + execution_breakpoints_armed=True, + team1_watchpoint_armed=False, + ) + status.log("trace armed", attached=attached, tracer_pid=tracer_pid(pid)) + if not attached: + terminate_gdb(process, status, pid) + gdb_handle.close() + status.write(state="failed", phase="attach_verification") + return 4 + if args.arm_check_seconds > 0: + time.sleep(args.arm_check_seconds) + events = read_events(args.trace) + counts = event_counts(events) + still_attached = tracer_pid(pid) == process.pid and process.poll() is None + terminate_gdb(process, status, pid) + gdb_handle.close() + passed = ( + still_attached + and counts.get("trace_armed", 0) == 1 + and counts.get("trace_error", 0) == 0 + and tracer_pid(pid) == 0 + and target_state(pid) != "T" + ) + status.write( + state="arm_check_passed" if passed else "arm_check_failed", + trace_events=counts, + attached_before_detach=still_attached, + tracer_pid_after_detach=tracer_pid(pid), + target_state_after_detach=target_state(pid), + ) + status.log("arm check complete", passed=passed, trace_events=counts) + return 0 if passed else 5 + + # A final record that already exists before arming cannot prove execution + # crossed creation under the debugger. Fail closed instead of converting an + # already-built match into a trusted zero-hit result. + initial_heap = scan_process( + pid, + args.fixture_index, + include_fixture=False, + writable_anon_only=True, + ) + records_preexisting = len(initial_heap["match_teams"]) >= 2 + records_cleared = not records_preexisting + if records_preexisting and args.wait_for_record_clear: + status.write( + state="waiting_for_record_clear", + locations=serialise_locations(initial_heap), + target_crossed_match_team_creation=False, + ) + status.log( + "trace armed; waiting for old match-team records to disappear", + team_ids=[team.team_id for team in initial_heap["match_teams"]], + ) + while time.time() < deadline: + if process.poll() is not None or not Path(f"/proc/{pid}").exists(): + terminate_gdb(process, status, pid) + gdb_handle.close() + status.write(state="failed", phase="record_clear") + return 5 + heap = scan_process( + pid, + args.fixture_index, + include_fixture=False, + writable_anon_only=True, + ) + if not heap["match_teams"]: + records_cleared = True + status.write( + state="records_cleared", + cleared_unix=time.time(), + tracer_pid=tracer_pid(pid), + gdb_alive=process.poll() is None, + target_state=target_state(pid), + ) + status.log( + "old match-team records disappeared; next records are a fresh creation", + tracer_pid=tracer_pid(pid), + ) + break + time.sleep(2) + if not records_cleared: + terminate_gdb(process, status, pid) + gdb_handle.close() + status.write(state="timed_out", phase="record_clear") + return 3 + elif records_preexisting: + counts = event_counts(read_events(args.trace)) + terminate_gdb(process, status, pid) + gdb_handle.close() + status.write( + state="armed_too_late", + phase="preexisting_records", + trace_events=counts, + locations=serialise_locations(initial_heap), + target_crossed_match_team_creation=False, + tracer_pid_after_detach=tracer_pid(pid), + target_state_after_detach=target_state(pid), + ) + status.log( + "match-team records pre-existed trace; no writer claim", + team_ids=[team.team_id for team in initial_heap["match_teams"]], + ) + return 6 + + + fixture = None + latest_locations = {"fixtures": [], "match_teams": [], "match_configs": []} + last_fixture_scan = 0.0 + records_seen_at = None + record_control = None + try: + while time.time() < deadline: + if process.poll() is not None or not Path(f"/proc/{pid}").exists(): + status.write( + state="failed", + phase="monitor", + gdb_returncode=process.poll(), + target_exists=Path(f"/proc/{pid}").exists(), + ) + return 5 + + now = time.time() + if fixture is None and now - last_fixture_scan >= 8: + full = scan_process(pid, args.fixture_index, include_fixture=True) + last_fixture_scan = now + if full["fixtures"]: + fixture = full["fixtures"][0] + latest_locations["fixtures"] = full["fixtures"] + status.log( + "fixture located", + address=fixture.address, + selected_address=fixture.selected_address, + selected_index=fixture.selected_index, + selected_team_id=fixture.selected_team_id, + ) + + heap = scan_process( + pid, + args.fixture_index, + include_fixture=False, + writable_anon_only=True, + ) + latest_locations["match_teams"] = heap["match_teams"] + latest_locations["match_configs"] = heap["match_configs"] + events = read_events(args.trace) + counts = event_counts(events) + is_attached = tracer_pid(pid) == process.pid + watch_armed = counts.get("team1_watchpoint_armed", 0) > 0 + status.write( + state="capturing" if len(heap["match_teams"]) < 2 else "records_observed", + tracer_pid=tracer_pid(pid), + gdb_alive=process.poll() is None, + target_state=target_state(pid), + trace_events=counts, + team1_watchpoint_armed=watch_armed, + locations=serialise_locations(latest_locations), + ) + + if len(heap["match_teams"]) >= 2: + if records_seen_at is None: + if not is_attached or process.poll() is not None: + status.write( + state="failed", + phase="record_creation_control", + tracer_pid=tracer_pid(pid), + gdb_alive=process.poll() is None, + trace_events=counts, + ) + return 5 + records_seen_at = now + record_control = { + "gdb_alive": process.poll() is None, + "tracer_pid": tracer_pid(pid), + "attached": is_attached, + "execution_breakpoints_armed": counts.get("trace_armed", 0) == 1, + "team1_watchpoint_armed": watch_armed, + } + status.log( + "two match-team records located", + team_ids=[team.team_id for team in heap["match_teams"]], + trace_events=counts, + **record_control, + ) + if now - records_seen_at >= args.post_record_wait: + break + time.sleep(3) + finally: + terminate_gdb(process, status, pid) + gdb_handle.close() + + events = read_events(args.trace) + counts = event_counts(events) + final = { + "state": "captured", + "pid": pid, + "cards_base": cards, + "fixture": asdict(fixture) if fixture else None, + "locations": serialise_locations(latest_locations), + "trace_events": counts, + "record_creation_control": record_control, + "gdb_alive_at_record_creation": bool( + record_control and record_control["gdb_alive"] and record_control["attached"] + ), + "target_crossed_match_team_creation": len(latest_locations["match_teams"]) >= 2, + "candidate_entry_hit": counts.get("candidate_entry", 0) > 0, + "team1_write_hit": counts.get("team1_write_post", 0) > 0, + "opponent_lookup_store_hit": counts.get("opponent_lookup_store_pre", 0) > 0, + "tracer_pid_after_detach": tracer_pid(pid), + "target_state_after_detach": target_state(pid), + "records_preexisting": records_preexisting, + "records_cleared_before_capture": records_cleared, + } + status.write(**final) + status.log("capture complete", **final) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())