e909c7c133
Nothing here changes what a tool computes; it changes where tools look. - tools/re-capture: 33 censuses globbed /work/sylph_extract, a path that has existed nowhere since /work became a clone, so they matched nothing and printed empty results. They now resolve the disc through a new disc.py from $SYLPHEED_DISC and exit loudly without it (the #44 fix, generalised). Nine scripts that imported siblings from the retired Reborn checkout or an old session scratchpad now import from their own directory. unitgroup.py only needs the variable when --pak is not given. - sylpheed-xex: the loader only ever uses the XEX2 retail key. The dead devkit key and a doc comment claiming a devkit fallback that does not exist are gone; Project Sylpheed is a retail XEX2, so no XEX1 key either. - sylpheed-viewer: real_font_rasterizes looked for /tmp/sylph_extract and so always skipped. It reads $SYLPHEED_DISC now, and passes against the disc. - Comments and docs that named xenia-rs, the Reborn repository or /work/*.pe as places to look now name sylpheed.db, Canary's ppc_context.h and the flat .pe; docs/re/README.md no longer says the native Canary build does not run. Historical records keep their original paths: findings that were measured against /work/xenia-rs/sylpheed.db still say so. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
374 lines
17 KiB
Python
Executable File
374 lines
17 KiB
Python
Executable File
#!/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.
|
|
|
|
The database is found via `$SYLPHEED_DB` (alias `$SYLPH_XEXDB`), else `<repo root>/sylpheed.db`. It is a
|
|
build artefact, not a tracked file; if neither exists, zq.py says so and prints
|
|
the command that builds one.
|
|
"""
|
|
import duckdb, sys, os, pathlib, 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)
|
|
|
|
REGEN = "sylph-xexdb dis <xex|iso> --db sylpheed.db --analyze sql"
|
|
|
|
|
|
def _resolve_db():
|
|
"""Locate the database: `$SYLPHEED_DB` / `$SYLPH_XEXDB`, else `<repo root>/sylpheed.db`.
|
|
|
|
🔴 This used to be one hardcoded absolute path, and it pointed *inside*
|
|
`xenia-rs` -- a repository that has since been archived. It resolved on
|
|
exactly one machine and would have started failing there too, with `duckdb` raising about a missing file rather
|
|
than anything saying why.
|
|
|
|
The database is a build artefact of several hundred MB and is not in the
|
|
repository, so there is nothing to fall back *to*: if neither location has
|
|
one, say so and print how to build it. A default that silently resolves to
|
|
the wrong database is worse than no default -- see issue #16.
|
|
"""
|
|
# `SYLPHEED_DB` is the documented contract (the decoder container sets it);
|
|
# `SYLPH_XEXDB` was this script's own name for it and stays as an alias.
|
|
for var in ('SYLPHEED_DB', 'SYLPH_XEXDB'):
|
|
env = os.environ.get(var)
|
|
if env:
|
|
p = pathlib.Path(env).expanduser()
|
|
if not p.is_file():
|
|
sys.exit(f'${var} is set but is not a file:\n {p}')
|
|
return p
|
|
# Relative to this script, not to the caller's cwd: `zq.py` is run from
|
|
# wherever the investigation happens to be.
|
|
p = pathlib.Path(__file__).resolve().parent.parent / 'sylpheed.db'
|
|
if p.is_file():
|
|
return p
|
|
sys.exit(
|
|
f'no database found.\n'
|
|
f' looked for: {p}\n'
|
|
f' set $SYLPHEED_DB to an existing one, or build it with:\n'
|
|
f' {REGEN}'
|
|
)
|
|
|
|
|
|
# Resolved lazily enough that `zq.py` with no command still prints its usage on
|
|
# a machine that has no database yet.
|
|
_WANTS_DB = len(sys.argv) > 1 and sys.argv[1] not in ('-h', '--help', 'help')
|
|
DB = _resolve_db() if _WANTS_DB else None
|
|
c = duckdb.connect(str(DB), read_only=True) if DB is not None else None
|
|
H = lambda x: '0x%08x' % x
|
|
|
|
|
|
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()
|