re(frida): engine op-path tracer (tick/operand hook, per-script offsets)
Reads cur_ctx_index/frame_pc/frame_codebase from the engine ctx per executed op and emits (codebase, offset=(pc-codebase)/4) to build/engine-optrace.jsonl. Recon result: the tick hook (0x410fb0) does NOT expose ctx via ecx (0 entries); the operand hook (0x41b940) is the working capture (100% of offsets land on valid SC0000 instruction starts). Writes a tracer-live.flag so the capture can be gated on hooks-installed before the scene loads (else the entry burst is missed). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gamer147
165
tools/frida/trace_engine_ops.py
Normal file
165
tools/frida/trace_engine_ops.py
Normal file
@@ -0,0 +1,165 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Engine op-path tracer for the differential offset-path oracle (docs/engine-re.md).
|
||||
|
||||
Attaches to the running game and, per executed op, reads the current coroutine's PC + codebase from the
|
||||
engine ctx and emits (codebase, offset=(pc-codebase)/4) to build/engine-optrace.jsonl in execution order.
|
||||
Diffed against the VM's trace (Age.Cli trace --trace-json) by tools/diff_optrace.py — the first divergence
|
||||
is the opcode/branch we modeled wrong.
|
||||
|
||||
READ-ONLY (Interceptor + memory reads; no patching), following probe_frame_cadence.py / capture_gfx_objects.py.
|
||||
Two selectable hooks (Task 1 recon gate decides which is safe):
|
||||
--hook tick -> adv_interpreter_tick 0x410fb0 (one op per tick; clean 1:1 signal). CModule on this
|
||||
path crashed the game before; PLAIN-JS here is untested -> the recon gate proves it.
|
||||
--hook operand -> vm_operand_fetch 0x41b940 (PROVEN-safe, probe_frame_cadence.py uses it). Fires
|
||||
per-operand, so we dedupe consecutive same-pc; misses zero-operand ops (markers/no-ops,
|
||||
which don't branch) -> acceptable for a control-flow diff.
|
||||
|
||||
py -3.11 -u -X utf8 tools/frida/trace_engine_ops.py [--hook tick|operand] [seconds] [proc]
|
||||
|
||||
Let the deterministic opening auto-advance to the first wait-for-input, then stop (Ctrl-C or the timeout).
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parents[2]
|
||||
OUT = REPO / "build" / "engine-optrace.jsonl"
|
||||
LIVE = REPO / "build" / "tracer-live.flag" # touched once hooks are installed; removed on exit
|
||||
|
||||
TICK_OFF = 0x10fb0 # adv_interpreter_tick (0x410fb0); one op per tick, thiscall ecx=ctx
|
||||
OPFETCH_OFF = 0x1b940 # vm_operand_fetch (0x41b940); per-operand, thiscall ecx=ctx (PROVEN-safe)
|
||||
IDX_OFF = 0x53d14 # cur_ctx_index
|
||||
PC_BASE = 0x53d2c # frame_pc ; +idx*0x78 = current instruction pointer
|
||||
CB_BASE = 0x53d28 # frame_codebase; +idx*0x78 = base pointer of the running script's code
|
||||
PC_STRIDE = 0x78
|
||||
|
||||
JS = r"""
|
||||
const HOOK_OFF=%d, IDX_OFF=%d, PC_BASE=%d, CB_BASE=%d, PC_STRIDE=%d, DEDUPE=%d;
|
||||
const mod = Process.getModuleByName('AGE.EXE');
|
||||
let buf = [], total = 0, lastPc = -1;
|
||||
const FLUSH_N = 2000;
|
||||
|
||||
function flush(){ if (buf.length){ send({kind:'batch', rows: buf}); buf = []; } }
|
||||
|
||||
Interceptor.attach(mod.base.add(HOOK_OFF), {
|
||||
onEnter(args){
|
||||
const ctx = this.context.ecx; // thiscall: ecx = engine ctx
|
||||
let idx, pc, cb;
|
||||
try {
|
||||
idx = ctx.add(IDX_OFF).readS32();
|
||||
if (idx < 0 || idx >= 64) return;
|
||||
pc = ctx.add(PC_BASE + idx*PC_STRIDE).readU32();
|
||||
cb = ctx.add(CB_BASE + idx*PC_STRIDE).readU32();
|
||||
} catch(e){ return; }
|
||||
if (DEDUPE && pc === lastPc) return; // operand mode: one entry per instruction
|
||||
lastPc = pc;
|
||||
const off = (pc - cb) >>> 2; // byte delta / 4 = bytecode word index
|
||||
buf.push([cb >>> 0, off]);
|
||||
total++;
|
||||
if (buf.length >= FLUSH_N) flush();
|
||||
}
|
||||
});
|
||||
|
||||
setInterval(flush, 200); // flush the tail so nothing is lost on detach
|
||||
send({kind:'ready', base: mod.base.toString(), hook: '0x'+HOOK_OFF.toString(16), total: 0});
|
||||
rpc.exports = { stats(){ return total; }, flush(){ flush(); return total; } };
|
||||
"""
|
||||
|
||||
|
||||
def capture(hook, seconds, proc):
|
||||
import frida
|
||||
OUT.parent.mkdir(parents=True, exist_ok=True)
|
||||
hook_off = TICK_OFF if hook == "tick" else OPFETCH_OFF
|
||||
dedupe = 1 if hook == "operand" else 0
|
||||
js = JS % (hook_off, IDX_OFF, PC_BASE, CB_BASE, PC_STRIDE, dedupe)
|
||||
|
||||
count = {"rows": 0}
|
||||
f = 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("kind") == "ready":
|
||||
print(f"[frida] hook live @ base {pl['base']} + {pl['hook']} (mode={hook})"); return
|
||||
if pl.get("kind") == "batch":
|
||||
for cb, off in pl["rows"]:
|
||||
f.write(json.dumps({"codebase": cb, "offset": off}) + "\n")
|
||||
count["rows"] += len(pl["rows"])
|
||||
|
||||
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("[frida] not found. .exe processes:",
|
||||
[(p.pid, p.name) for p in procs if p.name.lower().endswith(".exe")])
|
||||
f.close(); return 2
|
||||
|
||||
crashed = {"v": False}
|
||||
session.on("detached", lambda reason, *a: crashed.update(v=(reason != "application-requested")) or
|
||||
print(f"[frida] session detached: {reason}"))
|
||||
script = session.create_script(js)
|
||||
script.on("message", on_message)
|
||||
script.load() # synchronous: the Interceptor is installed before this returns
|
||||
LIVE.write_text("live", encoding="utf-8") # signal to the orchestrator: hooks are live, safe to trigger
|
||||
print(f"[frida] attached to {proc}; capturing {seconds}s ({hook} hook). HOOK LIVE — trigger now.")
|
||||
print(" >>> let the opening auto-advance to the first wait-for-input. <<<")
|
||||
try:
|
||||
for _ in range(seconds):
|
||||
time.sleep(1)
|
||||
except KeyboardInterrupt:
|
||||
print("[frida] stopped early (Ctrl-C).")
|
||||
try:
|
||||
script.exports_sync.flush() # final flush of the tail
|
||||
time.sleep(0.3)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
session.detach()
|
||||
except Exception:
|
||||
pass
|
||||
f.close()
|
||||
try:
|
||||
LIVE.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
print(f"\n[trace] wrote {count['rows']} entries -> {OUT}")
|
||||
if crashed["v"]:
|
||||
print("[GATE] the hook DESTABILIZED the game (unexpected detach). If this was --hook tick, "
|
||||
"re-run with --hook operand (proven-safe).")
|
||||
if count["rows"] == 0:
|
||||
print("[GATE] no entries captured — the hook never fired (game idle / wrong hook / ecx!=ctx). "
|
||||
"Ensure the game is EXECUTING the opening while capturing.")
|
||||
return 3
|
||||
# quick shape: distinct codebases + top few by volume
|
||||
from collections import Counter
|
||||
c = Counter()
|
||||
with open(OUT, encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
c[json.loads(line)["codebase"]] += 1
|
||||
print(f"[trace] distinct codebases: {len(c)}")
|
||||
for cb, n in c.most_common(8):
|
||||
print(f" 0x{cb:08x}: {n} ops")
|
||||
return 0
|
||||
|
||||
|
||||
def main():
|
||||
args = sys.argv[1:]
|
||||
hook = "tick"
|
||||
if "--hook" in args:
|
||||
hook = args[args.index("--hook") + 1]
|
||||
rest = [a for a in args if not a.startswith("-") and a not in ("tick", "operand")]
|
||||
seconds = int(rest[0]) if rest and rest[0].isdigit() else 20
|
||||
proc = next((a for a in rest if not a.isdigit()), "AGE.EXE")
|
||||
if hook not in ("tick", "operand"):
|
||||
print("usage: trace_engine_ops.py [--hook tick|operand] [seconds] [proc]"); return 1
|
||||
return capture(hook, seconds, proc)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user