feat(assets): solve asset resolution (SYS4INI per-scene section manifest)

resId -> files[section_base(scene) + resId]. SYS4INI's file list is
sectioned, one per scene (SCxxxx.BIN + its cross-archive asset manifest);
file_number is the index within the section. Unified for set-texture,
play-bgm, play-voice. Fully static/general -> no per-scene capture.

- tools/parse_sys4ini.py: SYS4INI (S4IC422, LZSS) -> build/asset-index.json
- tools/resolve_asset.py: sections + (scene,resId) resolver -> build/asset-sections.json
- validated: 97% structural, SC0000 17/17 vs Frida, 586/595 captured loads
- opcodes.toml: set-texture/create/draw-texture, play-bgm/voice enriched (frida-grounded)
- Frida tooling (capture_load_order all-archive, correlate_scope, ...) + vm0 --settex
- docs: asset-resolution-re (step2 SOLVED), global-memory-re (shelved), tools-reference

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gamer147
2026-07-06 20:48:22 -04:00
parent b92e815850
commit a61c0c9abd
18 changed files with 1887 additions and 87 deletions

148
tools/correlate_scope.py Normal file
View File

@@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""Correlate the VM's set-texture(resId) trace with the game's Frida load order to LOCALIZE the
asset-resolution scope selector (docs/asset-resolution-re.md step 2, scope-selector hunt).
Inputs:
build/settex-<SCENE>.json our VM's ordered set-texture(resId) trace + bytecode offsets
(produce with: py -3.11 -X utf8 tools/vm0.py --settex <SCENE>)
build/frida-load-order-result.json the game's ordered (name, file_number=resId) loads
(produce with tools/frida/capture_load_order.py --analyze)
build/asset-index.json for segmenting DATA2 into packages
Method: `resId == file_number`, and each loaded file belongs to exactly one DATA2 "package"
(a monotonic-file_number run). So the game's load order is a readout of the active package over
time. We greedily align each Frida load to the next same-resId set-texture in the VM trace, which
pins it to a bytecode offset. Where the active package CHANGES between consecutive loads, some
instruction in the bytecode span between their offsets flipped the scope -- the scope selector.
This report lists the aligned loads, flags package transitions, and points at the byte spans to
inspect (dump them with the disassembly in build/disasm/<SCENE>.asm).
Usage: py -3.11 -X utf8 tools/correlate_scope.py <SCENE> (e.g. SC0000)
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import paths
def packages():
"""Segment DATA2 (directory order) into monotonic-file_number runs; return name->pkg index."""
idx = json.loads((paths.BUILD / "asset-index.json").read_text(encoding="utf-8"))
d2 = [f for f in idx["files"] if f["archive"] == "DATA2.ALF"]
name2pkg, pkg, prev = {}, 0, None
for f in d2:
fn = f["file_number"]
if prev is not None and fn <= prev:
pkg += 1
name2pkg[f["name"]] = pkg
prev = fn
return name2pkg
# boilerplate ops to hide when dumping the between-loads span (leave the structural/effectful ones)
NOISE = {"mov", "add", "sub", "mul", "div", "mod", "and", "or", "sar", "shl", "xor",
"eq", "ne", "lt", "lte", "gr", "gre", "jcc", "jmp", "set-string",
"show-text", "end-text-line", "wait-for-input", "stmt-begin", "stmt-end",
"block-mark", "cond-block", "label-def", "line-id?", "comment", "set-font",
"lookup-array", "lookup-array-2d", "stmt-desc?", "bit-set", "bit-reset",
"check-bit", "text-param?", "gfx-geom?", "count?", "resolve-handle?",
"create-texture", "draw-texture", "draw-string", "draw?", "draw-blit?"}
def disasm_map(scene):
"""offset(int) -> stripped disasm line, from build/disasm/<SCENE>.asm."""
path = paths.BUILD / "disasm" / f"{scene}.asm"
out = {}
if not path.exists():
return out
for ln in path.read_text(encoding="utf-8").splitlines():
s = ln.strip()
if s[:2] == "0x" and ":" in s:
off = int(s.split(":", 1)[0], 16)
out[off] = s.split(":", 1)[1].strip()
return out
def dump_span(trace, ta, tb, dis):
"""Print the significant (non-NOISE) ops executed between trace indices ta..tb."""
seen = 0
for t in range(ta, min(tb, len(trace))):
off = int(trace[t], 16)
line = dis.get(off, "")
mnem = line.split()[0] if line else ""
if mnem and mnem not in NOISE:
print(f" {trace[t]:>8} {line}")
seen += 1
if not seen:
print(" (no structural/effectful ops in span — only boilerplate)")
def main() -> int:
args = [a for a in sys.argv[1:] if not a.startswith("-")]
if not args:
raise SystemExit(__doc__)
scene = args[0].upper().removesuffix(".BIN")
settex_path = paths.BUILD / f"settex-{scene}.json"
loads_path = paths.BUILD / "frida-load-order-result.json"
if not settex_path.exists():
raise SystemExit(f"missing {settex_path.name} — run: tools/vm0.py --settex {scene}")
if not loads_path.exists():
raise SystemExit(f"missing {loads_path.name} — run tools/frida/capture_load_order.py --analyze")
sx = json.loads(settex_path.read_text(encoding="utf-8"))
vm, trace = sx["settex"], sx.get("trace", []) # [{i,off,resId,slot,trace_i}]
loads = json.loads(loads_path.read_text(encoding="utf-8"))["load_order"] # [{name,file_number}]
name2pkg = packages()
dis = disasm_map(scene)
# greedy align: each Frida load -> next same-resId set-texture in VM order
aligned, j, unmatched = [], 0, 0
for ld in loads:
fn = ld["file_number"]
k = j
while k < len(vm) and vm[k]["resId"] != fn:
k += 1
if k < len(vm):
aligned.append({"vm": vm[k], "resId": fn, "name": ld["name"], "pkg": name2pkg.get(ld["name"])})
j = k + 1
else:
aligned.append({"vm": None, "resId": fn, "name": ld["name"], "pkg": name2pkg.get(ld["name"])})
unmatched += 1
print(f"scene {scene}: {len(vm)} VM set-textures, {len(loads)} Frida loads, "
f"{len(loads)-unmatched} aligned ({unmatched} unmatched)\n")
print(f"{'off':>8} {'resId':>5} {'pkg':>4} name")
prev = None
transitions = []
for a in aligned:
off = a["vm"]["off"] if a["vm"] else None
flag = ""
if prev is not None and a["pkg"] != prev["pkg"]:
flag = f" <<< PACKAGE {prev['pkg']} -> {a['pkg']}"
transitions.append((prev, a))
print(f"{str(off):>8} {a['resId']:>5} {str(a['pkg']):>4} {a['name']}{flag}")
prev = a
print()
if not transitions:
print("No package transitions in this capture — play deeper into the scene to cross one.")
return 0
print(f"{len(transitions)} package transition(s). Significant ops executed across each "
f"(the scope selector should be here):\n")
for pa, pb in transitions:
print(f" === pkg {pa['pkg']} ({pa['name']}) -> pkg {pb['pkg']} ({pb['name']}) ===")
if pa["vm"] and pb["vm"]:
dump_span(trace, pa["vm"]["trace_i"], pb["vm"]["trace_i"], dis)
else:
print(" (unaligned — cannot pin the span)")
print()
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -21,15 +21,30 @@ Prereq: `py -3.11 -m pip install frida` (core only — `frida-tools` CLI is not
## Tools
- `capture_graphics.py`hooks `ReadFile` on the graphics archives (`DATA2/DATA5*.ALF`), resolving
each handle→path via `GetFinalPathNameByHandleW` (cached) + the read offset. Log →
`build/frida-reads.log` (`path<TAB>offset<TAB>size`).
- **`capture_load_order.py`**★ the working asset-resolution capture. Hooks `ReadFile` on
`DATA2.ALF`; each asset load starts with header reads **at its exact archive offset**, so exact-start
reads give the clean per-asset **load order** (→ names via `build/asset-index.json`). `--analyze`
prints the order + file_number. This **confirmed `resId == SYS4INI file_number`** (the load order
matches our engine's `set-texture(resId)` order 1:1). Attach by pid, replay the scene, Ctrl-C, `--analyze`.
- `capture_graphics.py` — original `ReadFile` logger for `DATA2/DATA5` (`path<TAB>offset<TAB>size`
`build/frida-reads.log`); pair with `tools/resolve_frida_reads.py`.
- `locate_resource_load.py` — phase-1 locator (back-trace asset-opens → native load chain).
- `capture_resid_args.py` — phase-2 probe (decoder args/context; showed the loader carries only offsets).
- `find_globals_base.py`, `find_global_by_sequence.py`**runtime global-variable RE (SHELVED)**. Full
write-up, findings, memory landmarks, and the recommended resume plan: **`docs/global-memory-re.md`**.
TL;DR: the VM global memory is structured/multi-store (not flat int32) and `G[0x62424]` is a transient
arg-register — heuristic value scans can't pin it. Resume from a *stable* anchor (the name global
`0x279`) and target the interpreter's address resolution, not scans.
## Known limitations (see docs/asset-resolution-re.md)
## Runtime architecture (learned 2026-07-06 — read before writing new hooks)
- **File-I/O offsets are noisy** — spans don't match extracted AGF sizes; the game likely
**memory-maps** the archives (so `ReadFile` offsets are OS paging noise, not clean per-asset
loads) and/or uses async/`OVERLAPPED` reads. The robust hook is the game's **internal
load-by-id function** (find via the opcode dispatch), not file I/O — a future tool.
- Correlation still needs the **`SYS4INI` (S4IC422) asset index** parsed to turn an archive offset
into a filename. That parser is the first foundational RE step.
- **Attach, don't spawn**, and **use the pid** (`frida.get_local_device().enumerate_processes()` — the
module-level `frida.enumerate_processes()` was removed in frida 17.x). The process is `AGE.EXE`.
- **The game is packed.** Its main VM logic runs from a per-run heap `r-x` region (~30 MB, nonstable
base). So you **cannot** hook the `set-texture`/VM handlers at a fixed `AGE.EXE+off` — only stable
library code (e.g. the AGF decoder `AGE.EXE+0x74f1f`) keeps a fixed offset.
- **Archives are NOT memory-mapped.** No archive-sized region exists; the game streams them through a
small heap **block-cache via `ReadFile`** (128 KB blocks + big reads). The earlier "memory-mapped"
note was wrong — the 128 KB reads are the block cache, not OS paging.
- **The reliable oracle is the `ReadFile` offset stream** → names via the SYS4INI index
(`tools/parse_sys4ini.py``build/asset-index.json`). Native-handler hooking is blocked by the packer.

View File

@@ -0,0 +1,168 @@
#!/usr/bin/env python3
"""Recover the game's per-asset LOAD ORDER across ALL archives (docs/asset-resolution-re.md).
Assets are typed by name prefix and split across archives: DATA1 = CS/CP/CB/CA/BG/... (ADV
sprites, map sprites, battle & icon portraits, backgrounds), DATA2 = EV/EVM event CGs,
DATA5 = movies. Resolution is per-(archive, type): a set-texture(resId, slot) picks a type via
the slot, and resId indexes within that type. So we must watch EVERY DATA*.ALF, not just DATA2.
Hooks ReadFile on all DATA*.ALF; each asset load begins with header reads at its exact archive
offset, so exact-start reads give the ordered load list -> resolved to (archive, name, prefix,
file_number) via build/asset-index.json. Feed the result to tools/correlate_scope.py to align
with the VM's set-texture(resId) trace and learn the type/scope rule.
Flow: game running -> attach -> replay a scene -> Ctrl-C -> --analyze.
Output: build/frida-load-order.jsonl, build/frida-load-order-result.json.
"""
import json
import re
import sys
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
OUT = REPO / "build" / "frida-load-order.jsonl"
INDEX = REPO / "build" / "asset-index.json"
def prefix(name):
m = re.match(r"([A-Za-z]+)", name)
return m.group(1) if m else "?"
def index_by_archive():
"""{archive: {offset: (name, file_number, prefix)}} and sorted offset lists for containment."""
idx = json.loads(INDEX.read_text(encoding="utf-8"))
exact, arr = {}, {}
for f in idx["files"]:
a = f["archive"]
exact.setdefault(a, {})[f["offset"]] = (f["name"], f["file_number"], prefix(f["name"]))
arr.setdefault(a, []).append((f["offset"], f["size"], f["name"], f["file_number"]))
for a in arr:
arr[a].sort()
return exact, arr
JS = r"""
const k32 = Process.getModuleByName('kernel32.dll');
const GetFinalPathNameByHandleW = new NativeFunction(
k32.findExportByName('GetFinalPathNameByHandleW'), 'uint32', ['pointer','pointer','uint32','uint32']);
const SetFilePointer = new NativeFunction(
k32.findExportByName('SetFilePointer'), 'uint32', ['pointer','int32','pointer','uint32']);
const NUL = ptr(0); const cache = {};
function pathOf(h){ const k=h.toString(); let v=cache[k]; if(v!==undefined) return v; let p=null;
try{ const b=Memory.alloc(1040); const n=GetFinalPathNameByHandleW(h,b,519,0);
if(n>0&&n<519) p=b.readUtf16String(); }catch(e){} cache[k]=p; return p; }
Interceptor.attach(k32.findExportByName('ReadFile'), {
onEnter(args){
const p = pathOf(args[0]); if(!p) return;
const m = /DATA(\d)\.ALF$/i.exec(p); if(!m) return;
const size = args[2].toInt32();
const ov = args[4]; let off = -1;
try{ off = ov.isNull()? SetFilePointer(args[0],0,NUL,1) : ov.add(8).readU32(); }catch(e){}
if(off < 0) return;
send({kind:'read', archive:'DATA'+m[1]+'.ALF', offset: off, size: size});
}
});
send({ready:true});
"""
def analyze():
if not OUT.exists():
raise SystemExit(f"no log: {OUT}")
import bisect
exact, arr = index_by_archive()
recs = [json.loads(l) for l in OUT.read_text(encoding="utf-8").splitlines() if l.strip()]
reads = [r for r in recs if r.get("kind") == "read"]
per_arc = {}
for r in reads:
per_arc[r["archive"]] = per_arc.get(r["archive"], 0) + 1
print(f"{len(reads)} reads across archives: {per_arc}\n")
def contain(a, off):
v = arr.get(a)
if not v:
return None
i = bisect.bisect_right(v, (off, float("inf"), "", 0)) - 1
if i < 0:
return None
o, s, n, fn = v[i]
return (n, fn) if off < o + s else None
# exact-start order (unambiguous), plus containment first-touch (fuller, noisier)
exact_order, contain_order, seen_e, seen_c = [], [], set(), set()
for r in reads:
a, off = r["archive"], r["offset"]
if off in exact.get(a, {}):
name, fn, pfx = exact[a][off]
if name not in seen_e:
seen_e.add(name); exact_order.append({"archive": a, "name": name, "file_number": fn, "prefix": pfx})
hit = contain(a, off)
if hit and hit[0] not in seen_c:
seen_c.add(hit[0])
contain_order.append({"archive": a, "name": hit[0], "file_number": hit[1], "prefix": prefix(hit[0])})
print(f"=== exact-start load order ({len(exact_order)}) ===")
for e in exact_order:
print(f" {e['archive'][:5]} {e['prefix']:<4} {e['name']:<14} fn={e['file_number']} (resId 0x{e['file_number']:x})")
print(f"\n=== containment first-touch ({len(contain_order)}, includes prefetch noise) ===")
for e in contain_order:
print(f" {e['archive'][:5]} {e['prefix']:<4} {e['name']:<14} fn={e['file_number']}")
res = {"load_order": exact_order, "containment_order": contain_order,
"reads_per_archive": per_arc}
(REPO / "build" / "frida-load-order-result.json").write_text(
json.dumps(res, ensure_ascii=False, indent=1), encoding="utf-8")
print("\n-> build/frida-load-order-result.json")
return 0
def capture(proc):
import frida
OUT.parent.mkdir(parents=True, exist_ok=True)
log = open(OUT, "w", encoding="utf-8")
exact, _ = index_by_archive()
def on_message(msg, data):
if msg.get("type") == "error":
print("[frida-error]", msg.get("description")); return
if msg.get("type") != "send":
return
pl = msg["payload"]
if pl.get("ready"):
print("[frida] ReadFile hook live on ALL DATA*.ALF — replay the scene now."); return
log.write(json.dumps(pl, ensure_ascii=False) + "\n"); log.flush()
if pl.get("kind") == "read":
hit = exact.get(pl["archive"], {}).get(pl["offset"])
if hit:
print(f"LOAD {pl['archive'][:5]} {hit[2]:<4} {hit[0]} fn={hit[1]}")
target = int(proc) if str(proc).isdigit() else proc
try:
session = frida.attach(target)
except frida.ProcessNotFoundError:
procs = frida.get_local_device().enumerate_processes()
print("AGE-like:", [(p.pid, p.name) for p in procs if "age" in p.name.lower()])
return 2
script = session.create_script(JS)
script.on("message", on_message)
script.load()
print(f"[frida] attached to {proc}; logging all-archive reads -> {OUT}")
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
print("\n[frida] stopped. Now: py -3.11 -X utf8 tools/frida/capture_load_order.py --analyze")
return 0
def main():
if "--analyze" in sys.argv:
return analyze()
proc = next((a for a in sys.argv[1:] if not a.startswith("-")), "AGE.EXE")
return capture(proc)
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,194 @@
#!/usr/bin/env python3
"""Phase 2 of the native resId->filename crack (see docs/asset-resolution-re.md step 2a).
Phase 1 (locate_resource_load.py) found the stable asset-load call chain in AGE.EXE:
AGE.EXE+0x16d5d7 -> AGE.EXE+0x74f1f -> AGE.EXE+0x397b -> ReadFile
This hooks the two upper frames and, at entry, dumps their arguments (raw dwords + any
ASCII a pointer argument targets), tagged with thread id. It also keeps the ReadFile ->
offset -> asset-name resolver. Interleaving the two streams lets us pair each load call
with the asset it produced and find which argument carries the resId (== 37 for EV052CA,
39 EV052DA, 43 EV052DC, 46 EV052DB) or a pointer to the SYS4INI entry (name / file_number).
Once identified, that argument IS the resId->name mapping at the source. Read-only.
Flow: game running -> this attaches -> replay the opening CGs -> Ctrl-C -> --analyze.
Output: build/frida-resid-args.jsonl (ordered events).
"""
import json
import sys
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
OUT = REPO / "build" / "frida-resid-args.jsonl"
INDEX = REPO / "build" / "asset-index.json"
HANDLER_OFFSETS = [0x74f1f] # phase 1 chain; 0x16d5d7 is a return addr (not a callable entry)
def data2_offsets():
idx = json.loads(INDEX.read_text(encoding="utf-8"))
return {f["offset"]: (f["name"], f["file_number"])
for f in idx["files"] if f["archive"] == "DATA2.ALF"}
JS_TEMPLATE = r"""
const OFFSETS = new Set(__OFFSETS__);
const HANDLERS = __HANDLERS__;
const PS = Process.pointerSize;
const age = Process.getModuleByName('AGE.EXE');
// --- load-handler arg dumps ---
function scanAscii(base, len) { // printable ASCII runs (>=4) as "hexoff:text"
const out = [];
try {
const u = new Uint8Array(base.readByteArray(len));
let start = -1;
for (let i = 0; i <= u.length; i++) {
const c = i < u.length ? u[i] : 0;
if (c >= 0x20 && c < 0x7f) { if (start < 0) start = i; }
else { if (start >= 0 && i - start >= 4)
out.push(start.toString(16) + ':' + String.fromCharCode.apply(null, u.subarray(start, i)));
start = -1; }
}
} catch (e) {}
return out;
}
HANDLERS.forEach(function(off) {
const addr = age.base.add(off);
Interceptor.attach(addr, {
onEnter(args) {
const sp = this.context.sp, ebp = this.context.ebp;
const raw = [], cargs = [];
for (let i = 1; i <= 8; i++) { // [sp+PS*i] = this fn's arg i
try { raw.push(sp.add(PS * i).readPointer().toString()); } catch (e) { raw.push('0x0'); }
}
for (let i = 2; i <= 9; i++) { // [ebp+PS*i] = CALLER's args (ebp is caller's at entry)
try { cargs.push(ebp.add(PS * i).readPointer().toString()); } catch (e) { cargs.push('0x0'); }
}
// scan the loader context object (arg1) and any pointer arg's target for ASCII (filenames?)
const ctx = scanAscii(ptr(raw[0]), 0x400);
const argStr = {};
raw.concat(cargs).forEach(function(v, k) {
const s = scanAscii(ptr(v), 0x48);
if (s.length) argStr[k] = s;
});
send({kind: 'call', off: '0x' + off.toString(16), tid: this.threadId,
ret: '0x' + this.returnAddress.sub(age.base).toString(16),
args: raw, cargs: cargs, ctx: ctx, argStr: argStr});
}
});
});
// --- ReadFile -> asset-start resolver (ground-truth name per load) ---
const k32 = Process.getModuleByName('kernel32.dll');
const GetFinalPathNameByHandleW = new NativeFunction(
k32.findExportByName('GetFinalPathNameByHandleW'), 'uint32', ['pointer','pointer','uint32','uint32']);
const SetFilePointer = new NativeFunction(
k32.findExportByName('SetFilePointer'), 'uint32', ['pointer','int32','pointer','uint32']);
const NUL = ptr(0); const cache = {};
function pathOf(h) { const key = h.toString(); let v = cache[key]; if (v !== undefined) return v;
let p = null; try { const buf = Memory.alloc(1040);
const n = GetFinalPathNameByHandleW(h, buf, 519, 0);
if (n > 0 && n < 519) p = buf.readUtf16String(); } catch (e) {} cache[key] = p; return p; }
Interceptor.attach(k32.findExportByName('ReadFile'), {
onEnter(args) {
const p = pathOf(args[0]); if (!p || !/data2\.alf$/i.test(p)) return;
const size = args[2].toInt32(); if (size > 4096) return;
const ov = args[4]; let off = -1;
try { off = ov.isNull() ? SetFilePointer(args[0], 0, NUL, 1) : ov.add(8).readU32(); } catch (e) {}
if (!OFFSETS.has(off)) return;
send({kind: 'read', tid: this.threadId, offset: off});
}
});
send({ready: true});
"""
def analyze():
if not OUT.exists():
raise SystemExit(f"no log: {OUT}")
import re
recs = [json.loads(l) for l in OUT.read_text(encoding="utf-8").splitlines() if l.strip()]
calls = [r for r in recs if r["kind"] == "call"]
reads = [r for r in recs if r["kind"] == "read"]
print(f"{len(calls)} decode calls, {len(reads)} reads\n")
# every ASCII string the hook surfaced (ctx object + pointer-arg targets)
def strings_of(c):
out = list(c.get("ctx", []))
for v in c.get("argStr", {}).values():
out.extend(v)
return [s.split(":", 1)[1] for s in out]
seen = {}
for c in calls:
for s in strings_of(c):
seen[s] = seen.get(s, 0) + 1
evlike = {s: n for s, n in seen.items() if re.search(r"EVM?\d|\.AGF|AGF", s, re.I)}
print(f"distinct ASCII strings surfaced: {len(seen)}")
print(f"asset-name-like strings ({len(evlike)}):")
for s, n in sorted(evlike.items(), key=lambda kv: -kv[1])[:60]:
print(f" x{n:<4} {s!r}")
if not evlike:
print(" (none — filename not in the context object; try the caller frame / go one level up)")
print("\n top non-EV strings (for orientation):")
for s, n in sorted(seen.items(), key=lambda kv: -kv[1])[:20]:
print(f" x{n:<4} {s!r}")
return 0
def capture(proc):
import frida
offs = data2_offsets()
js = (JS_TEMPLATE
.replace("__OFFSETS__", json.dumps(sorted(offs.keys())))
.replace("__HANDLERS__", json.dumps(HANDLER_OFFSETS)))
OUT.parent.mkdir(parents=True, exist_ok=True)
log = open(OUT, "w", encoding="utf-8")
def on_message(msg, data):
if msg.get("type") == "error":
print("[frida-error]", msg.get("description")); return
if msg.get("type") != "send":
return
pl = msg["payload"]
if pl.get("ready"):
print("[frida] resId-arg hooks live — replay the opening CGs now."); return
log.write(json.dumps(pl, ensure_ascii=False) + "\n"); log.flush()
if pl["kind"] == "read":
name, fn = offs.get(pl["offset"], ("?", -1))
print(f"READ {name} (fn={fn})")
else:
import re
names = [s.split(":", 1)[1] for s in pl.get("ctx", [])
if re.search(r"EVM?\d|\.AGF", s.split(":", 1)[1], re.I)]
if names:
print(f" call ret=AGE.EXE+{pl['ret']} names={names}")
target = int(proc) if proc.isdigit() else proc
try:
session = frida.attach(target)
except frida.ProcessNotFoundError:
procs = frida.get_local_device().enumerate_processes()
print(f"'{proc}' not found. AGE-like:", [(p.pid, p.name) for p in procs if "age" in p.name.lower()])
return 2
script = session.create_script(js)
script.on("message", on_message)
script.load()
print(f"[frida] attached to {proc}; hooks at AGE.EXE+{[hex(h) for h in HANDLER_OFFSETS]}; log -> {OUT}")
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
print("\n[frida] stopped. Now: py -3.11 -X utf8 tools/frida/capture_resid_args.py --analyze")
return 0
def main():
if "--analyze" in sys.argv:
return analyze()
proc = next((a for a in sys.argv[1:] if not a.startswith("-")), "AGE.EXE")
return capture(proc)
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,141 @@
#!/usr/bin/env python3
"""Locate the VM int-global G[0x62424] (the CG resId) in memory by a differential value scan --
the confirmed-anchor bootstrap for runtime global observation (docs/asset-resolution-re.md step 2).
G[0x62424] is unambiguously a VM global-int (`mov (global-int 0x62424) ...`), and because
resId == SYS4INI file_number, each DATA2 asset-start ReadFile tells us the EXACT value it holds
at that instant (the asset's file_number). So we self-drive a Cheat-Engine-style scan: on the
first CG load, scan the heap for int32 == fn; on each later load, keep only candidates that now
equal the new fn. The monotonic distinct opening sequence (35,37,39,43,46,122,129,...) collapses
the set to G[0x62424] in a few steps -- no native-copy contamination, no manual timing.
Once found, its address anchors the VM int-global store; from there we derive the address->memory
mapping and read any global live.
Flow: title screen -> attach -> start new game -> advance the opening SLOWLY (one CG at a time).
"""
import json
import sys
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
INDEX = REPO / "build" / "asset-index.json"
BG_MIN_SIZE = 500_000 # backgrounds are big AGFs (~1MB); portraits/sprites are far smaller
def data2_off2fn():
"""{offset: file_number} for BACKGROUND-sized DATA2 assets only (excludes sprite/portrait
loads, which churn G[0x62424]/other resId globals between background changes)."""
idx = json.loads(INDEX.read_text(encoding="utf-8"))
return {f["offset"]: f["file_number"] for f in idx["files"]
if f["archive"] == "DATA2.ALF" and f["size"] >= BG_MIN_SIZE}
JS_TEMPLATE = r"""
const OFF2FN = __OFF2FN__; // {offset: file_number == resId}
const k32 = Process.getModuleByName('kernel32.dll');
const GetFinalPathNameByHandleW = new NativeFunction(
k32.findExportByName('GetFinalPathNameByHandleW'), 'uint32', ['pointer','pointer','uint32','uint32']);
const SetFilePointer = new NativeFunction(
k32.findExportByName('SetFilePointer'), 'uint32', ['pointer','int32','pointer','uint32']);
const NUL = ptr(0); const cache = {};
function pathOf(h){ const k=h.toString(); let v=cache[k]; if(v!==undefined) return v; let p=null;
try{ const b=Memory.alloc(1040); const n=GetFinalPathNameByHandleW(h,b,519,0);
if(n>0&&n<519) p=b.readUtf16String(); }catch(e){} cache[k]=p; return p; }
function u32le(v){const b=[v&0xff,(v>>>8)&0xff,(v>>>16)&0xff,(v>>>24)&0xff];
return b.map(x=>('0'+x.toString(16)).slice(-2)).join(' ');}
const CAP = 600000;
function scanValue(v){
const out=[]; const pat=u32le(v);
const ranges=Process.enumerateRanges('rw-');
for(const r of ranges){
let m; try{ m=Memory.scanSync(r.base, r.size, pat); }catch(e){ continue; }
for(const x of m){ out.push(x.address); if(out.length>=CAP) return out; }
}
return out;
}
function isStack(a){ // heuristic: self-referential / return-addr neighbourhood
const r=Process.findRangeByAddress(a);
return r && r.size < 0x200000; // small rw- region = likely a stack
}
let cands=null, lastOff=-1, step=0;
function keepEq(v){ cands=cands.filter(a=>{ try{ return a.readU32()===(v>>>0); }catch(e){ return false; } }); }
Interceptor.attach(k32.findExportByName('ReadFile'), {
onEnter(args){
const p=pathOf(args[0]); if(!p || !/data2\.alf$/i.test(p)) return;
const size=args[2].toInt32(); if(size>4096) return;
const ov=args[4]; let off=-1;
try{ off = ov.isNull()? SetFilePointer(args[0],0,NUL,1) : ov.add(8).readU32(); }catch(e){}
if(!(off in OFF2FN)) return;
if(off===lastOff) return; lastOff=off;
const fn=OFF2FN[off];
if(cands===null){ cands=scanValue(fn); }
else { keepEq(fn); }
step++;
send({step:step, phase:'load', fn:fn, count:cands.length});
// STABILITY FILTER: 900ms later (during the pause before the next click) the global still
// holds fn, but transient stack copies have been overwritten -> drop them.
setTimeout(function(){
if(cands===null) return;
keepEq(fn);
send({step:step, phase:'stable', fn:fn, count:cands.length,
addrs: cands.length<=12 ? cands.map(a=>({a:a.toString(), stack:isStack(a)})) : []});
}, 1400);
}
});
send({ready:true});
"""
def main():
import frida
args = [a for a in sys.argv[1:] if not a.startswith("-")]
proc = args[0] if args else "AGE.EXE"
js = JS_TEMPLATE.replace("__OFF2FN__", json.dumps(data2_off2fn()))
def on_message(msg, data):
if msg.get("type") == "error":
print("[frida-error]", msg.get("description")); return
if msg.get("type") != "send":
return
pl = msg["payload"]
if pl.get("ready"):
print("[frida] scan hook live — advance the opening one CG at a time, pausing ~1.5s each.")
return
ph = pl.get("phase")
print(f"[step {pl['step']} {ph:>6}] resId={pl['fn']} (0x{pl['fn']:x}) candidates={pl['count']}")
if ph == "stable" and pl.get("addrs"):
for e in pl["addrs"]:
print(f" {e['a']} {'(stack)' if e['stack'] else '<== STABLE global candidate'}")
stable = [e for e in pl["addrs"] if not e["stack"]]
if 0 < len(stable) <= 3:
print(" >>> stable non-stack survivors — likely G[0x62424].")
dev = frida.get_local_device()
target = int(proc) if str(proc).isdigit() else proc
try:
session = frida.attach(target)
except frida.ProcessNotFoundError:
print("AGE-like:", [(p.pid, p.name) for p in dev.enumerate_processes() if "age" in p.name.lower()])
return 2
script = session.create_script(js)
script.on("message", on_message)
script.load()
print(f"[frida] attached to {proc}; narrowing G[0x62424] by the resId sequence.")
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,183 @@
#!/usr/bin/env python3
"""Locate the game's INT-GLOBAL array in memory by a known-value signature scan, so we can
read VM globals live (e.g. G[0x62424] = the CG resId) -- see docs/asset-resolution-re.md step 2.
The `*INIT` scripts write thousands of known constants to known global-int addresses at boot:
`mov (global-int ADDR) IMM`. Our VM addresses globals as a flat int array, so at runtime the
game holds `int_globals[ADDR]` at `B + ADDR*4` for some base B. We build a signature of those
(ADDR, value) pairs, find a long CONTIGUOUS run as a rare multi-dword anchor, `Memory.scan` for
it, and verify each candidate B against many scattered pairs. Unique high-match B = the array.
This unlocks runtime global observation generally (the roadmap's VM-validation cornerstone):
read `resId` (G[0x62424]) directly at each CG load, and watch scope-selector globals.
build: py -3.11 -X utf8 tools/frida/find_globals_base.py --build-sig
scan: py -3.11 -u -X utf8 tools/frida/find_globals_base.py <pid>
"""
import collections
import json
import sys
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
SIG = REPO / "build" / "globals-signature.json"
sys.path.insert(0, str(REPO / "tools"))
MOV, T_GINT, T_IMM = 0x55, 3, 0
INIT_SCRIPTS = ["EBINIT", "ITINIT", "SKINIT", "CGINIT"]
def build_signature():
import paths, sys4load
pairs = collections.defaultdict(collections.Counter)
for name in INIT_SCRIPTS:
p = paths.GAME_DIR / f"{name}.BIN"
if not p.exists():
p = paths.DATA1 / f"{name}.BIN"
if not p.exists():
continue
for ins in sys4load.load(p).instructions:
if (ins.opcode == MOV and len(ins.args) >= 2
and ins.args[0][0] == T_GINT and ins.args[1][0] == T_IMM):
pairs[ins.args[0][1]][ins.args[1][1]] += 1
# stable, distinctive, single-write addresses
single = {a: next(iter(vc)) for a, vc in pairs.items()
if len(vc) == 1 and vc.most_common(1)[0][1] == 1
and 8 < next(iter(vc)) < 0x7fffffff}
# longest contiguous run (addr, addr+1, ...) -> rare multi-dword anchor
addrs = sorted(single)
best = (None, 0)
i = 0
while i < len(addrs):
j = i
while j + 1 < len(addrs) and addrs[j + 1] == addrs[j] + 1:
j += 1
if j - i + 1 > best[1]:
best = (addrs[i], j - i + 1)
i = j + 1
anchor_addr, anchor_len = best
anchor = [(anchor_addr + k, single[anchor_addr + k]) for k in range(anchor_len)]
# scattered verification pairs spread across the address range
spread = addrs[::max(1, len(addrs) // 120)][:120]
verify = [(a, single[a]) for a in spread]
SIG.parent.mkdir(parents=True, exist_ok=True)
SIG.write_text(json.dumps({"anchor_addr": anchor_addr, "anchor": anchor, "verify": verify},
ensure_ascii=False), encoding="utf-8")
print(f"signature: {len(single)} distinctive pairs; "
f"anchor run @0x{anchor_addr:x} len {anchor_len} ({anchor_len*4} bytes); "
f"{len(verify)} verify pairs -> {SIG.relative_to(REPO)}")
print("anchor values:", [v for _, v in anchor[:12]])
return 0
JS_TEMPLATE = r"""
const ANCHOR_ADDR = __ANCHOR_ADDR__;
const ANCHOR_VALS = __ANCHOR_VALS__; // consecutive int32 values at ANCHOR_ADDR..
const VERIFY = __VERIFY__; // [[addr,val],...]
// build the anchor byte pattern (little-endian int32 each)
function u32le(v){ const b=[v&0xff,(v>>>8)&0xff,(v>>>16)&0xff,(v>>>24)&0xff];
return b.map(x=>('0'+x.toString(16)).slice(-2)).join(' '); }
const pattern = ANCHOR_VALS.map(u32le).join(' ');
function verifyBase(B){
let ok=0, tot=0;
for(const pv of VERIFY){
tot++;
try { if(B.add(pv[0]*4).readU32() === (pv[1]>>>0)) ok++; } catch(e){}
}
return {ok:ok, tot:tot};
}
const ranges = Process.enumerateRanges('rw-').filter(r=>r.size >= 1024*1024);
let best=null;
// robustness ladder: long anchor is rare but fragile to any changed value; short prefixes
// catch it if a value moved. Each candidate is confirmed by the 120 scattered verify pairs.
const LENS = [ANCHOR_VALS.length, 64, 16, 4].filter((v,i,a)=>v<=ANCHOR_VALS.length && a.indexOf(v)===i);
for(const L of LENS){
const pat = ANCHOR_VALS.slice(0,L).map(u32le).join(' ');
let hits=0;
ranges.forEach(function(r){
let matches; try { matches = Memory.scanSync(r.base, r.size, pat); } catch(e){ return; }
if(matches.length > 8000) return; // too common at this length, skip
hits += matches.length;
matches.forEach(function(m){
const B = m.address.sub(ANCHOR_ADDR*4);
const v = verifyBase(B);
if(v.ok >= 30 && (!best || v.ok>best.ok))
best = {base: B.toString(), ok:v.ok, tot:v.tot, anchor_at: m.address.toString(), anchor_len:L};
});
});
send({phase:'scan', anchor_len:L, hits:hits, found: !!best});
if(best) break;
}
if(best){
// read G[0x62424] (the CG resId) as a sanity value
let resid=null; try{ resid = ptr(best.base).add(0x62424*4).readU32(); }catch(e){}
best.G_62424 = resid;
send({phase:'found', best:best});
} else {
send({phase:'notfound'});
}
"""
def scan(pid):
import frida
if not SIG.exists():
raise SystemExit("no signature; run --build-sig first")
sig = json.loads(SIG.read_text(encoding="utf-8"))
js = (JS_TEMPLATE
.replace("__ANCHOR_ADDR__", str(sig["anchor_addr"]))
.replace("__ANCHOR_VALS__", json.dumps([v for _, v in sig["anchor"]]))
.replace("__VERIFY__", json.dumps(sig["verify"])))
out = {}
def on_message(msg, data):
if msg.get("type") == "error":
print("[frida-error]", msg.get("description")); return
if msg.get("type") != "send":
return
pl = msg["payload"]
ph = pl.get("phase")
if ph == "scan":
print(f"[scan] anchor prefix {pl['anchor_len']} dwords: {pl['hits']} raw hits"
+ (" -> base confirmed" if pl.get("found") else ""))
elif ph == "found":
b = pl["best"]
out["base"] = b["base"]
print(f"[FOUND] int-global base = {b['base']} (verify {b['ok']}/{b['tot']} pairs, "
f"anchor @ {b['anchor_at']})")
print(f" G[0x62424] (CG resId right now) = {b['G_62424']} (0x{b['G_62424']:x})"
if b.get("G_62424") is not None else " G[0x62424] unreadable")
elif ph == "notfound":
print("[scan] anchor pattern not found in any heap range (globals not resident, "
"wrong element size, or array not yet populated)")
dev = frida.get_local_device()
target = int(pid) if str(pid).isdigit() else pid
try:
session = frida.attach(target)
except frida.ProcessNotFoundError:
print("AGE-like:", [(p.pid, p.name) for p in dev.enumerate_processes() if "age" in p.name.lower()])
return 2
script = session.create_script(js)
script.on("message", on_message)
script.load()
if "base" in out:
print(f"\nint-global base found: {out['base']}. Next: hook the loader and read "
f"G[0x62424] there for definitive (resId,name) pairs.")
return 0
def main():
if "--build-sig" in sys.argv:
return build_signature()
pid = next((a for a in sys.argv[1:] if not a.startswith("-")), "AGE.EXE")
return scan(pid)
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,178 @@
#!/usr/bin/env python3
"""Phase 1 of the native resId->filename crack: LOCATE the resource-load / set-texture
handler in the running game by back-tracing every asset-open (see docs/asset-resolution-re.md
step 2, option a).
Idea: to issue a ReadFile at an asset's exact archive offset the game must have *already*
resolved resId -> (archive, offset) inside its native load-by-id / set-texture (op 0x1f9)
handler. So at each asset-start read, a stack backtrace passes straight through that handler.
We backtrace only on reads whose offset EXACTLY equals a DATA2 asset offset (from
build/asset-index.json), resolve the offset -> asset name, and log module-relative frames.
Frames are AGE.EXE-relative (base subtracted) so they're stable across runs despite ASLR.
Offline, `aggregate` ranks the AGE.EXE return addresses that recur across the MOST distinct
assets: the load-by-id / set-texture handler is the frame common to every asset-open. That
address (module+offset) becomes the hook target for phase 2 (read the resId argument ->
definitive resId->name table). No game state is modified.
Flow (see tools/frida/README.md):
1. Launch the game (via `AGE Patch.exe`) to the title.
2. py -3.11 -u -X utf8 tools/frida/locate_resource_load.py
3. Start a new game so the SC0000 opening auto-plays; let a dozen CGs load.
4. Ctrl-C. Then: py -3.11 -X utf8 tools/frida/locate_resource_load.py --aggregate
Output: build/frida-resource-bt.jsonl ({name, offset, size, frames:[...]} per asset-open).
"""
import json
import sys
import time
from pathlib import Path
REPO = Path(__file__).resolve().parents[2] # age-reimpl/
OUT = REPO / "build" / "frida-resource-bt.jsonl"
INDEX = REPO / "build" / "asset-index.json"
def load_data2_offsets():
"""{offset -> name} for DATA2.ALF from the asset index (asset-start detector + resolver)."""
idx = json.loads(INDEX.read_text(encoding="utf-8"))
return {f["offset"]: f["name"] for f in idx["files"] if f["archive"] == "DATA2.ALF"}
def aggregate():
"""Rank AGE.EXE-relative frames by how many DISTINCT assets they appear under."""
if not OUT.exists():
raise SystemExit(f"no capture log: {OUT} (run the capture first)")
from collections import defaultdict
assets_per_frame = defaultdict(set) # frame-string -> set(asset names)
depth_of_frame = defaultdict(list) # frame-string -> [stack depths]
n = 0
for ln in OUT.read_text(encoding="utf-8").splitlines():
if not ln.strip():
continue
rec = json.loads(ln)
n += 1
for depth, fr in enumerate(rec.get("frames", [])):
if fr.startswith("AGE.EXE+"): # ignore kernel32/ntdll/CRT frames
assets_per_frame[fr].add(rec["name"])
depth_of_frame[fr].append(depth)
distinct_assets = {rec["name"] for rec in
(json.loads(l) for l in OUT.read_text(encoding="utf-8").splitlines() if l.strip())}
print(f"{n} asset-open events over {len(distinct_assets)} distinct assets")
print("AGE.EXE frames ranked by distinct-asset coverage (handler = covers ~all):")
ranked = sorted(assets_per_frame.items(), key=lambda kv: (-len(kv[1]), fr_depth(depth_of_frame[kv[0]])))
for fr, assets in ranked[:25]:
d = depth_of_frame[fr]
print(f" {fr:<22} assets={len(assets):<3} avg_depth={sum(d)/len(d):.1f}")
print("\nThe handler is the frame covering the most distinct assets at a shallow, stable depth.")
print("Hook it in phase 2 to read the resId argument.")
return 0
def fr_depth(depths):
return sum(depths) / len(depths)
JS_TEMPLATE = r"""
const OFFSETS = new Set(__OFFSETS__); // exact DATA2 asset-start offsets
const k32 = Process.getModuleByName('kernel32.dll');
const GetFinalPathNameByHandleW = new NativeFunction(
k32.findExportByName('GetFinalPathNameByHandleW'), 'uint32', ['pointer','pointer','uint32','uint32']);
const SetFilePointer = new NativeFunction(
k32.findExportByName('SetFilePointer'), 'uint32', ['pointer','int32','pointer','uint32']);
const NUL = ptr(0);
const cache = {};
function pathOf(h) {
const key = h.toString();
let v = cache[key]; if (v !== undefined) return v;
let p = null;
try { const buf = Memory.alloc(1040);
const n = GetFinalPathNameByHandleW(h, buf, 519, 0);
if (n > 0 && n < 519) p = buf.readUtf16String(); } catch (e) {}
cache[key] = p; return p;
}
function frameStr(addr) {
const m = Process.findModuleByAddress(addr);
if (!m) return addr.toString();
return m.name + '+0x' + addr.sub(m.base).toString(16);
}
const rf = k32.findExportByName('ReadFile');
Interceptor.attach(rf, {
onEnter(args) {
const p = pathOf(args[0]);
if (!p || !/data2\.alf$/i.test(p)) return;
const size = args[2].toInt32();
if (size > 4096) return; // header reads only (asset-start burst)
const ov = args[4];
let off = -1;
try { off = ov.isNull() ? SetFilePointer(args[0], 0, NUL, 1) : ov.add(8).readU32(); } catch (e) {}
if (!OFFSETS.has(off)) return; // only exact asset-start offsets
let frames = [];
try {
frames = Thread.backtrace(this.context, Backtracer.ACCURATE).map(frameStr);
} catch (e) {
try { frames = Thread.backtrace(this.context, Backtracer.FUZZY).map(frameStr); } catch (e2) {}
}
send({offset: off, size: size, frames: frames});
}
});
send({ready: true});
"""
def capture(proc):
import frida
offsets = load_data2_offsets()
js = JS_TEMPLATE.replace("__OFFSETS__", json.dumps(sorted(offsets.keys())))
OUT.parent.mkdir(parents=True, exist_ok=True)
log = open(OUT, "w", encoding="utf-8")
seen = set()
def on_message(msg, data):
if msg.get("type") == "error":
print("[frida-error]", msg.get("description")); return
if msg.get("type") != "send":
return
pl = msg["payload"]
if pl.get("ready"):
print("[frida] backtrace hook live — start a new game; let the opening load CGs.")
return
name = offsets.get(pl["offset"], "?")
rec = {"name": name, "offset": pl["offset"], "size": pl["size"], "frames": pl["frames"]}
log.write(json.dumps(rec, ensure_ascii=False) + "\n"); log.flush()
if name not in seen:
seen.add(name)
age = [f for f in pl["frames"] if f.startswith("AGE.EXE+")]
print(f"OPEN {name} ({len(pl['frames'])} frames, {len(age)} in AGE.EXE)")
for f in pl["frames"][:8]:
print(" " + f)
target = int(proc) if proc.isdigit() else proc
try:
session = frida.attach(target)
except frida.ProcessNotFoundError:
procs = frida.get_local_device().enumerate_processes() # frida 17.x: device method
print(f"process '{proc}' not found. AGE-like:",
[(p.pid, p.name) for p in procs if "age" in p.name.lower()])
return 2
script = session.create_script(js)
script.on("message", on_message)
script.load()
print(f"[frida] attached to {proc}; {len(offsets)} DATA2 asset offsets loaded; log -> {OUT}")
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
print("\n[frida] stopped. Now: py -3.11 -X utf8 tools/frida/locate_resource_load.py --aggregate")
return 0
def main():
if "--aggregate" in sys.argv:
return aggregate()
proc = next((a for a in sys.argv[1:] if not a.startswith("-")), "AGE.EXE")
return capture(proc)
if __name__ == "__main__":
sys.exit(main())

225
tools/parse_sys4ini.py Normal file
View File

@@ -0,0 +1,225 @@
#!/usr/bin/env python3
"""Parse SYS4INI.BIN (Eushully AGE asset index) -> build/asset-index.json.
SYS4INI.BIN is the authoritative directory the game (and BinExtractALF, which is
based on asmodean's exs4alf) uses to locate every asset inside the DATA*.ALF
archives. It maps name <-> archive <-> offset <-> size and is the reusable
"answer key" for resId->file resolution (see docs/asset-resolution-re.md): it
both names every asset and gives archive-offset->name (to rescue noisy Frida
file-I/O offsets).
Container format (little-endian x86), signature "S4IC422 " at offset 0:
0x000 char signature[?] "S4IC" family; "S4IC" -> data at 0x134
... (title / padding)
0x134 uint32 packed_size length of the LZSS stream that follows
0x138 byte[] lzss_stream packed_size bytes, runs to EOF
The LZSS stream (GARbro-style: 0x1000 ring buffer, zero-filled, init pos 0xFEE,
control bits LSB->MSB, 1=literal, 0=two-byte back-reference: offset=(hi&0xf0)<<4
| lo, length=3+(hi&0xf)) decompresses to a plain directory:
uint32 arc_count
{ char name[256] } x arc_count archive filenames (DATA1.ALF ..)
uint32 file_count
{ char name[64]; uint32 arc_id; one record per asset
uint32 file_number; uint32 offset;
uint32 size } x file_count (80 bytes each)
Entries whose name is "@" are placeholders and skipped (matches GARbro/exs4alf).
Usage: py -3.11 -X utf8 tools/parse_sys4ini.py [--check]
--check cross-validate against the extracted/ ground truth and .ALF sizes
"""
from __future__ import annotations
import json
import struct
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import paths
# S4IC family: signature -> offset of the packed-size dword that precedes the stream.
DATA_OFFSETS = {b"S4AC": 0x114, b"S4IC": 0x134, b"S3IC": 0x134, b"S3IN": 0x12C}
FRAME_SIZE = 0x1000
FRAME_INIT_POS = 0xFEE
ARC_NAME_LEN = 256
FILE_NAME_LEN = 64
FILE_ENTRY_FMT = "<64s4I" # name[64], arc_id, file_number, offset, size
FILE_ENTRY_LEN = struct.calcsize(FILE_ENTRY_FMT) # 80
def lzss_decompress(src: bytes) -> bytes:
"""GARbro-compatible LZSS (0x1000 ring buffer, init pos 0xFEE, threshold 3)."""
frame = bytearray(FRAME_SIZE) # zero-filled
fpos = FRAME_INIT_POS
out = bytearray()
i, n = 0, len(src)
while i < n:
ctl = src[i]; i += 1
for bit in (1, 2, 4, 8, 0x10, 0x20, 0x40, 0x80):
if ctl & bit: # literal
if i >= n:
return bytes(out)
b = src[i]; i += 1
out.append(b)
frame[fpos] = b; fpos = (fpos + 1) & 0xFFF
else: # back-reference
if i + 1 >= n:
return bytes(out)
lo, hi = src[i], src[i + 1]; i += 2
off = ((hi & 0xF0) << 4) | lo
for _ in range(3 + (hi & 0x0F)):
b = frame[off & 0xFFF]; off += 1
out.append(b)
frame[fpos] = b; fpos = (fpos + 1) & 0xFFF
return bytes(out)
def _cstr(buf: bytes) -> str:
"""Decode a null-terminated cp932 field (filenames are ASCII in practice)."""
return buf.split(b"\x00", 1)[0].decode("cp932", errors="replace")
def parse(path: Path) -> dict:
raw = path.read_bytes()
sig4 = raw[:4]
if sig4 not in DATA_OFFSETS:
raise SystemExit(f"{path.name}: unknown signature {raw[:8]!r}")
magic = _cstr(raw[:8])
doff = DATA_OFFSETS[sig4]
packed_size = struct.unpack_from("<I", raw, doff)[0]
stream = raw[doff + 4: doff + 4 + packed_size]
if len(stream) != packed_size:
raise SystemExit(f"{path.name}: packed stream truncated "
f"({len(stream)} of {packed_size} bytes)")
blob = lzss_decompress(stream)
p = 0
(arc_count,) = struct.unpack_from("<I", blob, p); p += 4
if not 0 < arc_count < 0x1000:
raise SystemExit(f"{path.name}: implausible arc_count {arc_count}")
archives = []
for _ in range(arc_count):
archives.append(_cstr(blob[p:p + ARC_NAME_LEN])); p += ARC_NAME_LEN
(file_count,) = struct.unpack_from("<I", blob, p); p += 4
if not 0 < file_count < 0x400000:
raise SystemExit(f"{path.name}: implausible file_count {file_count}")
need = p + file_count * FILE_ENTRY_LEN
if need > len(blob):
raise SystemExit(f"{path.name}: directory truncated (need {need}, "
f"have {len(blob)} decompressed bytes)")
files = []
for _ in range(file_count):
name_b, arc_id, file_number, offset, size = struct.unpack_from(FILE_ENTRY_FMT, blob, p)
p += FILE_ENTRY_LEN
name = _cstr(name_b)
if name == "@" or not name:
continue
files.append({
"name": name,
"archive": archives[arc_id] if 0 <= arc_id < arc_count else None,
"arc_id": arc_id,
"file_number": file_number,
"offset": offset,
"size": size,
})
return {
"source": path.name,
"magic": magic,
"packed_size": packed_size,
"decompressed_size": len(blob),
"archive_count": arc_count,
"archives": archives,
"file_count": file_count,
"entry_count": len(files),
"files": files,
}
def check(index: dict) -> int:
"""Cross-validate against extracted/ counts and real .ALF file sizes."""
problems = 0
per_arc: dict[str, int] = {}
for f in index["files"]:
per_arc[f["archive"]] = per_arc.get(f["archive"], 0) + 1
print("per-archive entry counts (from SYS4INI):")
for a in index["archives"]:
print(f" {a:<16} {per_arc.get(a, 0)}")
# 1) offset+size must fit inside the real archive on disk.
for a in index["archives"]:
alf = paths.GAME_DIR / a
if not alf.exists():
print(f" ! archive not on disk: {a}")
continue
asize = alf.stat().st_size
over = [f for f in index["files"]
if f["archive"] == a and f["offset"] + f["size"] > asize]
if over:
problems += len(over)
print(f" ! {a}: {len(over)} entries run past EOF ({asize} bytes); "
f"e.g. {over[0]['name']} @ {over[0]['offset']}+{over[0]['size']}")
else:
print(f" ok {a}: all entries within {asize} bytes")
# 2) spot-check extracted-folder file sizes against the index (name -> size).
by_name = {f["name"].upper(): f for f in index["files"]}
checked = mismatch = 0
for d in ("DATA1", "DATA2", "DATA3", "DATA4", "DATA5"):
folder = paths.EXTRACTED / d
if not folder.is_dir():
continue
for fp in list(folder.glob("*"))[:200]:
if not fp.is_file():
continue
e = by_name.get(fp.name.upper())
if e is None:
continue
checked += 1
if e["size"] != fp.stat().st_size:
mismatch += 1
if mismatch <= 5:
print(f" ! size mismatch {fp.name}: index {e['size']} "
f"vs extracted {fp.stat().st_size}")
print(f"size spot-check: {checked} matched by name, {mismatch} size mismatches")
problems += mismatch
print("CHECK OK" if problems == 0 else f"CHECK: {problems} problems")
return problems
def main() -> int:
do_check = "--check" in sys.argv[1:]
src = paths.GAME_DIR / "SYS4INI.BIN"
if not src.exists():
raise SystemExit(f"not found: {src}")
index = parse(src)
print(f"{index['source']}: magic={index['magic']!r} "
f"packed={index['packed_size']} -> {index['decompressed_size']} bytes")
print(f"archives ({index['archive_count']}): {index['archives']}")
print(f"files: {index['file_count']} declared, {index['entry_count']} real "
f"(after skipping '@' placeholders)")
for f in index["files"][:5]:
print(f" {f['name']:<16} {f['archive']:<12} "
f"off={f['offset']:>12} size={f['size']:>10} #{f['file_number']}")
out = paths.BUILD / "asset-index.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(index, ensure_ascii=False, indent=1), encoding="utf-8")
print(f"-> {out.relative_to(paths.REPO)}")
if do_check:
print("--- validation ---")
return 1 if check(index) else 0
return 0
if __name__ == "__main__":
sys.exit(main())

116
tools/resolve_asset.py Normal file
View File

@@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""Static, general asset resolver: (scene, resId) -> asset file. Solves asset resolution
(docs/asset-resolution-re.md) with NO runtime capture.
Mechanism (proven): SYS4INI's file list is organized into SECTIONS, one per scene -- each is a
`SCxxxx.BIN` script entry followed by that scene's asset MANIFEST (all assets it references, across
every archive and type: EV/BG/CS/AE event & sprite graphics, OGG/WAV audio, ...). `file_number` is
the 0-based index within the section. So a bytecode resId resolves as:
resId -> files[ section_base(scene) + resId ]
where section_base(scene) is the start of the SYS4INI section containing the scene's script.
This is the same rule for set-texture(resId), play-bgm(id), play-voice(id) -- one unified manifest.
Validated: 97% of files fit `fn == position - section_base`; SC0000's opening resolves 17/17 vs
Frida ground truth; 586/595 captured loads across all sections satisfy `files[base+fn] == name`.
Usage:
py -3.11 -X utf8 tools/resolve_asset.py --build # emit build/asset-sections.json
py -3.11 -X utf8 tools/resolve_asset.py <SCENE> [resId] # resolve one, or dump the manifest
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import paths
def load_index():
return json.loads((paths.BUILD / "asset-index.json").read_text(encoding="utf-8"))["files"]
def sections(files):
"""Split the SYS4INI file list into sections at each file_number reset (fn <= prev).
Returns (section_start_per_position[list], sections[list of (start, end, scene_name|None])."""
base_of, secs, start, prev = [], [], 0, -1
for i, f in enumerate(files):
if f["file_number"] <= prev:
secs.append((start, i - 1))
start = i
base_of.append(start)
prev = f["file_number"]
secs.append((start, len(files) - 1))
# attach the scene script (SCxxxx.BIN) that owns each section, if any
out = []
for s, e in secs:
scene = next((files[k]["name"] for k in range(s, e + 1)
if re.match(r"SC\d+\.BIN$", files[k]["name"])), None)
out.append({"start": s, "end": e, "scene": scene})
return base_of, out
def scene_base(files, base_of, scene):
key = scene.upper()
if not key.endswith(".BIN"):
key += ".BIN"
pos = next((i for i, f in enumerate(files) if f["name"].upper() == key), None)
if pos is None:
raise SystemExit(f"scene not in SYS4INI: {scene}")
return base_of[pos]
def resolve(files, base, resid):
p = base + resid
return files[p] if 0 <= p < len(files) else None
def main() -> int:
files = load_index()
base_of, secs = sections(files)
if "--build" in sys.argv:
scene_bases = {}
for sec in secs:
if sec["scene"]:
scene_bases[sec["scene"].removesuffix(".BIN")] = sec["start"]
out = paths.BUILD / "asset-sections.json"
out.write_text(json.dumps(
{"note": "SYS4INI sections; resolve resId -> files[section_base + resId]. "
"See docs/asset-resolution-re.md.",
"section_count": len(secs),
"scene_base": scene_bases,
"sections": secs}, ensure_ascii=False, indent=1), encoding="utf-8")
print(f"{len(secs)} sections, {len(scene_bases)} scenes -> {out.relative_to(paths.REPO)}")
return 0
args = [a for a in sys.argv[1:] if not a.startswith("-")]
if not args:
raise SystemExit(__doc__)
scene = args[0]
base = scene_base(files, base_of, scene)
sec = next(s for s in secs if s["start"] == base)
print(f"{scene}: section [{sec['start']}..{sec['end']}] base {base} "
f"({sec['end']-sec['start']+1} entries)")
if len(args) > 1:
resid = int(args[1], 0)
f = resolve(files, base, resid)
print(f" resId {resid} -> {f['archive']} {f['name']} (offset {f['offset']}, size {f['size']})"
if f else f" resId {resid} -> out of range")
return 0
# dump the scene's manifest (graphics + audio), skipping the leading script entry
print(" manifest (resId -> asset):")
for p in range(base, sec["end"] + 1):
resid = p - base
f = files[p]
print(f" {resid:>4} {f['archive'][:5]} {f['name']}")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,119 @@
#!/usr/bin/env python3
"""Resolve Frida archive-read offsets -> asset names via build/asset-index.json.
The runtime capture (tools/frida/capture_graphics.py) logs raw ReadFile spans on the
DATA*.ALF archives as `<path>\t<offset>\t<size>` lines. On their own those offsets are
opaque and mixed with OS memory-map paging (the doc's "Frida file-I/O is noisy"). With
the SYS4INI asset index as the answer key we can turn each offset back into the *asset*
it belongs to, and thereby recover the real per-scene **asset load order** — the ground
truth for cracking resId->filename (see docs/asset-resolution-re.md, step 2).
Two read signals per asset (observed): a burst of tiny header reads starting exactly at
the asset's archive offset (delta 0), then one bulk read of the payload. Uniform 0x20000
(131072-byte) reads are memory-map paging and are dropped. We treat a read whose offset
*exactly* equals an index entry's offset as an unambiguous "asset-start" event; the
ordered, de-duplicated sequence of those is the load order.
Usage: py -3.11 -X utf8 tools/resolve_frida_reads.py [reads.log] [-o out.json]
default reads.log = build/frida-reads.log ; default out = build/frida-asset-loads.json
"""
from __future__ import annotations
import bisect
import json
import sys
from pathlib import Path
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import paths
PAGING_SIZE = 131072 # 0x20000 uniform memory-map paging reads -> noise
def load_index() -> dict:
idx = json.loads((paths.BUILD / "asset-index.json").read_text(encoding="utf-8"))
by_arc: dict[str, list[tuple[int, int, str]]] = {}
for f in idx["files"]:
by_arc.setdefault(f["archive"], []).append((f["offset"], f["size"], f["name"]))
for a in by_arc:
by_arc[a].sort()
return by_arc
def resolve(by_arc, arc, off):
"""Return (name, entry_offset, size, delta) for the asset whose range holds `off`."""
arr = by_arc.get(arc)
if not arr:
return None
i = bisect.bisect_right(arr, (off, float("inf"), "")) - 1
if i < 0:
return None
o, s, n = arr[i]
return (n, o, s, off - o) if off < o + s else None
def main() -> int:
args = [a for a in sys.argv[1:] if not a.startswith("-")]
out_flag = next((sys.argv[i + 1] for i, a in enumerate(sys.argv) if a == "-o"), None)
log = Path(args[0]) if args else paths.BUILD / "frida-reads.log"
out = Path(out_flag) if out_flag else paths.BUILD / "frida-asset-loads.json"
if not log.exists():
raise SystemExit(f"reads log not found: {log}")
by_arc = load_index()
# exact-offset -> name per archive (asset-start detector)
exact = {a: {o: n for o, _, n in v} for a, v in by_arc.items()}
per_arc: dict[str, int] = {}
paging = unresolved = total = 0
starts = [] # ordered (arc, name) asset-start events (deduped consecutively)
contained = set() # every distinct asset any read touched
for ln in log.read_text(encoding="utf-8").splitlines():
parts = ln.split("\t")
if len(parts) != 3:
continue
path, off_s, size_s = parts
arc = path.replace("\\", "/").rsplit("/", 1)[-1]
off, size = int(off_s), int(size_s)
total += 1
per_arc[arc] = per_arc.get(arc, 0) + 1
if size == PAGING_SIZE:
paging += 1
continue
hit = resolve(by_arc, arc, off)
if hit is None:
unresolved += 1
continue
name = hit[0]
contained.add((arc, name))
if off in exact.get(arc, {}): # exact asset-start
ev = (arc, exact[arc][off])
if not starts or starts[-1] != ev:
starts.append(ev)
result = {
"source_log": log.name,
"asset_index": "asset-index.json",
"total_reads": total,
"paging_reads_dropped": paging,
"unresolved_reads": unresolved,
"reads_per_archive": per_arc,
"distinct_assets_touched": len(contained),
"load_order_count": len(starts),
"load_order": [{"archive": a, "name": n} for a, n in starts],
}
out.write_text(json.dumps(result, ensure_ascii=False, indent=1), encoding="utf-8")
print(f"{log.name}: {total} reads ({paging} paging dropped, {unresolved} unresolved)")
print(f"reads/archive: {per_arc}")
print(f"distinct assets touched: {len(contained)}; "
f"asset-start load order: {len(starts)} events")
for a, n in starts:
print(f" {a:<12} {n}")
print(f"-> {out.relative_to(paths.REPO)}")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -57,9 +57,11 @@ class Frame:
class VM:
def __init__(self, scr: sys4load.Sys4Script, verbose=False, emit_cap=EMIT_CAP):
def __init__(self, scr: sys4load.Sys4Script, verbose=False, emit_cap=EMIT_CAP, record_trace=False):
self.scr = scr
self.verbose = verbose
self.record_trace = record_trace
self.trace = [] # executed code offsets (only if record_trace)
self.code = scr.instructions
self.by_off = {ins.offset: idx for idx, ins in enumerate(self.code)}
self.G = collections.defaultdict(int) # global-int bank (flat address space)
@@ -67,6 +69,7 @@ class VM:
self.fr = Frame()
self.callstack = [] # return indices for call/ret
self.text = [] # captured show-text as (str_offset, text)
self.settex = [] # set-texture calls: (code_offset, resId, slot)
self.emit_seen = collections.Counter() # per-offset emit count (loop-guard)
self.emit_cap = emit_cap
self.halt_reason = None # 'exit' | 'LOOP:...' | 'STEP-LIMIT' | 'ret-underflow'
@@ -135,6 +138,8 @@ class VM:
ins = self.code[pc]
op = ins.opcode
self.exec_count[op] += 1
if self.record_trace:
self.trace.append(ins.offset)
nxt = self.step(ins, pc)
if nxt is None: # halt
break
@@ -214,6 +219,12 @@ class VM:
"display-furigana", "dev_ukn"):
return pc + 1
if lbl == "set-texture": # 0x1f9 (resId, slot, flag) — trace the load
resid = self.read(a[0]) if a else None
slot = self.read(a[1]) if len(a) > 1 else None
self.settex.append((ins.offset, resid, slot, len(self.trace))) # +trace index
return pc + 1
if op in MARKERS: # classified no-op markers
return pc + 1
@@ -426,6 +437,34 @@ def run_trace(out_path):
return 0
def run_settex(name):
"""Execute a scene and dump its set-texture(resId) trace in execution order.
This is the VM side of the asset-resolution scope-selector correlation
(docs/asset-resolution-re.md): each entry is {i, off, resId, slot}, and aligning this
ordered resId sequence with the game's Frida load order pins every load to a bytecode
offset -> localizes where the active CG package/scope switches.
"""
scripts = paths.scripts()
key = name.upper() if name.upper().endswith(".BIN") else name.upper() + ".BIN"
if key not in scripts:
raise SystemExit(f"scene not found: {name}")
vm = VM(sys4load.load(scripts[key]), record_trace=True)
vm.run()
out = paths.BUILD / f"settex-{key.removesuffix('.BIN')}.json"
rows = [{"i": i, "off": f"0x{off:x}", "resId": rid, "slot": slot, "trace_i": ti}
for i, (off, rid, slot, ti) in enumerate(vm.settex)]
out.write_text(json.dumps({"scene": key, "halt": vm.halt_reason, "steps": vm.steps,
"count": len(rows), "settex": rows,
"trace": [f"0x{o:x}" for o in vm.trace]},
ensure_ascii=False), encoding="utf-8")
print(f"{key}: {len(rows)} set-texture calls (halt={vm.halt_reason}, steps={vm.steps}) "
f"-> {out.relative_to(paths.REPO)}")
for r in rows[:20]:
print(f" #{r['i']:<3} {r['off']:>8} resId={r['resId']} (0x{r['resId']:x}) slot={r['slot']}")
return 0
def main(argv=None):
argv = argv if argv is not None else sys.argv[1:]
if not argv or argv[0] == "--test":
@@ -434,6 +473,8 @@ def main(argv=None):
return run_sweep(limit=int(argv[1]) if len(argv) > 1 else None)
if argv[0] == "--scene":
return run_one_scene(argv[1])
if argv[0] == "--settex":
return run_settex(argv[1])
if argv[0] == "--trace":
return run_trace(argv[1])
return run_file(argv[0])