wip: extract the xexdb tool closure
This commit is contained in:
MechaCat02
216
tools/zq.py
Executable file
216
tools/zq.py
Executable file
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Sylpheed static-analysis helper over DuckDB `sylpheed.db`.
|
||||
|
||||
Hides the gotchas: DECIMAL bounds (DuckDB rejects 0x literals), read-only connect,
|
||||
and the fact that the engine vtable / rdata is NOT in the DB (read it from guest
|
||||
memory with `xenia-rs exec ... --dump-addr=0x<va>` instead).
|
||||
|
||||
Usage:
|
||||
zq.py dis <lo_hex> <hi_hex> # disassemble [lo,hi) (jump-table words shown as .long)
|
||||
zq.py fn <pc_hex> # function containing pc (address,name,end)
|
||||
zq.py xref <target_hex> # xrefs whose target == addr (callers)
|
||||
zq.py callers <vtable_off_dec> # call-sites of vtable slot at byte offset N
|
||||
# (finds `lwz r11, N(r11)` + reports the fn)
|
||||
zq.py grep <substr> # instructions whose operands LIKE %substr%
|
||||
zq.py find <word_hex> # instructions whose raw word == value (e.g. a ptr)
|
||||
|
||||
zq.py switch <pc_hex> # recovered switch cases for the bctr at/near pc
|
||||
zq.py switches [fn_hex] # every recovered switch (optionally in one function)
|
||||
zq.py classes [substr] # RTTI class names (+ vtable, method count)
|
||||
zq.py class <name> # one class: bases, vtable, virtual methods
|
||||
zq.py str <substr> # string literals matching, with referencing functions
|
||||
|
||||
zq.py xdbf [substr] # XDBF title text (all locales); substr filters
|
||||
zq.py ach # XDBF achievements (id, gamerscore, name, descriptions)
|
||||
|
||||
A command that needs a table the current DB predates prints what to regenerate
|
||||
rather than a SQL error.
|
||||
"""
|
||||
import duckdb, sys
|
||||
|
||||
DB = '/home/fabi/RE - Project Sylpheed/xenia-rs/sylpheed.db'
|
||||
c = duckdb.connect(DB, read_only=True)
|
||||
H = lambda x: '0x%08x' % x
|
||||
|
||||
REGEN = ("xenia-rs dis <xex|iso> --db sylpheed.db --analyze sql")
|
||||
|
||||
|
||||
def _need(*tables):
|
||||
"""Exit with a regeneration hint if any table is missing from this DB."""
|
||||
have = {r[0] for r in c.execute(
|
||||
"SELECT table_name FROM information_schema.tables").fetchall()}
|
||||
missing = [t for t in tables if t not in have]
|
||||
if missing:
|
||||
sys.exit(f"this db predates {', '.join(missing)} — regenerate with:\n {REGEN}")
|
||||
|
||||
|
||||
def _has_col(table, col):
|
||||
return any(r[0] == col for r in c.execute(
|
||||
"SELECT column_name FROM information_schema.columns WHERE table_name=?",
|
||||
[table]).fetchall())
|
||||
|
||||
|
||||
def _fn(pc):
|
||||
r = c.execute('SELECT address,name,end_address FROM functions WHERE address<=? AND end_address>? '
|
||||
'ORDER BY address DESC LIMIT 1', [pc, pc]).fetchall()
|
||||
return f'{r[0][1]}({H(r[0][0])})' if r else '?'
|
||||
|
||||
|
||||
def cmd_dis(lo, hi):
|
||||
data_col = 'is_data' if _has_col('instructions', 'is_data') else 'false'
|
||||
rows = c.execute(f'SELECT address,mnemonic,operands,raw,{data_col} FROM instructions '
|
||||
'WHERE address>=? AND address<? ORDER BY address', [lo, hi]).fetchall()
|
||||
for a, m, o, raw, is_data in rows:
|
||||
if is_data:
|
||||
print(H(a), '.long', H(raw & 0xffffffff), ' ; jump-table data')
|
||||
else:
|
||||
print(H(a), m, o)
|
||||
|
||||
|
||||
def cmd_switch(pc):
|
||||
_need('jump_tables', 'jump_table_entries')
|
||||
r = c.execute('SELECT bctr_pc,function,table_address,kind,entry_count FROM jump_tables '
|
||||
'WHERE bctr_pc>=? ORDER BY bctr_pc LIMIT 1', [pc]).fetchall()
|
||||
if not r:
|
||||
sys.exit('no recovered switch at or after %s' % H(pc))
|
||||
bctr, fn, tbl, kind, n = r[0]
|
||||
print(f'bctr {H(bctr)} in {_fn(bctr)} table={H(tbl)} kind={kind} cases={n}')
|
||||
for ci, tgt in c.execute('SELECT case_index,target_address FROM jump_table_entries '
|
||||
'WHERE bctr_pc=? ORDER BY case_index', [bctr]).fetchall():
|
||||
print(f' case {ci:>3} -> {H(tgt)}')
|
||||
|
||||
|
||||
def cmd_switches(fn):
|
||||
_need('jump_tables')
|
||||
q = ('SELECT bctr_pc,function,kind,entry_count,table_address FROM jump_tables '
|
||||
+ ('WHERE function=? ' if fn is not None else '') + 'ORDER BY bctr_pc')
|
||||
for bctr, f, kind, n, tbl in c.execute(q, [fn] if fn is not None else []).fetchall():
|
||||
print(H(bctr), f'{kind:<8}', f'cases={n:<4}', 'table=' + H(tbl), 'in', _fn(bctr))
|
||||
|
||||
|
||||
def cmd_classes(sub):
|
||||
_need('rtti_type_descriptors', 'rtti_locators')
|
||||
q = """SELECT td.demangled_name, c.vtable_address, c.subobject_offset,
|
||||
(SELECT count(*) FROM methods m WHERE m.vtable_address = c.vtable_address)
|
||||
FROM rtti_locators c
|
||||
JOIN rtti_type_descriptors td ON td.address = c.type_descriptor
|
||||
{} ORDER BY td.demangled_name, c.subobject_offset"""
|
||||
q = q.format('WHERE td.demangled_name ILIKE ?' if sub else '')
|
||||
for name, vt, off, nm in c.execute(q, [f'%{sub}%'] if sub else []).fetchall():
|
||||
loc = H(vt) if vt is not None else '-'
|
||||
print(f'{name:<60} vtable={loc} +0x{off:x} methods={nm}')
|
||||
|
||||
|
||||
def cmd_class(name):
|
||||
_need('rtti_type_descriptors', 'rtti_locators', 'rtti_base_classes')
|
||||
rows = c.execute("""SELECT c.address, c.vtable_address, c.class_hierarchy, c.subobject_offset
|
||||
FROM rtti_locators c
|
||||
JOIN rtti_type_descriptors td ON td.address = c.type_descriptor
|
||||
WHERE td.demangled_name = ?""", [name]).fetchall()
|
||||
if not rows:
|
||||
sys.exit(f'no RTTI class named {name!r} (try: zq.py classes {name})')
|
||||
for col, vt, chd, off in rows:
|
||||
print(f'== {name} (COL {H(col)}, subobject +0x{off:x})')
|
||||
bases = c.execute('SELECT base_index,name,mdisp,pdisp,vdisp FROM rtti_base_classes '
|
||||
'WHERE class_hierarchy=? AND base_index>0 ORDER BY base_index',
|
||||
[chd]).fetchall()
|
||||
for _, bn, md, pd, vd in bases:
|
||||
print(f' base {bn} mdisp={md} pdisp={pd} vdisp={vd}')
|
||||
if vt is None:
|
||||
print(' (no vtable located)')
|
||||
continue
|
||||
for slot, fa in c.execute('SELECT slot,function_address FROM methods '
|
||||
'WHERE vtable_address=? ORDER BY slot', [vt]).fetchall():
|
||||
print(f' vf{slot:<3} {H(fa)} {_fn(fa)}')
|
||||
|
||||
|
||||
def cmd_str(sub):
|
||||
sec = ', section' if _has_col('strings', 'section') else ", ''"
|
||||
rows = c.execute(f'SELECT address, encoding, content{sec} FROM strings '
|
||||
'WHERE content ILIKE ? ORDER BY address', [f'%{sub}%']).fetchall()
|
||||
for a, enc, content, section in rows:
|
||||
refs = c.execute("SELECT DISTINCT source_func FROM xrefs WHERE target=? AND source_func IS NOT NULL",
|
||||
[a]).fetchall()
|
||||
where = ', '.join(_fn(r[0]) for r in refs[:4]) or '(no xref)'
|
||||
print(f'{H(a)} [{enc}{"/" + section if section else ""}] {content!r}\n <- {where}')
|
||||
|
||||
|
||||
def cmd_xdbf(args):
|
||||
"""XDBF title text across every shipped locale."""
|
||||
sub = args[0] if args else ""
|
||||
rows = c.execute(
|
||||
"SELECT string_id, english, japanese FROM v_xdbf_text "
|
||||
"WHERE (? = '' OR english ILIKE '%' || ? || '%' OR japanese ILIKE '%' || ? || '%') "
|
||||
"ORDER BY string_id",
|
||||
[sub, sub, sub],
|
||||
).fetchall()
|
||||
for sid, en, ja in rows:
|
||||
print(f"{sid:6} {en or ''}")
|
||||
if ja and ja != en:
|
||||
print(f" ja: {ja}")
|
||||
print(f"({len(rows)} strings)")
|
||||
|
||||
|
||||
def cmd_ach(_args):
|
||||
"""XDBF achievements in the title's default language."""
|
||||
rows = c.execute(
|
||||
"SELECT id, gamerscore, name, unlocked_desc, locked_desc "
|
||||
"FROM xdbf_achievements ORDER BY id"
|
||||
).fetchall()
|
||||
total = 0
|
||||
for aid, gs, name, unlocked, locked in rows:
|
||||
total += gs or 0
|
||||
print(f"{aid:3} | {gs:3}G | {name}")
|
||||
print(f" unlocked: {unlocked}")
|
||||
print(f" locked : {locked}")
|
||||
print(f"\n{len(rows)} achievements, {total}G")
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print(__doc__); return
|
||||
cmd, args = sys.argv[1], sys.argv[2:]
|
||||
if cmd == 'dis':
|
||||
cmd_dis(int(args[0], 16), int(args[1], 16))
|
||||
elif cmd == 'fn':
|
||||
print(_fn(int(args[0], 16)))
|
||||
elif cmd == 'xref':
|
||||
t = int(args[0], 16)
|
||||
for s, k, i, sf in c.execute('SELECT source,kind,instruction,source_func FROM xrefs '
|
||||
'WHERE target=? ORDER BY source', [t]).fetchall():
|
||||
print(H(s), k, 'in', _fn(s), ':', i)
|
||||
elif cmd == 'callers':
|
||||
off = int(args[0]) # decimal byte offset, e.g. 196 for vtable[49]
|
||||
pat = f'r11, {off}(r11)'
|
||||
for (a,) in c.execute("SELECT address FROM instructions WHERE mnemonic='lwz' AND operands=? "
|
||||
'ORDER BY address', [pat]).fetchall():
|
||||
print(H(a), 'in', _fn(a))
|
||||
elif cmd == 'grep':
|
||||
for a, m, o in c.execute("SELECT address,mnemonic,operands FROM instructions "
|
||||
"WHERE operands LIKE ? ORDER BY address", [f'%{args[0]}%']).fetchall():
|
||||
print(H(a), m, o, ' in', _fn(a))
|
||||
elif cmd == 'find':
|
||||
for (a,) in c.execute('SELECT address FROM instructions WHERE raw=? ORDER BY address',
|
||||
[int(args[0], 16)]).fetchall():
|
||||
print(H(a))
|
||||
elif cmd == 'switch':
|
||||
cmd_switch(int(args[0], 16))
|
||||
elif cmd == 'switches':
|
||||
cmd_switches(int(args[0], 16) if args else None)
|
||||
elif cmd == 'classes':
|
||||
cmd_classes(args[0] if args else None)
|
||||
elif cmd == 'class':
|
||||
cmd_class(args[0])
|
||||
elif cmd == 'str':
|
||||
cmd_str(args[0])
|
||||
elif cmd == 'xdbf':
|
||||
cmd_xdbf(args)
|
||||
elif cmd == 'ach':
|
||||
cmd_ach(args)
|
||||
else:
|
||||
print(__doc__)
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user