funman300
265 lines
8.5 KiB
Python
265 lines
8.5 KiB
Python
"""Hardware-only trace of CardsGameSetupAdapter query 13 and overwrite input.
|
|
|
|
Breakpoints:
|
|
|
|
FUN_180031340 entry incoming teamId/side/context
|
|
0x18003148f pre-call query id, selector, output/count pointers
|
|
0x180031495 post-call complete 48-byte records and count
|
|
0x180031861 submit original incoming teamId sent to engine
|
|
|
|
This proves whether query 13 influences the overwrite. No INT3/software
|
|
breakpoints, client writes, or game input.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import struct
|
|
import time
|
|
import traceback
|
|
|
|
import gdb
|
|
|
|
CARDS_IMAGE_BASE = 0x180000000
|
|
ENTRY = 0x180031340
|
|
QUERY_PRE = 0x18003148F
|
|
QUERY_POST = 0x180031495
|
|
SUBMIT = 0x180031861
|
|
MAX_RECORDS = 100
|
|
RECORD_SIZE = 48
|
|
_STATE = None
|
|
|
|
|
|
def _reg(name: str) -> int:
|
|
return int(gdb.parse_and_eval(f"${name}"))
|
|
|
|
|
|
def _read(address: int, size: int) -> bytes | None:
|
|
if not address or address < 0 or size < 0:
|
|
return None
|
|
try:
|
|
return bytes(gdb.selected_inferior().read_memory(address, size))
|
|
except gdb.error:
|
|
return None
|
|
|
|
|
|
def _u64(address: int) -> int | None:
|
|
data = _read(address, 8)
|
|
return struct.unpack("<Q", data)[0] if data else None
|
|
|
|
|
|
def _i32(address: int) -> int | None:
|
|
data = _read(address, 4)
|
|
return struct.unpack("<i", data)[0] if data else None
|
|
|
|
|
|
def _thread() -> dict:
|
|
thread = gdb.selected_thread()
|
|
if thread is None:
|
|
return {}
|
|
return {"name": thread.name, "ptid": list(thread.ptid), "global_num": thread.global_num}
|
|
|
|
|
|
def _registers() -> dict:
|
|
names = (
|
|
"rax", "rbx", "rcx", "rdx", "rsi", "rdi", "rbp", "rsp",
|
|
"r8", "r9", "r10", "r11", "r12", "r13", "r14", "r15", "rip",
|
|
)
|
|
return {name: _reg(name) for name in names}
|
|
|
|
|
|
def _printable_pointer(pointer: int) -> str | None:
|
|
data = _read(pointer, 96)
|
|
if not data:
|
|
return None
|
|
raw = data.split(b"\0", 1)[0]
|
|
if len(raw) < 3:
|
|
return None
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
return text if all(char.isprintable() for char in text) else None
|
|
|
|
|
|
def _decode_record(data: bytes, address: int) -> dict:
|
|
words = list(struct.unpack("<12i", data))
|
|
qwords = list(struct.unpack("<6Q", data))
|
|
strings = {}
|
|
for index, pointer in enumerate(qwords):
|
|
text = _printable_pointer(pointer)
|
|
if text:
|
|
strings[f"qword_{index}"] = {"pointer": pointer, "text": text}
|
|
interesting = {
|
|
str(value): [index * 4 for index, word in enumerate(words) if word == value]
|
|
for value in (73, 240, 241, 243, 130000, 130001)
|
|
if value in words
|
|
}
|
|
return {
|
|
"address": address,
|
|
"hex": data.hex(),
|
|
"i32": words,
|
|
"u32": [value & 0xFFFFFFFF for value in words],
|
|
"f32": list(struct.unpack("<12f", data)),
|
|
"qwords": qwords,
|
|
"strings": strings,
|
|
"interesting_values": interesting,
|
|
}
|
|
|
|
|
|
class State:
|
|
def __init__(self, path: str, cards_base: int):
|
|
self.path = path
|
|
self.cards_base = cards_base
|
|
self.index = 0
|
|
self.calls = {}
|
|
|
|
def log(self, kind: str, **payload):
|
|
self.index += 1
|
|
event = {
|
|
"event": kind,
|
|
"event_index": self.index,
|
|
"time_unix": time.time(),
|
|
"thread": _thread(),
|
|
**payload,
|
|
}
|
|
with open(self.path, "a", encoding="utf-8") as handle:
|
|
handle.write(json.dumps(event, sort_keys=True) + "\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
|
|
def thread_key(self):
|
|
return tuple(_thread().get("ptid", ()))
|
|
|
|
|
|
class HardwareBreakpoint(gdb.Breakpoint):
|
|
def __init__(self, state: State, image_va: int):
|
|
self.state = state
|
|
self.image_va = image_va
|
|
address = state.cards_base + (image_va - CARDS_IMAGE_BASE)
|
|
super().__init__(f"*0x{address:x}", type=gdb.BP_HARDWARE_BREAKPOINT, internal=False)
|
|
self.silent = True
|
|
|
|
|
|
class EntryBreakpoint(HardwareBreakpoint):
|
|
def stop(self):
|
|
try:
|
|
self.state.log(
|
|
"game_setup_entry",
|
|
incoming_context=_reg("rcx"),
|
|
incoming_side=_reg("rdx") & 0xFFFFFFFF,
|
|
incoming_team_id=_reg("r8") & 0xFFFFFFFF,
|
|
incoming_r9=_reg("r9"),
|
|
registers=_registers(),
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="entry", error=str(exc), traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
class QueryPreBreakpoint(HardwareBreakpoint):
|
|
def stop(self):
|
|
try:
|
|
rsp = _reg("rsp")
|
|
adapter = _reg("rcx")
|
|
vtable = _u64(adapter)
|
|
count_pointer = _u64(rsp + 0x20)
|
|
state = {
|
|
"adapter": adapter,
|
|
"adapter_vtable": vtable,
|
|
"query_target": _u64(vtable + 0xE0) if vtable else None,
|
|
"query_id": _reg("rdx") & 0xFFFFFFFF,
|
|
"selector": _reg("r8") & 0xFFFFFFFF,
|
|
"output_buffer": _reg("r9"),
|
|
"count_pointer": count_pointer,
|
|
"sixth_argument": _u64(rsp + 0x28),
|
|
"count_before": _i32(count_pointer) if count_pointer else None,
|
|
"saved_incoming_team_id": _i32(rsp + 0x34),
|
|
"saved_side": _i32(rsp + 0x50),
|
|
"saved_engine_context": _u64(rsp + 0x68),
|
|
"adapter_prefix_hex": (_read(adapter, 0x100) or b"").hex(),
|
|
}
|
|
self.state.calls[self.state.thread_key()] = state
|
|
self.state.log(
|
|
"query13_pre",
|
|
**state,
|
|
registers=_registers(),
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="query_pre", error=str(exc), traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
class QueryPostBreakpoint(HardwareBreakpoint):
|
|
def stop(self):
|
|
try:
|
|
state = self.state.calls.get(self.state.thread_key(), {})
|
|
count_pointer = state.get("count_pointer")
|
|
output = state.get("output_buffer")
|
|
count = _i32(count_pointer) if count_pointer else None
|
|
safe_count = min(max(count or 0, 0), MAX_RECORDS)
|
|
records = []
|
|
for index in range(safe_count):
|
|
address = output + index * RECORD_SIZE
|
|
data = _read(address, RECORD_SIZE)
|
|
if data and len(data) == RECORD_SIZE:
|
|
records.append(_decode_record(data, address))
|
|
self.state.log(
|
|
"query13_post",
|
|
query_state=state,
|
|
count_after=count,
|
|
records=records,
|
|
saved_incoming_team_id_after=_i32(_reg("rsp") + 0x34),
|
|
saved_side_after=_i32(_reg("rsp") + 0x50),
|
|
registers=_registers(),
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="query_post", error=str(exc), traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
class SubmitBreakpoint(HardwareBreakpoint):
|
|
def stop(self):
|
|
try:
|
|
rsp = _reg("rsp")
|
|
self.state.log(
|
|
"game_setup_submit",
|
|
submitted_team_id=_reg("r8") & 0xFFFFFFFF,
|
|
submitted_side=_reg("rdx") & 0xFFFFFFFF,
|
|
engine_context=_reg("rcx"),
|
|
saved_incoming_team_id=_i32(rsp + 0x34),
|
|
saved_side=_i32(rsp + 0x50),
|
|
registers=_registers(),
|
|
backtrace=gdb.execute("bt 24", to_string=True),
|
|
)
|
|
except Exception as exc:
|
|
self.state.log("trace_error", where="submit", error=str(exc), traceback=traceback.format_exc())
|
|
return False
|
|
|
|
|
|
def _on_exit(event):
|
|
if _STATE is not None:
|
|
_STATE.log("inferior_exited", detail=str(event))
|
|
|
|
|
|
def start_trace(log_path: str, cards_base: int):
|
|
global _STATE
|
|
open(log_path, "w", encoding="utf-8").close()
|
|
_STATE = State(log_path, cards_base)
|
|
points = {
|
|
"entry": EntryBreakpoint(_STATE, ENTRY),
|
|
"query_pre": QueryPreBreakpoint(_STATE, QUERY_PRE),
|
|
"query_post": QueryPostBreakpoint(_STATE, QUERY_POST),
|
|
"submit": SubmitBreakpoint(_STATE, SUBMIT),
|
|
}
|
|
gdb.events.exited.connect(_on_exit)
|
|
_STATE.log(
|
|
"trace_armed",
|
|
breakpoints={name: {"number": bp.number, "image_va": bp.image_va} for name, bp in points.items()},
|
|
record_size=RECORD_SIZE,
|
|
hardware_only=True,
|
|
client_memory_writes=False,
|
|
)
|