tool(fifa17-recon): trace Offline Seasons team assignment hardware-only

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.
This commit is contained in:
funman300
2026-08-25 17:25:41 +00:00
parent 0701ac94e1
commit 433a9b22dd
6 changed files with 1728 additions and 0 deletions
@@ -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("<Q", 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}
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,
)
@@ -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("<i", data)[0] 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):
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,
)
@@ -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("<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,
)
@@ -0,0 +1,170 @@
"""Hardware-only origin trace for CardsDLL team-pair submissions.
Distinguishes the three callers of the engine team-id service that can submit a
full two-team pair, plus the mode-76 builder that prepares its pair:
0x1800c7583 correct fixture pair control
0x1800c6c23 generic pair submitter
0x1800c8dc1 mode-76 pair submitter
0x1800c8bf0 mode-76 pair builder entry
No INT3/software breakpoints. No client memory writes.
"""
from __future__ import annotations
import json
import os
import struct
import time
import traceback
import gdb
CARDS_IMAGE_BASE = 0x180000000
SITES = {
0x1800C7583: ("fixture_pair_submit", "r14", "rsi"),
0x1800C6C23: ("generic_pair_submit", "r14", "rsi"),
0x1800C8DC1: ("mode76_pair_submit", "r15", "rbp"),
}
MODE76_BUILDER = 0x1800C8BF0
_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 _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,
)
+306
View File
@@ -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/<pid>/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("<iBBHii", team_id, 1, round_id, 0, 1, 400)
for round_id, team_id in enumerate(teams)
)
def readable_regions(pid: int, *, writable_anon_only: bool = False):
with open(f"/proc/{pid}/maps") as maps:
for line in maps:
match = re.match(
r"([0-9a-f]+)-([0-9a-f]+)\s+(\S{4})\s+\S+\s+\S+\s+\S+\s*(.*)",
line,
)
if not match:
continue
lo, hi = int(match.group(1), 16), int(match.group(2), 16)
perms, path = match.group(3), match.group(4).strip()
if perms[0] != "r" or path.startswith(("/dev", "/memfd")):
continue
if hi - lo > 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("<i", buf, offset)[0]
def scan_fixture_buffer(buf: bytes, base: int, selected_index: int) -> 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(
"<iBBHii", buf, at
)
records.append(
{
"team_id": team_id,
"difficulty": difficulty,
"round_id": parsed_round,
"reward_mult": reward_mult,
"coins": coins,
}
)
found.append(
Fixture(
address=base + offset,
selected_address=base + offset + selected_index * 16,
selected_index=selected_index,
selected_team_id=records[selected_index]["team_id"],
records=records,
)
)
offset = buf.find(pattern, offset + 4)
return found
def scan_match_team_buffer(buf: bytes, base: int) -> 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("<iii", team, 0x14, 130000, 0x54001, 0x54002)
struct.pack_into("<11i", team, 0x20, *range(11))
struct.pack_into("<12i", team, 0x4C, *range(20, 32))
config = bytearray(0x20 + 0x50)
config[:0x20] = PARTICIPANT_PREFIX
pair = 0x20
struct.pack_into("<ii", config, pair, 130000, 130000)
struct.pack_into("<4I", config, pair + 0x1C, F01, F01, F01, F01)
struct.pack_into("<iiii", config, pair + 0x38, 11, 11, 0, 5)
buf = b"X" * 32 + fixture + b"Y" * 32 + team + b"Z" * 32 + config
fixtures = scan_fixture_buffer(buf, 0x1000, 0)
teams = scan_match_team_buffer(buf, 0x1000)
configs = scan_match_config_buffer(buf, 0x1000)
assert len(fixtures) == 1 and fixtures[0].selected_team_id == 73
assert len(teams) == 1 and teams[0].team_id == 130000
assert len(configs) == 1 and configs[0].team_id_1 == 130000
# The transient handle is never part of the anchor.
struct.pack_into("<i", team, 0x18, -1)
assert len(scan_match_team_buffer(bytes(team), 0)) == 1
print("offline_match_locator selftest: PASS")
def main() -> 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())
+512
View File
@@ -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/<pid>/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())