Sylpheed/tools/zq.py

#!/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.

🔴 THAT LAST LINE USED TO SAY "read it from guest memory with
`xenia-rs exec ... --dump-addr=0x<va>`". That emulator is retired, and the
command no longer exists -- see docs/agents/CONSOLIDATION.md. Two replacements,
both static, neither needing anything to run:

  * the extracted PE is a FLAT VA DUMP: byte offset = VA - 0x82000000, so
    `dd`/`xxd` on `*.pe` reads any address directly;
  * `sylph-xexdb extract <xex>` regenerates that `.pe` (byte-identical, verified)
    alongside a metadata JSON.

Usage:
  zq.py dis  <lo_hex> <hi_hex>      # disassemble [lo,hi)  (data words as .long; imports named)
  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 gaps                        # dangling refs = known detector gaps (see db.rs schema docs)
  zq.py imp  [substr]               # imports + how many call sites each has
  zq.py impcalls <substr>           # every call site of the matching imports

  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, signal

# Output is routinely piped into `head`; without this Python prints a
# BrokenPipeError traceback when the reader closes early.
signal.signal(signal.SIGPIPE, signal.SIG_DFL)

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 _import_names():
    """thunk head VA -> 'library::name' for every import, or {} on an old db."""
    if not _has_col('instructions', 'import_address'):
        return {}
    return {a: (f'{lib}::{nm}' if nm else f'{lib}::ordinal_{o:#06x}')
            for a, lib, nm, o in c.execute(
                'SELECT address,library,name,ordinal FROM imports').fetchall()}


def cmd_dis(lo, hi):
    data_col = 'is_data' if _has_col('instructions', 'is_data') else 'false'
    has_imp = _has_col('instructions', 'import_address')
    imp_cols = 'import_address,import_role' if has_imp else 'NULL,NULL'
    names = _import_names()
    rows = c.execute(f'SELECT address,mnemonic,operands,raw,{data_col},{imp_cols} '
                     'FROM instructions WHERE address>=? AND address<? ORDER BY address',
                     [lo, hi]).fetchall()
    for a, m, o, raw, is_data, imp_a, imp_role in rows:
        who = names.get(imp_a, H(imp_a) if imp_a else None)
        if is_data:
            # is_data covers two different hazards; say which one this is.
            why = f'import record -> {who}' if imp_role == 'record' else 'jump-table data'
            print(H(a), '.long', H(raw & 0xffffffff), f' ; {why}')
        elif imp_role == 'call':
            print(H(a), m, o, f'  ; -> IMPORT {who}')
        elif imp_role == 'thunk':
            print(H(a), m, o, f'  ; import thunk {who}')
        else:
            print(H(a), m, o)


def cmd_gaps():
    """Dangling references the schema deliberately does NOT declare as FKs.

    Each of these is a real gap in a detector, kept queryable rather than
    hidden: declaring the FK would abort the build instead of reporting it.
    """
    checks = [
        ("vtables RTTI names but M3 missed",
         "SELECT count(*) FROM rtti_locators l WHERE l.vtable_address IS NOT NULL "
         "AND NOT EXISTS (SELECT 1 FROM vtables v WHERE v.address=l.vtable_address)"),
        ("vtable slots -> undetected function",
         "SELECT count(DISTINCT function_address) FROM methods m WHERE NOT EXISTS "
         "(SELECT 1 FROM functions f WHERE f.address=m.function_address)"),
        ("funcptr-array slots -> undetected function",
         "SELECT count(DISTINCT function_address) FROM function_pointer_array_entries e "
         "WHERE NOT EXISTS (SELECT 1 FROM functions f WHERE f.address=e.function_address)"),
        ("dispatch candidates -> undetected function",
         "SELECT count(DISTINCT method_address) FROM indirect_dispatch_candidates x "
         "WHERE NOT EXISTS (SELECT 1 FROM functions f WHERE f.address=x.method_address)"),
        ("virtual sites with no candidates (truncated)",
         "SELECT count(*) FROM indirect_dispatch_sites WHERE truncated"),
    ]
    for label, q in checks:
        try:
            print(f"{c.execute(q).fetchone()[0]:>7}  {label}")
        except Exception as e:
            print(f"      ?  {label} ({str(e).splitlines()[0][:50]})")
    print("\n  -- RTTI-named vftables the vtable scanner missed --")
    try:
        rows = c.execute(
            "SELECT l.vtable_address, t.demangled_name FROM rtti_locators l "
            "JOIN rtti_type_descriptors t ON t.address=l.type_descriptor "
            "WHERE l.vtable_address IS NOT NULL AND NOT EXISTS "
            "(SELECT 1 FROM vtables v WHERE v.address=l.vtable_address) "
            "ORDER BY 2 LIMIT 15").fetchall()
        for a, nm in rows:
            print(f"    {H(a)}  {nm}")
    except Exception as e:
        print("   ", str(e).splitlines()[0][:70])


def cmd_imp(pat=None):
    """Import call sites, grouped by import. `pat` filters on library::name."""
    _need('imports')
    if not _has_col('instructions', 'import_address'):
        sys.exit(f'this db predates instructions.import_address — regenerate with:\n  {REGEN}')
    q = ("SELECT im.library, im.name, im.ordinal, im.address, count(i.address) "
         "FROM imports im LEFT JOIN instructions i "
         "  ON i.import_address = im.address AND i.import_role = 'call' "
         "WHERE im.record_type = 1 GROUP BY 1,2,3,4 ORDER BY 5 DESC, 1, 2")
    for lib, nm, o, addr, n in c.execute(q).fetchall():
        label = f'{lib}::{nm}' if nm else f'{lib}::ordinal_{o:#06x}'
        if pat and pat.lower() not in label.lower():
            continue
        print(f'{n:>5} calls  {H(addr)}  {label}')


def cmd_impcalls(pat):
    """Every call site of the imports matching `pat`."""
    if not _has_col('instructions', 'import_address'):
        sys.exit(f'this db predates instructions.import_address — regenerate with:\n  {REGEN}')
    q = ("SELECT i.address, i.mnemonic, im.library, im.name, im.ordinal "
         "FROM instructions i JOIN imports im ON im.address = i.import_address "
         "WHERE i.import_role = 'call' ORDER BY i.address")
    for a, m, lib, nm, o in c.execute(q).fetchall():
        label = f'{lib}::{nm}' if nm else f'{lib}::ordinal_{o:#06x}'
        if pat.lower() not in label.lower():
            continue
        print(H(a), m, 'in', _fn(a), '->', label)


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 == 'gaps':
        cmd_gaps()
    elif cmd == 'imp':
        cmd_imp(args[0] if args else None)
    elif cmd == 'impcalls':
        cmd_impcalls(args[0])
    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()