Every call-graph result in this repo was computed over direct (E8) edges.
5,045 of the 5,207 functions named by a vftable slot have zero direct call
sites, so all of those results were lower bounds. Lane Z's dominant RNG
consumer hung off exactly such an edge.
tools/vtable_map.py builds, from the RTTI walk plus a full sweep to the next
function start (rule 17):
* vftable -> class -> sub-object offset -> slot -> target, and its inverse
* the class hierarchy from the RTTI base lists, so an abstract interface
with one concrete override resolves uniquely
* constructor-derived member typing (ctor result -> [this+d])
* the slot index at every indirect call site, with a backward register
resolver that refuses to cross a branch target rather than guess
* `this`-carrier spans and this/member call-graph propagation of class
Validation (12/12): rediscovers ServerTradeManagerImpl slot 10 ->
GenerateTradeRaidEncounters from the dispatch at 0x007d8469 with nothing
hand-fed, and re-derives the *Impl rule for both managers. Receiver-class
pinning reaches only 2.5% of the 6,398 virtual sites, at 0.6% out-of-range
against a 70% chance baseline; the displacement-only route measured worse
than random (81% vs 58%) and is rejected outright.
Closes lane K's tier-4 blind spot: all nine phase-23 calls and both phase-33
calls named. Four of the eleven reach a draw on the strategic generator
(StrategyServer+0x16c) at eight instruction-verified sites, none ever
observed firing — so "the tail draws nothing" is a property of eight turns,
not of the code. Also resolves the nine parked inlined-draw functions to
their vtable roots (correcting how that was recorded: none is itself in a
vftable; their topmost ancestors are), and finds 14,958 inter-function tail
jump edges without which three of them look like dead code.
1257 lines
51 KiB
Python
1257 lines
51 KiB
Python
#!/usr/bin/env python3
|
|
"""Indirect-call-edge resolver for the SOTS1 exe (lane V2).
|
|
|
|
Why this exists
|
|
---------------
|
|
Every call-graph result this campaign has produced -- reachability, "no
|
|
caller", closure sizes -- was computed over **direct** (E8 rel32) edges. In a
|
|
1,598-class C++ binary every `virtual` method is reached through
|
|
`call [reg+disp]` / `call reg` instead, so all of those results are lower
|
|
bounds. Lane Z's live hooking proved the cost: the single largest RNG consumer
|
|
of a strategic turn, `GenerateTradeRaidEncounters` 0x00893290, has **zero**
|
|
direct call sites in the 41,411-function image. Its only reference is
|
|
`Game::ServerTradeManagerImpl` vftable 0x00a31b74 slot 10, dispatched from
|
|
`call edx` at 0x007d8469 -- one instruction before a direct call the existing
|
|
closure did follow.
|
|
|
|
What this tool does
|
|
-------------------
|
|
vtables vftable VA -> class, sub-object offset, slot -> target
|
|
inverse function VA -> [(vftable, class, offset, slot)]
|
|
hierarchy class -> derived classes (from RTTI base lists)
|
|
ctors function VA -> class it installs a vptr for
|
|
sites every indirect call site, with its **slot index** recovered
|
|
resolve receiver typing where it is pinnable; honest UNPINNED else
|
|
|
|
Method notes that matter
|
|
------------------------
|
|
* Bodies are swept to the **next function start**, never `fva + sizeInBytes`
|
|
(rule 17: Ghidra's size understates ~11% of bodies and sometimes ends
|
|
mid-instruction).
|
|
* Backward register resolution refuses to cross an intra-function branch
|
|
**target**. A def separated from its use by a label is not a def we can
|
|
prove reaches the use, and it is reported as unresolved rather than
|
|
guessed. Every unmodelled opcode also stops the walk.
|
|
* `call [reg+disp]` with `reg` never loaded from `[obj+0]` is not a virtual
|
|
dispatch (import thunks, function pointers in tables). Those are reported
|
|
separately, not folded into the vtable answer.
|
|
|
|
Usage
|
|
-----
|
|
uv run python3 tools/vtable_map.py build # -> dumps/vtables.json
|
|
uv run python3 tools/vtable_map.py who 0x00893290 # vtables containing fn
|
|
uv run python3 tools/vtable_map.py vt 0x00a31b74 # dump one vtable
|
|
uv run python3 tools/vtable_map.py site 0x007d8469
|
|
uv run python3 tools/vtable_map.py sites 0x007d92a0 # all in a function
|
|
uv run python3 tools/vtable_map.py impls Game::ServerTradeManager
|
|
uv run python3 tools/vtable_map.py callers 0x00893290 # indirect callers
|
|
uv run python3 tools/vtable_map.py stats
|
|
"""
|
|
import bisect
|
|
import json
|
|
import os
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO = os.path.dirname(HERE)
|
|
sys.path.insert(0, HERE)
|
|
|
|
import x86disp as X # noqa: E402
|
|
|
|
EXE = os.path.join(REPO, "dumps", "sots.exe")
|
|
FUNCS = os.path.join(REPO, "dumps", "functions.json")
|
|
RTTI = os.path.join(REPO, "dumps", "rtti.json")
|
|
OUT = os.path.join(REPO, "dumps", "vtables.json")
|
|
|
|
R32 = X.R32
|
|
|
|
|
|
# ------------------------------------------------------------- reg write sets
|
|
CLOB_CALL = frozenset(("eax", "ecx", "edx"))
|
|
|
|
|
|
def writes(raw):
|
|
"""Registers written by one instruction.
|
|
|
|
Returns (set, modelled). `modelled=False` means "this opcode is not in the
|
|
table" -- the caller must treat it as clobbering everything and stop.
|
|
Only the forms MSVC 7.1 actually emits are modelled; the rest stop the walk
|
|
rather than being guessed at, which is what keeps the backward resolver
|
|
sound.
|
|
"""
|
|
i = 0
|
|
o66 = False
|
|
while i < len(raw) and raw[i] in (0x66, 0x67, 0xF0, 0xF2, 0xF3,
|
|
0x2E, 0x36, 0x3E, 0x26, 0x64, 0x65):
|
|
if raw[i] == 0x66:
|
|
o66 = True
|
|
i += 1
|
|
if i >= len(raw):
|
|
return set(), False
|
|
rep = 0xF3 in raw[:i] or 0xF2 in raw[:i]
|
|
op = raw[i]
|
|
i += 1
|
|
|
|
def modrm():
|
|
if i >= len(raw):
|
|
return None, None, None
|
|
m = raw[i]
|
|
return m >> 6, (m >> 3) & 7, m & 7
|
|
|
|
if op == 0x0F:
|
|
if i >= len(raw):
|
|
return set(), False
|
|
op2 = raw[i]
|
|
i += 1
|
|
mod, reg, rm = modrm()
|
|
if 0x80 <= op2 <= 0x8F: # jcc rel32
|
|
return set(), True
|
|
if 0x90 <= op2 <= 0x9F: # setcc r/m8
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if 0x40 <= op2 <= 0x4F: # cmovcc r32, r/m32
|
|
return {R32[reg]}, True
|
|
if op2 in (0xAF, 0xB6, 0xB7, 0xBE, 0xBF, 0xBC, 0xBD,
|
|
0x2C, 0x2D, 0x5A, 0x5B):
|
|
# imul / movzx / movsx / bsf / bsr / cvttss2si / cvtss2si
|
|
if op2 in (0x5A, 0x5B):
|
|
return set(), True # cvt*ps*, xmm dest
|
|
return {R32[reg]}, True
|
|
if op2 == 0x7E: # movd r/m32, mm/xmm
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op2 in (0x6E, 0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17,
|
|
0x28, 0x29, 0x2A, 0x2E, 0x2F, 0x51, 0x54, 0x55, 0x56,
|
|
0x57, 0x58, 0x59, 0x5C, 0x5D, 0x5E, 0x5F, 0x6F, 0x7F,
|
|
0xD6, 0xEF, 0xC6, 0x12 | 0):
|
|
return set(), True # SSE/MMX, no GP dest
|
|
if op2 in (0xA2,): # cpuid
|
|
return {"eax", "ebx", "ecx", "edx"}, True
|
|
if op2 in (0xA3, 0xAB, 0xB3, 0xBB): # bt/bts/btr/btc
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op2 in (0xC0, 0xC1): # xadd
|
|
return ({R32[rm], R32[reg]} if mod == 3 else {R32[reg]}), True
|
|
if op2 in (0xB0, 0xB1): # cmpxchg
|
|
return {"eax"} | ({R32[rm]} if mod == 3 else set()), True
|
|
if op2 == 0x31: # rdtsc
|
|
return {"eax", "edx"}, True
|
|
if op2 == 0x0B or op2 == 0x1F: # ud2 / nop
|
|
return set(), True
|
|
return set(), False
|
|
|
|
mod, reg, rm = modrm()
|
|
|
|
if op in (0x88, 0x89): # mov r/m, r
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op in (0x8A, 0x8B): # mov r, r/m
|
|
return {R32[reg]}, True
|
|
if op == 0x8D: # lea
|
|
return {R32[reg]}, True
|
|
if op in (0xC6, 0xC7): # mov r/m, imm
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if 0xB0 <= op <= 0xB7:
|
|
return {R32[op - 0xB0]}, True # mov r8, imm8
|
|
if 0xB8 <= op <= 0xBF:
|
|
return {R32[op - 0xB8]}, True # mov r32, imm32
|
|
if op < 0x40 and (op & 7) < 6 and (op & 0x38) != 0x38:
|
|
# add/or/adc/sbb/and/sub/xor family (0x38..0x3D is cmp -> excluded)
|
|
lo = op & 7
|
|
if lo in (0, 1):
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if lo in (2, 3):
|
|
return {R32[reg]}, True
|
|
return {"eax"}, True
|
|
if 0x38 <= op <= 0x3D: # cmp
|
|
return set(), True
|
|
if 0x40 <= op <= 0x47:
|
|
return {R32[op - 0x40]}, True # inc
|
|
if 0x48 <= op <= 0x4F:
|
|
return {R32[op - 0x48]}, True # dec
|
|
if 0x50 <= op <= 0x57:
|
|
return set(), True # push
|
|
if 0x58 <= op <= 0x5F:
|
|
return {R32[op - 0x58]}, True # pop
|
|
if op in (0x68, 0x6A):
|
|
return set(), True # push imm
|
|
if op in (0x69, 0x6B): # imul r, r/m, imm
|
|
return {R32[reg]}, True
|
|
if 0x70 <= op <= 0x7F:
|
|
return set(), True # jcc rel8
|
|
if op in (0x80, 0x81, 0x83): # group1 r/m, imm
|
|
if reg == 7:
|
|
return set(), True # cmp
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op in (0x84, 0x85):
|
|
return set(), True # test
|
|
if op in (0x86, 0x87): # xchg
|
|
return ({R32[rm], R32[reg]} if mod == 3 else {R32[reg]}), True
|
|
if op == 0x8F: # pop r/m
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if 0x90 <= op <= 0x97:
|
|
return ({"eax", R32[op - 0x90]} if op != 0x90 else set()), True
|
|
if op == 0x98:
|
|
return {"eax"}, True # cwde
|
|
if op == 0x99:
|
|
return {"edx"}, True # cdq
|
|
if op == 0x9C or op == 0x9D:
|
|
return set(), True # pushfd/popfd
|
|
if 0xA0 <= op <= 0xA1:
|
|
return {"eax"}, True # mov eax, moffs
|
|
if 0xA2 <= op <= 0xA3:
|
|
return set(), True # mov moffs, eax
|
|
if 0xA4 <= op <= 0xA7: # movs/cmps
|
|
return {"esi", "edi"} | ({"ecx"} if rep else set()), True
|
|
if op in (0xA8, 0xA9):
|
|
return set(), True # test eax, imm
|
|
if 0xAA <= op <= 0xAF: # stos/lods/scas
|
|
s = {"edi"} if op in (0xAA, 0xAB, 0xAE, 0xAF) else {"esi"}
|
|
if op in (0xAC, 0xAD):
|
|
s |= {"eax"}
|
|
return s | ({"ecx"} if rep else set()), True
|
|
if op in (0xC0, 0xC1, 0xD0, 0xD1, 0xD2, 0xD3): # shifts
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op in (0xC2, 0xC3, 0xC9, 0xCC, 0xCD):
|
|
return set(), True # ret / leave / int
|
|
if op == 0xE8:
|
|
return set(CLOB_CALL), True # call rel32
|
|
if op in (0xE9, 0xEB):
|
|
return set(), True # jmp
|
|
if 0xD8 <= op <= 0xDF: # x87
|
|
if op == 0xDF and i < len(raw) and raw[i] == 0xE0:
|
|
return {"eax"}, True # fnstsw ax
|
|
return set(), True
|
|
if op in (0xF6, 0xF7): # group3
|
|
if reg in (0, 1):
|
|
return set(), True # test
|
|
if reg in (2, 3):
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
return {"eax", "edx"}, True # mul/imul/div/idiv
|
|
if op in (0xF8, 0xF9, 0xFC, 0xFD):
|
|
return set(), True # clc/stc/cld/std
|
|
if op == 0xFE:
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if op == 0xFF: # group5
|
|
if reg in (0, 1):
|
|
return ({R32[rm]} if mod == 3 else set()), True
|
|
if reg in (2, 3):
|
|
return set(CLOB_CALL), True # call r/m32
|
|
return set(), True # jmp / push
|
|
return set(), False
|
|
|
|
|
|
# ------------------------------------------------------------ function sweeps
|
|
class Code:
|
|
def __init__(self):
|
|
_, self.secs = X.load_pe(EXE)
|
|
self.funcs = X.load_funcs()
|
|
self.starts = [f[0] for f in self.funcs]
|
|
self.name = {f[0]: f[1] for f in self.funcs}
|
|
self.fstart = set(self.starts)
|
|
|
|
def sec(self, va):
|
|
return next((s for s in self.secs if s[0] <= va < s[1]), None)
|
|
|
|
def owner(self, va):
|
|
j = bisect.bisect_right(self.starts, va) - 1
|
|
return self.starts[j] if j >= 0 else None
|
|
|
|
def body(self, fva):
|
|
"""Decode a function to the NEXT function start (rule 17).
|
|
|
|
Returns (instrs, targets, desync) where instrs is a list of
|
|
[va, length, raw] and targets is the set of intra-body branch targets.
|
|
"""
|
|
sec = self.sec(fva)
|
|
if sec is None:
|
|
return [], set(), "no section"
|
|
sva, eva, buf, _ = sec
|
|
j = bisect.bisect_right(self.starts, fva)
|
|
limit = min(self.starts[j] if j < len(self.starts) else eva, eva)
|
|
i = fva - sva
|
|
end = limit - sva
|
|
ins = []
|
|
tgts = set()
|
|
desync = None
|
|
while i < end:
|
|
try:
|
|
ln, _ = X.decode(buf, i, end)
|
|
except X.Desync as e:
|
|
desync = str(e)
|
|
break
|
|
va = sva + i
|
|
raw = buf[i:i + ln]
|
|
ins.append([va, ln, raw])
|
|
t = branch_target(va, ln, raw)
|
|
if t is not None and fva <= t < limit:
|
|
tgts.add(t)
|
|
i += ln
|
|
return ins, tgts, desync
|
|
|
|
|
|
def branch_target(va, ln, raw):
|
|
"""Target of a direct jump, or None. Calls are excluded on purpose: a
|
|
`call` returns to the next instruction, so it is not a label."""
|
|
k = 0
|
|
while k < len(raw) and raw[k] in (0x66, 0x67, 0xF0, 0xF2, 0xF3,
|
|
0x2E, 0x36, 0x3E, 0x26, 0x64, 0x65):
|
|
k += 1
|
|
op = raw[k]
|
|
if op == 0xEB or 0x70 <= op <= 0x7F:
|
|
return va + ln + struct.unpack_from("<b", raw, ln - 1)[0]
|
|
if op == 0xE9 and ln - k == 5:
|
|
return va + ln + struct.unpack_from("<i", raw, k + 1)[0]
|
|
if op == 0x0F and k + 1 < len(raw) and 0x80 <= raw[k + 1] <= 0x8F:
|
|
return va + ln + struct.unpack_from("<i", raw, k + 2)[0]
|
|
return None
|
|
|
|
|
|
def call_rel32(va, ln, raw):
|
|
k = 0
|
|
while k < len(raw) and raw[k] in (0x66, 0x67, 0xF0, 0xF2, 0xF3,
|
|
0x2E, 0x36, 0x3E, 0x26, 0x64, 0x65):
|
|
k += 1
|
|
if raw[k] == 0xE8 and ln - k == 5:
|
|
return va + ln + struct.unpack_from("<i", raw, k + 1)[0]
|
|
return None
|
|
|
|
|
|
def parse_modrm_mem(raw, k):
|
|
"""(mod, reg, base, index, scale, disp) for the ModRM at raw[k]."""
|
|
m = raw[k]
|
|
mod, reg, rm = m >> 6, (m >> 3) & 7, m & 7
|
|
k += 1
|
|
base = index = None
|
|
scale = 1
|
|
if mod != 3:
|
|
if rm == 4:
|
|
sib = raw[k]
|
|
k += 1
|
|
scale = 1 << (sib >> 6)
|
|
idx = (sib >> 3) & 7
|
|
bse = sib & 7
|
|
index = None if idx == 4 else R32[idx]
|
|
base = None if (bse == 5 and mod == 0) else R32[bse]
|
|
elif rm == 5 and mod == 0:
|
|
base = None
|
|
else:
|
|
base = R32[rm]
|
|
disp = 0
|
|
if mod == 1:
|
|
disp = struct.unpack_from("<b", raw, k)[0]
|
|
elif mod == 2 or base is None:
|
|
disp = struct.unpack_from("<i", raw, k)[0]
|
|
return mod, reg, base, index, scale, disp
|
|
return mod, reg, R32[rm], None, 1, 0
|
|
|
|
|
|
def instr_kind(raw):
|
|
"""Classify one instruction into the few shapes the resolver needs."""
|
|
k = 0
|
|
while k < len(raw) and raw[k] in (0x66, 0x67, 0xF0, 0xF2, 0xF3,
|
|
0x2E, 0x36, 0x3E, 0x26, 0x64, 0x65):
|
|
k += 1
|
|
op = raw[k]
|
|
if op == 0x8B: # mov r32, r/m32
|
|
mod, reg, base, idx, sc, disp = parse_modrm_mem(raw, k + 1)
|
|
return ("movload", R32[reg], mod, base, idx, sc, disp)
|
|
if op == 0x89: # mov r/m32, r32
|
|
mod, reg, base, idx, sc, disp = parse_modrm_mem(raw, k + 1)
|
|
return ("movstore", R32[reg], mod, base, idx, sc, disp)
|
|
if op == 0x8D: # lea
|
|
mod, reg, base, idx, sc, disp = parse_modrm_mem(raw, k + 1)
|
|
return ("lea", R32[reg], mod, base, idx, sc, disp)
|
|
if op == 0xC7: # mov r/m32, imm32
|
|
mod, reg, base, idx, sc, disp = parse_modrm_mem(raw, k + 1)
|
|
imm = struct.unpack_from("<I", raw, len(raw) - 4)[0]
|
|
return ("movimm", None, mod, base, idx, sc, disp, imm)
|
|
if 0xB8 <= op <= 0xBF:
|
|
return ("regimm", R32[op - 0xB8], 3, None, None, 1, 0,
|
|
struct.unpack_from("<I", raw, k + 1)[0])
|
|
if op == 0xFF:
|
|
mod, reg, base, idx, sc, disp = parse_modrm_mem(raw, k + 1)
|
|
if reg in (2, 3):
|
|
return ("icall", None, mod, base, idx, sc, disp)
|
|
if reg == 6:
|
|
return ("push", None, mod, base, idx, sc, disp)
|
|
if 0x50 <= op <= 0x57:
|
|
return ("pushreg", R32[op - 0x50], 3, None, None, 1, 0)
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------------ RTTI side
|
|
def load_rtti():
|
|
with open(RTTI) as fh:
|
|
r = json.load(fh)
|
|
return ({int(k, 16): v for k, v in r["cols"].items()},
|
|
{int(k, 16): v for k, v in r["vftables"].items()})
|
|
|
|
|
|
PURECALL = 0x00924fb0
|
|
|
|
|
|
# -------------------------------------------------------------------- build
|
|
def build():
|
|
code = Code()
|
|
cols, vfts = load_rtti()
|
|
|
|
# --- inverse map: function -> the vtable slots that name it
|
|
inverse = {}
|
|
for vf, v in vfts.items():
|
|
for i, s in enumerate(v["slots"]):
|
|
inverse.setdefault(s, []).append([vf, v["class"], v["offset"], i])
|
|
|
|
# --- class hierarchy: base -> derived
|
|
derived = {}
|
|
classes = {}
|
|
for vf, v in vfts.items():
|
|
classes.setdefault(v["class"], []).append(vf)
|
|
for b in v["bases"]:
|
|
if b != v["class"]:
|
|
derived.setdefault(b, set()).add(v["class"])
|
|
derived = {k: sorted(v) for k, v in derived.items()}
|
|
|
|
# --- abstract classes: every non-dtor slot is purecall
|
|
abstract = {}
|
|
for vf, v in vfts.items():
|
|
n = len(v["slots"])
|
|
p = sum(1 for s in v["slots"] if s == PURECALL)
|
|
if p and p >= n - 1:
|
|
abstract.setdefault(v["class"], []).append([vf, v["offset"], n, p])
|
|
|
|
vtset = set(vfts)
|
|
|
|
# --- sweep: indirect call sites, vptr stores, direct calls
|
|
sites = []
|
|
ctor_installs = {} # func -> [[vftable, disp]]
|
|
direct = {} # caller -> [callee] (E8 rel32)
|
|
jumped = {} # caller -> [callee] (tail jump / split)
|
|
fn_desync = 0
|
|
swept = 0
|
|
stores = []
|
|
fieldstores = [] # [store va, func, basereg, disp, callee]
|
|
thisedges = {} # caller -> [callees that receive caller's `this`]
|
|
memberedges = {} # caller -> [[callee, member disp]]
|
|
footprints = {} # vftable -> member displacements its ctor writes
|
|
carriers = {} # function -> regs that hold `this` throughout
|
|
for fva, fname, fsz in code.funcs:
|
|
ins, tgts, desync = code.body(fva)
|
|
if not ins:
|
|
continue
|
|
swept += 1
|
|
if desync:
|
|
fn_desync += 1
|
|
kinds = [instr_kind(r) for _, _, r in ins]
|
|
wr = [writes(r) for _, _, r in ins]
|
|
_te, _me, carr = this_edges(ins, kinds, wr, tgts)
|
|
carrier = carr
|
|
dcs = []
|
|
jmps = []
|
|
for n, (va, ln, raw) in enumerate(ins):
|
|
t = call_rel32(va, ln, raw)
|
|
if t is not None:
|
|
dcs.append(t)
|
|
jt = branch_target(va, ln, raw)
|
|
if jt is not None and jt != fva and jt in code.fstart:
|
|
# a jump that lands on another function's start: a tail call,
|
|
# or a Ghidra split. Either way it is a real control-flow
|
|
# edge, and a closure built from E8 alone misses it.
|
|
jmps.append(jt)
|
|
k = kinds[n]
|
|
if k is None:
|
|
continue
|
|
if k[0] in ("movimm", "regimm") and k[7] in vtset:
|
|
# vptr installation (C7 /0) or the `mov reg,vftable` half of one
|
|
if k[0] == "movimm" and k[2] != 3:
|
|
ctor_installs.setdefault(fva, []).append([k[7], k[6]])
|
|
elif k[0] == "regimm":
|
|
ctor_installs.setdefault(fva, []).append([k[7], None])
|
|
if k[0] == "icall":
|
|
sites.append(resolve_site(code, fva, ins, kinds, wr, tgts,
|
|
n, k, carrier))
|
|
if k[0] == "movstore" and k[2] != 3 and k[3] is not None \
|
|
and k[4] is None:
|
|
stores.append((fva, n, k))
|
|
if dcs:
|
|
direct[fva] = sorted(set(dcs))
|
|
if jmps:
|
|
jumped[fva] = sorted(set(jmps))
|
|
te, me = _te, _me
|
|
if carr:
|
|
carriers[hex(fva)] = [[r, hex(a), hex(b)] for r, a, b in carr]
|
|
for t in te:
|
|
thisedges.setdefault(hex(fva), []).append(hex(t))
|
|
for t, dd in me:
|
|
memberedges.setdefault(hex(fva), []).append([hex(t), dd])
|
|
# field typing needs the whole body decoded, so it runs here
|
|
# constructor member-write footprint: every displacement written off
|
|
# the same register the vptr was installed through. A class's ctor
|
|
# *enumerates* its members (rule 5), which makes the footprint a far
|
|
# sharper owner signature than any single displacement.
|
|
for vf, dd in ctor_installs.get(fva, ()):
|
|
if dd != 0:
|
|
continue
|
|
vreg = None
|
|
for (_, n, k) in stores:
|
|
pass
|
|
for n2, k2 in enumerate(kinds):
|
|
if k2 and k2[0] == "movimm" and k2[2] != 3 and k2[6] == 0 \
|
|
and k2[7] == vf:
|
|
vreg = k2[3]
|
|
break
|
|
if vreg is None:
|
|
continue
|
|
foot = sorted({k2[6] for k2 in kinds
|
|
if k2 and k2[0] in ("movstore", "movimm")
|
|
and k2[2] != 3 and k2[3] == vreg and k2[4] is None
|
|
and 0 <= k2[6] < 0x4000})
|
|
footprints.setdefault(hex(fva), {})["cls"] = vfts[vf]["class"]
|
|
footprints[hex(fva)].setdefault("d", []).extend(foot)
|
|
for (_, n, k) in stores:
|
|
src = k[1]
|
|
p, why = back_def(ins, kinds, wr, tgts, n, src)
|
|
if p is None:
|
|
continue
|
|
t = None
|
|
if src == "eax":
|
|
t = call_rel32(*ins[p])
|
|
if t is None:
|
|
kp = kinds[p]
|
|
if kp and kp[0] == "movload" and kp[2] == 3 and kp[3] == "eax":
|
|
q, _ = back_def(ins, kinds, wr, tgts, p, "eax")
|
|
if q is not None:
|
|
t = call_rel32(*ins[q])
|
|
if t is None:
|
|
continue
|
|
fieldstores.append([hex(ins[n][0]), hex(fva), k[3], k[6], hex(t)])
|
|
stores.clear()
|
|
|
|
out = {
|
|
"inverse": {hex(k): v for k, v in inverse.items()},
|
|
"derived": derived,
|
|
"abstract": abstract,
|
|
"classVtables": {k: [hex(x) for x in v] for k, v in classes.items()},
|
|
"ctorInstalls": {hex(k): v for k, v in ctor_installs.items()},
|
|
"fieldStores": fieldstores,
|
|
"thisEdges": thisedges,
|
|
"memberEdges": memberedges,
|
|
"carriers": carriers,
|
|
"footprints": {k: {"cls": v["cls"], "d": sorted(set(v.get("d", [])))}
|
|
for k, v in footprints.items()},
|
|
"sites": sites,
|
|
"direct": {hex(k): [hex(x) for x in v] for k, v in direct.items()},
|
|
"jumped": {hex(k): [hex(x) for x in v] for k, v in jumped.items()},
|
|
"stats": {"functionsSwept": swept, "desyncFunctions": fn_desync,
|
|
"vftables": len(vfts), "indirectSites": len(sites)},
|
|
}
|
|
with open(OUT, "w") as fh:
|
|
json.dump(out, fh)
|
|
kinds = {}
|
|
for s in sites:
|
|
kinds[s["kind"]] = kinds.get(s["kind"], 0) + 1
|
|
print(f"functions swept : {swept}")
|
|
print(f"desync functions : {fn_desync}")
|
|
print(f"vftables : {len(vfts)}")
|
|
print(f"indirect sites : {len(sites)}")
|
|
for k, v in sorted(kinds.items(), key=lambda x: -x[1]):
|
|
print(f" {k:<22}: {v}")
|
|
print(f"ctor vptr installs: {len(ctor_installs)} functions")
|
|
|
|
|
|
def this_edges(ins, kinds, wr, tgts):
|
|
"""Direct callees that are handed the caller's own `this` pointer.
|
|
|
|
MSVC parks `this` in a callee-saved register or an `ebp` slot in the
|
|
prologue and reloads it into `ecx` before each member call, so a member
|
|
function's class propagates down the direct call graph. That is the only
|
|
way to type the receiver inside a **non-virtual** method -- and the
|
|
functions this campaign cares about (`DetectEncounters`,
|
|
`OnAllCombatDone_Tail`) are exactly that: StrategyServer methods that
|
|
appear in no vftable.
|
|
|
|
Carriers are only *created* before the first intra-function label, where
|
|
the prologue lives and no branch has yet joined; they are killed anywhere
|
|
they are written. A carrier that is never killed therefore holds `this`
|
|
on every path, which is what makes the edge sound.
|
|
"""
|
|
regs = {"ecx"}
|
|
slots = set()
|
|
live = {"ecx": ins[0][0] if ins else 0}
|
|
spans = [] # [reg, first va, va it stops being `this`]
|
|
first_label = min(tgts) if tgts else None
|
|
out = set()
|
|
mem = set()
|
|
|
|
def kill(r, va):
|
|
if r in live:
|
|
spans.append([r, live.pop(r), va])
|
|
|
|
for n, (va, ln, raw) in enumerate(ins):
|
|
k = kinds[n]
|
|
t = call_rel32(va, ln, raw)
|
|
if t is not None:
|
|
m, why = back_def(ins, kinds, wr, tgts, n, "ecx")
|
|
if m is None:
|
|
if why == "no-def-in-body" and "ecx" in regs:
|
|
out.add(t)
|
|
else:
|
|
km = kinds[m]
|
|
if km and km[0] == "movload" and km[1] == "ecx":
|
|
if km[2] == 3 and km[3] in regs:
|
|
out.add(t)
|
|
elif km[2] != 3 and km[3] == "ebp" and km[6] in slots \
|
|
and km[4] is None:
|
|
out.add(t)
|
|
elif km[2] != 3 and km[4] is None and km[6] \
|
|
and km[3] in regs:
|
|
# ecx = [this + d]: the callee is a method of whatever
|
|
# class the *member* at +d holds
|
|
mem.add((t, km[6]))
|
|
# kills first -- an instruction that *defines* a carrier must not be
|
|
# seen as destroying it
|
|
w, modelled = wr[n]
|
|
if not modelled:
|
|
for r in list(regs):
|
|
kill(r, va)
|
|
regs.clear()
|
|
slots.clear()
|
|
else:
|
|
for r in regs & w:
|
|
kill(r, va)
|
|
regs -= w
|
|
if k and k[0] == "movstore" and k[2] != 3 and k[3] == "ebp" \
|
|
and k[4] is None and k[6] in slots:
|
|
slots.discard(k[6])
|
|
# Creation. A copy *from* a register that is provably `this` at this
|
|
# address makes the destination `this` too, wherever it sits -- MSVC
|
|
# reloads `this` from its stack home all over a large body. Creating
|
|
# a carrier out of thin air is still restricted to the prologue, where
|
|
# no branch has joined yet.
|
|
if k:
|
|
src = k[3] if k[0] in ("movload", "movstore") else None
|
|
fresh = (first_label is None or va < first_label)
|
|
if not (fresh or (src in regs and k[0] == "movload" and k[2] == 3)
|
|
or (k[0] == "movload" and k[2] != 3 and src == "ebp"
|
|
and k[4] is None and k[6] in slots)):
|
|
k = None
|
|
if k:
|
|
new = None
|
|
if k[0] == "movload" and k[2] == 3 and k[3] in regs:
|
|
new = k[1]
|
|
elif k[0] == "movstore" and k[2] != 3 and k[3] == "ebp" \
|
|
and k[4] is None and k[1] in regs:
|
|
slots.add(k[6])
|
|
elif k[0] == "movload" and k[2] != 3 and k[3] == "ebp" \
|
|
and k[4] is None and k[6] in slots:
|
|
new = k[1]
|
|
if new:
|
|
regs.add(new)
|
|
live.setdefault(new, va)
|
|
end = ins[-1][0] + 1 if ins else 0
|
|
for r in list(live):
|
|
kill(r, end)
|
|
return out, mem, spans
|
|
|
|
|
|
def back_def(ins, kinds, wr, tgts, n, reg):
|
|
"""Last definition of `reg` strictly before index n, or a reason it is not
|
|
provable. Refuses to cross a branch target or an unmodelled opcode."""
|
|
for m in range(n - 1, -1, -1):
|
|
if ins[m][0] in tgts:
|
|
return None, "crosses-label"
|
|
s, modelled = wr[m]
|
|
if not modelled:
|
|
return None, "unmodelled-opcode"
|
|
if reg in s:
|
|
return m, None
|
|
# a call clobbers eax/ecx/edx; already covered by wr
|
|
return None, "no-def-in-body"
|
|
|
|
|
|
def is_this(carrier, reg, va):
|
|
"""Was `reg` provably holding the incoming `this` at address va?
|
|
|
|
Spans end where the register is written -- including the `pop esi` of the
|
|
epilogue, which is why the carrier set has to be a span and not a single
|
|
set for the whole body.
|
|
"""
|
|
return any(r == reg and a <= va < b for r, a, b in carrier)
|
|
|
|
|
|
def resolve_site(code, fva, ins, kinds, wr, tgts, n, k, carrier=()):
|
|
"""Recover the vtable slot index and, if possible, the receiver expression
|
|
for one indirect call site."""
|
|
va = ins[n][0]
|
|
_, _, mod, base, idx, sc, disp = k
|
|
site = {"va": hex(va), "func": hex(fva), "name": code.name.get(fva, ""),
|
|
"kind": "unknown", "slot": None, "recv": None, "note": None}
|
|
|
|
if mod == 3:
|
|
# `call reg` -- the slot came from an earlier `mov reg,[vptr+disp]`
|
|
m, why = back_def(ins, kinds, wr, tgts, n, base)
|
|
if m is None:
|
|
site["kind"] = "call-reg-unresolved"
|
|
site["note"] = why
|
|
return site
|
|
km = kinds[m]
|
|
if km is None or km[0] != "movload" or km[2] == 3 or km[3] is None:
|
|
site["kind"] = "call-reg-nonmem"
|
|
return site
|
|
vreg, vdisp = km[3], km[6]
|
|
return _from_vptr(code, ins, kinds, wr, tgts, m, site, vreg, vdisp,
|
|
carrier)
|
|
|
|
if base is None:
|
|
# absolute [disp32] -- an import thunk or a global function pointer
|
|
site["kind"] = "call-abs"
|
|
site["note"] = hex(disp & 0xFFFFFFFF)
|
|
return site
|
|
if idx is not None:
|
|
site["kind"] = "call-indexed"
|
|
site["note"] = f"[{base}+{idx}*{sc}+0x{disp:x}]"
|
|
return site
|
|
# `call [reg+disp]` -- reg should be the vptr
|
|
return _from_vptr(code, ins, kinds, wr, tgts, n, site, base, disp,
|
|
carrier)
|
|
|
|
|
|
def _from_vptr(code, ins, kinds, wr, tgts, n, site, vreg, vdisp, carrier=()):
|
|
"""`vreg` is believed to hold a vptr; `vdisp` is the byte offset of the
|
|
slot. Prove the vptr by finding `mov vreg,[obj+0]`."""
|
|
if vdisp < 0 or vdisp % 4:
|
|
site["kind"] = "non-slot-disp"
|
|
site["note"] = hex(vdisp)
|
|
return site
|
|
site["slot"] = vdisp // 4
|
|
m, why = back_def(ins, kinds, wr, tgts, n, vreg)
|
|
if m is None:
|
|
site["kind"] = "vptr-unresolved"
|
|
site["note"] = why
|
|
return site
|
|
km = kinds[m]
|
|
if km is None:
|
|
site["kind"] = "vptr-unmodelled"
|
|
return site
|
|
if km[0] == "movload" and km[2] != 3 and km[3] is not None and km[6] == 0 \
|
|
and km[4] is None:
|
|
# mov vreg, [obj] -- a genuine vptr load
|
|
site["kind"] = "virtual"
|
|
site["recv"] = recv_expr(code, ins, kinds, wr, tgts, m, km[3], carrier)
|
|
return site
|
|
if km[0] == "movload" and km[2] != 3 and km[3] is None:
|
|
# mov vreg, [abs] -- vptr from a global object, or a global fn table
|
|
site["kind"] = "vptr-global"
|
|
site["note"] = hex(km[6] & 0xFFFFFFFF)
|
|
return site
|
|
site["kind"] = "not-vptr"
|
|
site["slot"] = None
|
|
site["note"] = km[0]
|
|
return site
|
|
|
|
|
|
def recv_expr(code, ins, kinds, wr, tgts, m, oreg, carrier=()):
|
|
"""Describe where the object pointer came from, one level up."""
|
|
if is_this(carrier, oreg, ins[m][0]):
|
|
return {"k": "this", "reg": oreg}
|
|
p, why = back_def(ins, kinds, wr, tgts, m, oreg)
|
|
if p is None:
|
|
# Only "no definition anywhere in the body" proves the value is the
|
|
# incoming register. "crosses-label" means a def may exist on a path
|
|
# we cannot see, and must not be read as `this`.
|
|
if why == "no-def-in-body":
|
|
return {"k": "entryreg", "reg": oreg}
|
|
return {"k": "unpinned", "reg": oreg, "note": why}
|
|
kp = kinds[p]
|
|
if kp is None:
|
|
return {"k": "unpinned", "reg": oreg, "note": "unmodelled"}
|
|
if kp[0] == "movload" and kp[2] != 3 and kp[3] is not None \
|
|
and kp[4] is None:
|
|
return {"k": "field", "base": kp[3], "disp": kp[6],
|
|
"this": is_this(carrier, kp[3], ins[p][0]),
|
|
"at": hex(ins[p][0])}
|
|
if kp[0] == "movload" and kp[2] != 3 and kp[3] is None:
|
|
return {"k": "global", "va": hex(kp[6] & 0xFFFFFFFF)}
|
|
if kp[0] == "movload" and kp[2] == 3:
|
|
if is_this(carrier, kp[3], ins[p][0]):
|
|
return {"k": "this", "reg": kp[3]}
|
|
return {"k": "reg", "reg": kp[3]}
|
|
if kp[0] == "lea":
|
|
return {"k": "lea", "base": kp[3], "disp": kp[6]}
|
|
t = call_rel32(*ins[p]) if oreg == "eax" else None
|
|
if t is not None:
|
|
return {"k": "callret", "target": hex(t),
|
|
"name": code.name.get(t, "")}
|
|
return {"k": "other", "form": kp[0], "at": hex(ins[p][0])}
|
|
|
|
|
|
# ---------------------------------------------------------------- resolution
|
|
def func_classes(d, vfts):
|
|
"""function VA -> {(class, sub-object offset)} the function is a method of,
|
|
plus the member-type index that falls out of the same fixpoint.
|
|
|
|
Four sources, all exact, iterated to a fixpoint because member typing and
|
|
method typing feed each other:
|
|
|
|
seed a function that *is* slot k of class C's vftable at sub-object +o
|
|
is a C method entered with `this` = obj+o;
|
|
seed a function that stores a C vftable pointer into [reg+d], d >= 0,
|
|
is a C constructor. (Negative displacements are inlined EH frames
|
|
parking a `std::bad_alloc` vptr on the stack, not construction.)
|
|
edge `mov ecx,<this carrier>; call F` -- F is a method of the same class;
|
|
edge `mov ecx,[<this carrier>+d]; call F` -- F is a method of whatever
|
|
class the member at +d holds, which the member index supplies.
|
|
|
|
The member index itself comes from `ctor result -> [this+d]` stores inside
|
|
functions whose own class is known, so every new method typing can add new
|
|
member typings and vice versa. Returns (fc, memberIndex, dispOnlyIndex).
|
|
"""
|
|
fc = {}
|
|
for f, rows in d["inverse"].items():
|
|
for vf, cls, off, slot in rows:
|
|
fc.setdefault(int(f, 16), set()).add((cls, off))
|
|
for f in d["ctorInstalls"]:
|
|
# A constructor receives the COMPLETE object, and installs each of its
|
|
# vptrs at [this + that vftable's sub-object offset]. So its `this`
|
|
# offset is 0, not the offset of whichever vftable it happens to
|
|
# install -- getting that wrong registers every member of the class at
|
|
# both +d and +d+4 and makes its own member typings ambiguous.
|
|
c = ctor_class(d, vfts, f)
|
|
if c is None:
|
|
continue
|
|
if not any(dd is not None and vf in vfts and dd == vfts[vf]["offset"]
|
|
for vf, dd in d["ctorInstalls"][f]):
|
|
continue
|
|
fc.setdefault(int(f, 16), set()).add((c, 0))
|
|
ap = os.path.join(REPO, "ghidra", "vtable-owners.json")
|
|
if os.path.exists(ap):
|
|
with open(ap) as fh:
|
|
for e in json.load(fh)["owners"]:
|
|
fc.setdefault(int(e["func"], 16), set()).add(
|
|
(e["class"], e.get("offset", 0)))
|
|
auth = {f: {c for c, o in v} for f, v in fc.items()}
|
|
te = {int(a, 16): [int(b, 16) for b in v] for a, v in d["thisEdges"].items()}
|
|
me = {int(a, 16): [(int(b, 16), dd) for b, dd in v]
|
|
for a, v in d["memberEdges"].items()}
|
|
ctorcls = {}
|
|
for callee in {x[4] for x in d["fieldStores"]}:
|
|
c = ctor_class(d, vfts, callee)
|
|
if c:
|
|
ctorcls[callee] = c
|
|
|
|
idx = anon = None
|
|
for _ in range(12):
|
|
idx, anon = {}, {}
|
|
for sva, fva, base, disp, callee in d["fieldStores"]:
|
|
c = ctorcls.get(callee)
|
|
if c is None:
|
|
continue
|
|
if base in ("ebp", "esp"):
|
|
continue
|
|
owners = fc.get(int(fva, 16))
|
|
if not owners:
|
|
anon.setdefault(disp, {}).setdefault(c, []).append(sva)
|
|
continue
|
|
for ocls, ooff in owners:
|
|
idx.setdefault((ocls, disp + ooff), {}) \
|
|
.setdefault(c, []).append(sva)
|
|
changed = False
|
|
def merge(b, new):
|
|
# An authoritative typing (vftable slot or vptr install) fixes the
|
|
# sub-object offset exactly. Propagation must not add a *second*
|
|
# offset for a class already fixed that way: a method entered on
|
|
# the +4 sub-object reads its members 4 lower, and letting both
|
|
# offsets stand turns every one of its member typings into a
|
|
# spurious ambiguity one slot away.
|
|
fixed = auth.get(b, set())
|
|
add = {(c, o) for c, o in new if c not in fixed}
|
|
n0 = len(fc.get(b, ()))
|
|
fc.setdefault(b, set()).update(add)
|
|
return len(fc[b]) != n0
|
|
|
|
for a, bs in te.items():
|
|
if a not in fc:
|
|
continue
|
|
for b in bs:
|
|
changed |= merge(b, fc[a])
|
|
for a, bs in me.items():
|
|
if a not in fc:
|
|
continue
|
|
for b, dd in bs:
|
|
got = set()
|
|
for ocls, ooff in fc[a]:
|
|
hits = idx.get((ocls, dd + ooff), {})
|
|
if len(hits) == 1:
|
|
got.add((next(iter(hits)), 0))
|
|
if not got:
|
|
continue
|
|
changed |= merge(b, got)
|
|
if not changed:
|
|
break
|
|
return fc, idx, anon
|
|
|
|
|
|
def ctor_class(d, vfts, callee):
|
|
"""The class a constructor constructs, or None.
|
|
|
|
A ctor installs its own vptrs *and* those of any base whose constructor the
|
|
compiler inlined, so several classes can appear at sub-object +0. The
|
|
most-derived one is the single candidate whose RTTI base list contains all
|
|
the others -- that is exactly what a base list is for. If no candidate
|
|
dominates, the function is not a constructor we can name and returns None
|
|
rather than a guess.
|
|
"""
|
|
inst = d["ctorInstalls"].get(callee, [])
|
|
cs = {vfts[v]["class"] for v, dd in inst
|
|
if v in vfts and dd is not None and dd >= 0 and vfts[v]["offset"] == 0}
|
|
if not cs:
|
|
return None
|
|
if len(cs) == 1:
|
|
return next(iter(cs))
|
|
bases = {}
|
|
for v, dd in inst:
|
|
if v in vfts and vfts[v]["offset"] == 0:
|
|
bases[vfts[v]["class"]] = set(vfts[v]["bases"])
|
|
for c in cs:
|
|
if cs - {c} <= bases.get(c, set()):
|
|
return c
|
|
return None
|
|
|
|
|
|
def field_index(d, vfts):
|
|
fc, idx, anon = func_classes(d, vfts)
|
|
return idx, anon, fc
|
|
|
|
|
|
def vtable_for(vfts, cls, off):
|
|
for vf, v in vfts.items():
|
|
if v["class"] == cls and v["offset"] == off:
|
|
return vf, v
|
|
return None, None
|
|
|
|
|
|
def resolve_all(d, vfts):
|
|
"""Attach a receiver class and a target function to every `virtual` site
|
|
we can pin. Returns (rows, counters)."""
|
|
fidx, anon, fc = field_index(d, vfts)
|
|
rows = []
|
|
ctr = {}
|
|
for s in d["sites"]:
|
|
if s["kind"] != "virtual":
|
|
continue
|
|
r = s["recv"] or {}
|
|
fva = int(s["func"], 16)
|
|
mine = fc.get(fva, set())
|
|
cand = None
|
|
how = None
|
|
if r.get("k") == "field" and r["base"] in ("ebp", "esp"):
|
|
# [ebp+8] is argument 1, [ebp-0x30] a local -- neither is a member
|
|
# of `this`, and typing them as one produced every out-of-range
|
|
# result the first version of V4 found.
|
|
how = "stack"
|
|
elif r.get("k") == "field" and not r.get("this"):
|
|
# the base register is not a proven `this` carrier, so which
|
|
# object's member this is cannot be established
|
|
how = "field-nonthis"
|
|
elif r.get("k") == "field":
|
|
hits = {}
|
|
for ocls, ooff in mine:
|
|
for cls, where in fidx.get((ocls, r["disp"] + ooff), {}).items():
|
|
hits.setdefault(cls, []).extend(where)
|
|
if len(hits) == 1:
|
|
cand, how = next(iter(hits)), "field"
|
|
elif len(hits) > 1:
|
|
how = "field-ambiguous"
|
|
elif not mine:
|
|
# no owning class for the *calling* function: fall back to a
|
|
# bare displacement match, which V4 shows is barely better
|
|
# than chance. Kept separate and never merged into `field`.
|
|
# A bare displacement match across 1,598 classes is not
|
|
# evidence: V4 measures it at ~81% out-of-range, *worse* than
|
|
# picking a vtable at random. Counted, never used.
|
|
how = "disp-only-rejected"
|
|
elif (r.get("k") == "this"
|
|
or (r.get("k") == "entryreg" and r.get("reg") == "ecx")) \
|
|
and len(mine) == 1:
|
|
cand, how = next(iter(mine))[0], "self"
|
|
elif r.get("k") == "callret":
|
|
c = ctor_class(d, vfts, r["target"])
|
|
if c:
|
|
cand, how = c, "callret"
|
|
ctr[how or "unpinned"] = ctr.get(how or "unpinned", 0) + 1
|
|
tgts = []
|
|
if cand:
|
|
# the receiver's static type may itself be abstract; the callable
|
|
# set is that class plus every class derived from it
|
|
fam = [cand] + list(d["derived"].get(cand, []))
|
|
for c in fam:
|
|
vf, v = vtable_for(vfts, c, 0)
|
|
if v and s["slot"] is not None and s["slot"] < len(v["slots"]):
|
|
t = v["slots"][s["slot"]]
|
|
if t != PURECALL:
|
|
tgts.append([c, hex(vf), hex(t)])
|
|
rows.append({**s, "cls": cand, "how": how, "targets": tgts})
|
|
return rows, ctr
|
|
|
|
|
|
# ------------------------------------------------------------------ validate
|
|
def validate(d, vfts, code):
|
|
ok = fail = 0
|
|
|
|
def check(label, cond, detail=""):
|
|
nonlocal ok, fail
|
|
if cond:
|
|
ok += 1
|
|
print(f" PASS {label} {detail}")
|
|
else:
|
|
fail += 1
|
|
print(f" FAIL {label} {detail}")
|
|
|
|
print("V1 the known case -- lane Z's virtual edge, rediscovered blind")
|
|
s = next((x for x in d["sites"] if x["va"] == hex(0x007d8469)), None)
|
|
check("site 0x007d8469 classified virtual", s and s["kind"] == "virtual",
|
|
str(s and s["kind"]))
|
|
check("slot recovered = 10", s and s["slot"] == 10, str(s and s["slot"]))
|
|
check("receiver = member +0x158", s and s["recv"]["k"] == "field"
|
|
and s["recv"]["disp"] == 0x158, str(s and s["recv"]))
|
|
rows, _ = resolve_all(d, vfts)
|
|
r = next((x for x in rows if x["va"] == hex(0x007d8469)), None)
|
|
check("receiver class = Game::ServerTradeManagerImpl",
|
|
r and r["cls"] == "Game::ServerTradeManagerImpl", str(r and r["cls"]))
|
|
check("target = 0x00893290 GenerateTradeRaidEncounters",
|
|
r and [t[2] for t in r["targets"]] == [hex(0x00893290)],
|
|
str(r and r["targets"]))
|
|
inv = d["inverse"].get(hex(0x00893290), [])
|
|
check("0x00893290 named by exactly one vtable slot", len(inv) == 1, str(inv))
|
|
ndirect = sum(1 for c in d["direct"].values() if hex(0x00893290) in c)
|
|
check("0x00893290 has zero direct call sites", ndirect == 0, str(ndirect))
|
|
|
|
print("\nV2 the *Impl rule -- abstract interface, one concrete override")
|
|
vf, v = vtable_for(vfts, "Game::ServerTradeManager", 0)
|
|
p = sum(1 for x in v["slots"] if x == PURECALL)
|
|
check("Game::ServerTradeManager is abstract",
|
|
p == len(v["slots"]) - 1, f"{p}/{len(v['slots'])} purecall")
|
|
check("exactly one derived class",
|
|
d["derived"].get("Game::ServerTradeManager") ==
|
|
["Game::ServerTradeManagerImpl"],
|
|
str(d["derived"].get("Game::ServerTradeManager")))
|
|
vf2, v2 = vtable_for(vfts, "Game::ServerTradeManagerImpl", 0)
|
|
check("Impl overrides every slot",
|
|
all(x != PURECALL for x in v2["slots"]) and
|
|
len(v2["slots"]) == len(v["slots"]), f"{len(v2['slots'])} slots")
|
|
vfs, vs = vtable_for(vfts, "Game::ServerSpyManager", 0)
|
|
check("Game::ServerSpyManager is itself concrete (no *Impl)",
|
|
all(x != PURECALL for x in vs["slots"]) and
|
|
not d["derived"].get("Game::ServerSpyManager"),
|
|
f"{len(vs['slots'])} slots, derived="
|
|
f"{d['derived'].get('Game::ServerSpyManager')}")
|
|
check("ServerSpyManager derives IServerSpyManager/ISpyManager/IStreamable",
|
|
"Game::IServerSpyManager" in vs["bases"], str(vs["bases"]))
|
|
|
|
print("\nV3 slot-index recovery -- out-of-range test on self-dispatch")
|
|
fc, _, _ = func_classes(d, vfts)
|
|
n = bad = 0
|
|
for x in d["sites"]:
|
|
if x["kind"] != "virtual" or not x["recv"]:
|
|
continue
|
|
if x["recv"]["k"] not in ("entryreg", "this"):
|
|
continue
|
|
cs = fc.get(int(x["func"], 16), set())
|
|
if len(cs) != 1:
|
|
continue
|
|
cls, off = next(iter(cs))
|
|
_, v = vtable_for(vfts, cls, off)
|
|
if not v:
|
|
continue
|
|
n += 1
|
|
if x["slot"] >= len(v["slots"]):
|
|
bad += 1
|
|
print(f" self-dispatch sites with a uniquely known receiver class: {n}")
|
|
print(f" slot index out of that class's vtable range : {bad}"
|
|
f" ({100.0 * bad / max(n, 1):.2f}%)")
|
|
|
|
print("\nV4 receiver pinning -- out-of-range test, with a random baseline")
|
|
sizes = sorted(len(v["slots"]) for v in vfts.values())
|
|
byhow = {}
|
|
for x in rows:
|
|
if not x["cls"]:
|
|
continue
|
|
_, v = vtable_for(vfts, x["cls"], 0)
|
|
b = (not v) or x["slot"] >= len(v["slots"])
|
|
rnd = 1 - sum(1 for z in sizes if z > x["slot"]) / len(sizes)
|
|
e = byhow.setdefault(x["how"], [0, 0, 0.0])
|
|
e[0] += 1
|
|
e[1] += 1 if b else 0
|
|
e[2] += rnd
|
|
print(f" {'pinning route':<18} {'sites':>6} {'out-of-range':>14} "
|
|
f"{'random':>9}")
|
|
for h, (n4, b4, rb) in sorted(byhow.items()):
|
|
print(f" {h:<18} {n4:>6} {b4:>6} ({100.0*b4/n4:5.1f}%) "
|
|
f"{100.0*rb/n4:8.1f}%")
|
|
|
|
print("\nV5 member typing -- do independent stores of one member agree?")
|
|
fidx, _, _ = field_index(d, vfts)
|
|
multi = agree = 0
|
|
for disp, m in fidx.items():
|
|
tot = sum(len(x) for x in m.values())
|
|
if tot < 2:
|
|
continue
|
|
multi += 1
|
|
if len(m) == 1:
|
|
agree += 1
|
|
print(f" members with >=2 independent construction stores: {multi}")
|
|
print(f" those where every store names the same class : {agree}"
|
|
f" ({100.0 * agree / max(multi, 1):.1f}%)")
|
|
|
|
print(f"\n{ok} pass, {fail} fail")
|
|
return 1 if fail else 0
|
|
|
|
|
|
# --------------------------------------------------------------------- query
|
|
def load():
|
|
if not os.path.exists(OUT):
|
|
sys.exit("no index; run: uv run python3 tools/vtable_map.py build")
|
|
with open(OUT) as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def main():
|
|
argv = sys.argv[1:]
|
|
if not argv or argv[0] == "build":
|
|
return build()
|
|
cmd = argv[0]
|
|
d = load()
|
|
cols, vfts = load_rtti()
|
|
code = Code()
|
|
|
|
if cmd == "who":
|
|
va = int(argv[1], 0)
|
|
rows = d["inverse"].get(hex(va), [])
|
|
if not rows:
|
|
print(f"0x{va:08x} is in no vftable")
|
|
for vf, cls, off, slot in rows:
|
|
print(f" vftable 0x{vf:08x} +0x{off:x} slot {slot:<3} {cls}")
|
|
elif cmd == "vt":
|
|
va = int(argv[1], 0)
|
|
v = vfts.get(va)
|
|
if not v:
|
|
return print("not a vftable")
|
|
print(f"0x{va:08x} {v['class']} sub-object +0x{v['offset']:x} "
|
|
f"{len(v['slots'])} slots bases={v['bases']}")
|
|
for i, s in enumerate(v["slots"]):
|
|
print(f" [{i:>2}] +0x{4 * i:<3x} 0x{s:08x} "
|
|
f"{code.name.get(s, '')}{' PURECALL' if s == PURECALL else ''}")
|
|
elif cmd == "impls":
|
|
cls = argv[1]
|
|
print(f"{cls}: derived = {d['derived'].get(cls, [])}")
|
|
if cls in d["abstract"]:
|
|
print(f" abstract vtables: {d['abstract'][cls]}")
|
|
elif cmd == "site":
|
|
va = int(argv[1], 0)
|
|
for s in d["sites"]:
|
|
if s["va"] == hex(va):
|
|
print(json.dumps(s, indent=2))
|
|
elif cmd == "sites":
|
|
fva = int(argv[1], 0)
|
|
for s in d["sites"]:
|
|
if s["func"] == hex(fva):
|
|
print(f" {s['va']} slot={s['slot']} {s['kind']} "
|
|
f"recv={s['recv']} {s['note'] or ''}")
|
|
elif cmd == "callers":
|
|
va = int(argv[1], 0)
|
|
rows = d["inverse"].get(hex(va), [])
|
|
slots = {r[3] for r in rows}
|
|
clss = {r[1] for r in rows}
|
|
print(f"target in vtables of {sorted(clss)} at slots {sorted(slots)}")
|
|
for s in d["sites"]:
|
|
if s["slot"] in slots and s["kind"] == "virtual":
|
|
print(f" {s['va']} slot {s['slot']} in {s['name']} "
|
|
f"({s['func']}) recv={s['recv']}")
|
|
elif cmd == "field":
|
|
# `field 0x158 [funcVA]` -- who stores a constructed object into +disp,
|
|
# and what class did the constructor install a vptr for?
|
|
disp = int(argv[1], 0)
|
|
only = int(argv[2], 0) if len(argv) > 2 else None
|
|
ci = d["ctorInstalls"]
|
|
for sva, fva, base, dsp, callee in d["fieldStores"]:
|
|
if dsp != disp:
|
|
continue
|
|
if only is not None and int(fva, 16) != only:
|
|
continue
|
|
inst = ci.get(callee, [])
|
|
cs = sorted({vfts[v]["class"] for v, _ in inst if v in vfts})
|
|
if not cs:
|
|
continue
|
|
print(f" {sva} in {code.name.get(int(fva, 16), '')} ({fva}) "
|
|
f"[{base}+0x{disp:x}] <- {callee} "
|
|
f"{code.name.get(int(callee, 16), '')} installs {cs}")
|
|
elif cmd == "owner":
|
|
# `owner <funcVA> <reg>` -- rank candidate classes for the object a
|
|
# register points at, by overlap with each class ctor's member-write
|
|
# footprint. A RANKER, never a filter (rule 9).
|
|
fva = int(argv[1], 0)
|
|
reg = argv[2] if len(argv) > 2 else None
|
|
ins, tgts, _ = code.body(fva)
|
|
kinds = [instr_kind(r) for _, _, r in ins]
|
|
touched = {}
|
|
for k in kinds:
|
|
if k and k[0] in ("movload", "movstore", "movimm", "lea") \
|
|
and k[2] != 3 and k[3] and k[4] is None and k[6] > 0:
|
|
touched.setdefault(k[3], set()).add(k[6])
|
|
for r in ([reg] if reg else sorted(touched)):
|
|
t = touched.get(r, set())
|
|
if len(t) < 3:
|
|
continue
|
|
# IDF weighting: +0x4 and +0x8 are members of nearly every class
|
|
# and carry no information; a rare offset like +0x158 does.
|
|
import math
|
|
df = {}
|
|
for rec in d["footprints"].values():
|
|
for x in set(rec["d"]):
|
|
df[x] = df.get(x, 0) + 1
|
|
nf = len(d["footprints"]) or 1
|
|
w = {x: math.log(nf / (1 + df.get(x, 0))) for x in t}
|
|
tot = sum(w.values()) or 1.0
|
|
sc = []
|
|
for cf, rec in d["footprints"].items():
|
|
f = set(rec["d"])
|
|
if not f:
|
|
continue
|
|
hit = t & f
|
|
sc.append((sum(w[x] for x in hit) / tot, len(hit),
|
|
rec["cls"], cf))
|
|
sc.sort(reverse=True)
|
|
print(f" {r}: {len(t)} distinct member offsets touched")
|
|
seen = set()
|
|
for frac, n, cls, cf in sc:
|
|
if cls in seen:
|
|
continue
|
|
seen.add(cls)
|
|
print(f" {frac * 100:5.1f}% {n:>3}/{len(t)} {cls}"
|
|
f" (ctor {cf})")
|
|
if len(seen) >= 5:
|
|
break
|
|
elif cmd == "resolve":
|
|
rows, ctr = resolve_all(d, vfts)
|
|
want = int(argv[1], 0) if len(argv) > 1 else None
|
|
for r in rows:
|
|
if want is not None and int(r["func"], 16) != want \
|
|
and int(r["va"], 16) != want:
|
|
continue
|
|
t = ", ".join(f"{c}::[{s['slot'] if False else ''}]{tv}"
|
|
for c, _, tv in r["targets"]) or "-"
|
|
print(f" {r['va']} slot {str(r['slot']):>3} "
|
|
f"{r['cls'] or 'UNPINNED':<38} {r['how'] or '':<16} -> {t}")
|
|
if want is None:
|
|
print(json.dumps(ctr, indent=2))
|
|
elif cmd == "validate":
|
|
validate(d, vfts, code)
|
|
elif cmd == "stats":
|
|
print(json.dumps(d["stats"], indent=2))
|
|
kinds = {}
|
|
for s in d["sites"]:
|
|
kinds[s["kind"]] = kinds.get(s["kind"], 0) + 1
|
|
for k, v in sorted(kinds.items(), key=lambda x: -x[1]):
|
|
print(f" {k:<22}: {v}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main() or 0)
|