conco/OpenMaidEngine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

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

Browse Source 
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 30d2734df4
commit 102fe1d021
18 changed files with 1887 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
tools/frida/README.md
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.

168
tools/frida/capture_load_order.py Normal file
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())

194
tools/frida/capture_resid_args.py Normal file
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())

141
tools/frida/find_global_by_sequence.py Normal file
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())

183
tools/frida/find_globals_base.py Normal file
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())

178
tools/frida/locate_resource_load.py Normal file
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())