re(frida): import-map recon — export scan + clustering gate
GATE-PASS: 23342 exports mapped, 277 in-range pointer matches clustering into a module-resident import table at RVA 0x16f000 (VA 0x56f000); 23 singletons. ~254 real imports vs pe-sieve's 17-in-noise. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
168
tools/frida/map_imports.py
Normal file
168
tools/frida/map_imports.py
Normal file
@@ -0,0 +1,168 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Map the packer's resolved import pointers (RVA -> dll!Func) from the LIVE game, so the
|
||||||
|
dynamically-resolved Win32 APIs can be named at their call sites in the /v2 Ghidra image.
|
||||||
|
|
||||||
|
Why: AGE.EXE ships a zeroed IAT resolved via GetProcAddress at load (pe-sieve /imp fails, see
|
||||||
|
docs/engine-re.md). But the resolved pointers sit in memory. This tool builds {runtime_addr ->
|
||||||
|
dll!Func} from the live process's module exports, then scans the 0x400000 module for aligned DWORDs
|
||||||
|
holding those addresses -> RVA -> name. RVAs into the fixed-base 0x400000 module are ASLR-stable, so
|
||||||
|
labels computed live apply to the earlier dump (docs/superpowers/specs/2026-07-09-frida-import-map-design.md).
|
||||||
|
|
||||||
|
Read-only Frida (enumerateModules/enumerateExports + memory reads) — the safe plain-JS pattern
|
||||||
|
(mirrors dump_engine.py); no spawn, no GetProcAddress hook, no patching.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
py -3.11 -u -X utf8 tools/frida/map_imports.py --recon # report clustering (THE GATE); writes nothing
|
||||||
|
py -3.11 -u -X utf8 tools/frida/map_imports.py # write build/import-map.json (Task 2)
|
||||||
|
"""
|
||||||
|
import struct
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
REPO = Path(__file__).resolve().parents[2]
|
||||||
|
BUILD = REPO / "build"
|
||||||
|
MODULE = "AGE.EXE"
|
||||||
|
DUMP_SIZE = 0x260000 # /v2 dump covers 0x400000..0x660000; RVA >= this is out-of-dump
|
||||||
|
CHUNK = 2 * 1024 * 1024 # keep each frida message small (matches dump_engine.py)
|
||||||
|
|
||||||
|
|
||||||
|
# ---- pure logic (unit-tested; no frida needed) ---------------------------------------------------
|
||||||
|
|
||||||
|
def build_export_index(exports):
|
||||||
|
"""[{address:int, name, module}] -> {int addr: 'module!name'} (first name wins; skip null)."""
|
||||||
|
idx = {}
|
||||||
|
for e in exports:
|
||||||
|
a = int(e["address"])
|
||||||
|
if a and a not in idx:
|
||||||
|
idx[a] = f"{e['module']}!{e['name']}"
|
||||||
|
return idx
|
||||||
|
|
||||||
|
|
||||||
|
def scan_pointer_matches(mem, base_va, index):
|
||||||
|
"""Aligned little-endian DWORD scan; return [(rva, value, name)] for values in index."""
|
||||||
|
out = []
|
||||||
|
n = len(mem) & ~3
|
||||||
|
for off in range(0, n, 4):
|
||||||
|
v = struct.unpack_from("<I", mem, off)[0]
|
||||||
|
name = index.get(v)
|
||||||
|
if name is not None:
|
||||||
|
out.append((off, v, name))
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def cluster_runs(rvas, stride=4):
|
||||||
|
"""Group sorted RVAs into contiguous aligned runs -> [(start_rva, count)]."""
|
||||||
|
runs = []
|
||||||
|
for r in sorted(rvas):
|
||||||
|
if runs and r == runs[-1][0] + runs[-1][1] * stride:
|
||||||
|
s, c = runs[-1]
|
||||||
|
runs[-1] = (s, c + 1)
|
||||||
|
else:
|
||||||
|
runs.append((r, 1))
|
||||||
|
return runs
|
||||||
|
|
||||||
|
|
||||||
|
# ---- live collection (frida) ---------------------------------------------------------------------
|
||||||
|
|
||||||
|
JS = r"""
|
||||||
|
const CHUNK = %d;
|
||||||
|
const MODULE = '%s';
|
||||||
|
|
||||||
|
// exports of every loaded module, one message per module (bounded size)
|
||||||
|
Process.enumerateModules().forEach(function (m) {
|
||||||
|
var ex = m.enumerateExports().map(function (e) {
|
||||||
|
return {address: e.address.toString(), name: e.name};
|
||||||
|
});
|
||||||
|
send({kind: 'exports', module: m.name, exports: ex});
|
||||||
|
});
|
||||||
|
|
||||||
|
// the target module image, chunked (scan happens in Python against the tested pure logic)
|
||||||
|
var mod = Process.getModuleByName(MODULE);
|
||||||
|
send({kind: 'module', base: mod.base.toString(), size: mod.size});
|
||||||
|
for (var off = 0; off < mod.size; off += CHUNK) {
|
||||||
|
var n = Math.min(CHUNK, mod.size - off);
|
||||||
|
var buf = null;
|
||||||
|
try { buf = mod.base.add(off).readByteArray(n); } catch (e) {}
|
||||||
|
if (buf === null) { send({kind: 'gap', off: off, n: n}); continue; }
|
||||||
|
send({kind: 'chunk', off: off, n: n}, buf);
|
||||||
|
}
|
||||||
|
send({kind: 'alldone'});
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
def collect():
|
||||||
|
"""Attach to the live game; return (exports_list, module_base_va, module_bytes)."""
|
||||||
|
import frida
|
||||||
|
exports = []
|
||||||
|
state = {"base": None, "size": 0, "done": False}
|
||||||
|
mem = bytearray()
|
||||||
|
|
||||||
|
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"]
|
||||||
|
k = pl.get("kind")
|
||||||
|
if k == "exports":
|
||||||
|
for e in pl["exports"]:
|
||||||
|
exports.append({"address": int(e["address"], 16), "name": e["name"], "module": pl["module"]})
|
||||||
|
elif k == "module":
|
||||||
|
state["base"] = int(pl["base"], 16)
|
||||||
|
state["size"] = pl["size"]
|
||||||
|
if len(mem) < pl["size"]:
|
||||||
|
mem.extend(b"\x00" * (pl["size"] - len(mem)))
|
||||||
|
elif k == "chunk":
|
||||||
|
mem[pl["off"]:pl["off"] + pl["n"]] = data
|
||||||
|
elif k == "gap":
|
||||||
|
print(f" [gap] +0x{pl['off']:x} n=0x{pl['n']:x}")
|
||||||
|
elif k == "alldone":
|
||||||
|
state["done"] = True
|
||||||
|
|
||||||
|
try:
|
||||||
|
session = frida.attach(MODULE)
|
||||||
|
except Exception:
|
||||||
|
dev = frida.get_local_device()
|
||||||
|
hits = [p for p in dev.enumerate_processes() if "age" in p.name.lower()]
|
||||||
|
if not hits:
|
||||||
|
print("[frida] AGE.EXE not found — launch the game to the title, then re-run.")
|
||||||
|
sys.exit(2)
|
||||||
|
session = dev.attach(hits[0].pid)
|
||||||
|
|
||||||
|
script = session.create_script(JS % (CHUNK, MODULE))
|
||||||
|
script.on("message", on_message)
|
||||||
|
script.load()
|
||||||
|
import time
|
||||||
|
for _ in range(600): # up to ~60s; exits early on alldone
|
||||||
|
if state["done"]:
|
||||||
|
break
|
||||||
|
time.sleep(0.1)
|
||||||
|
session.detach()
|
||||||
|
return exports, state["base"], bytes(mem[:state["size"]])
|
||||||
|
|
||||||
|
|
||||||
|
def recon():
|
||||||
|
exports, base, mem = collect()
|
||||||
|
idx = build_export_index(exports)
|
||||||
|
matches = scan_pointer_matches(mem, base, idx)
|
||||||
|
rvas = [r for r, _, _ in matches]
|
||||||
|
runs = cluster_runs(rvas)
|
||||||
|
big = [(s, c) for s, c in runs if c >= 3]
|
||||||
|
inrange = [r for r in rvas if r < DUMP_SIZE]
|
||||||
|
print(f"exports mapped = {len(idx)} module bytes = 0x{len(mem):x}")
|
||||||
|
print(f"pointer matches = {len(matches)} in-range (<0x{DUMP_SIZE:x}) = {len(inrange)}")
|
||||||
|
print(f"contiguous runs (>=3): {[hex(s) + ':' + str(c) for s, c in big]}")
|
||||||
|
print(f"singletons = {sum(1 for _, c in runs if c == 1)}")
|
||||||
|
if big and inrange:
|
||||||
|
print("GATE-PASS: module-resident import table present and in dump range -> proceed to Task 2.")
|
||||||
|
else:
|
||||||
|
print("GATE-FAIL: no in-module import table (pointers likely heap-resident) -> STOP, document.")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
if "--recon" in sys.argv[1:]:
|
||||||
|
recon()
|
||||||
|
else:
|
||||||
|
from map_imports_full import build # Task 2 writer
|
||||||
|
build()
|
||||||
72
tools/frida/test_map_imports.py
Normal file
72
tools/frida/test_map_imports.py
Normal file
@@ -0,0 +1,72 @@
|
|||||||
|
"""Unit tests for the pure scan/match/cluster logic in map_imports.py.
|
||||||
|
|
||||||
|
Run: py -3.11 -X utf8 tools/frida/test_map_imports.py (plain runner, no pytest; frida not needed).
|
||||||
|
"""
|
||||||
|
import struct
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from map_imports import build_export_index, scan_pointer_matches, cluster_runs
|
||||||
|
|
||||||
|
FAILS = []
|
||||||
|
|
||||||
|
|
||||||
|
def check(cond, msg):
|
||||||
|
if not cond:
|
||||||
|
FAILS.append(msg)
|
||||||
|
print("FAIL:", msg)
|
||||||
|
else:
|
||||||
|
print("ok:", msg)
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_index_canonicalizes():
|
||||||
|
idx = build_export_index([
|
||||||
|
{"address": 0x76d80e70, "name": "LoadLibraryA", "module": "kernel32.dll"},
|
||||||
|
{"address": 0x76d7f7f0, "name": "GetProcAddress", "module": "kernel32.dll"},
|
||||||
|
])
|
||||||
|
check(idx[0x76d80e70] == "kernel32.dll!LoadLibraryA", "index maps addr -> dll!func")
|
||||||
|
|
||||||
|
|
||||||
|
def test_export_index_first_name_wins_and_skips_zero():
|
||||||
|
idx = build_export_index([
|
||||||
|
{"address": 0x1000, "name": "First", "module": "a.dll"},
|
||||||
|
{"address": 0x1000, "name": "Second", "module": "a.dll"}, # alias -> ignored
|
||||||
|
{"address": 0x0, "name": "Nope", "module": "a.dll"}, # null addr -> skipped
|
||||||
|
])
|
||||||
|
check(idx[0x1000] == "a.dll!First" and 0x0 not in idx, "first name wins; null addr skipped")
|
||||||
|
|
||||||
|
|
||||||
|
def test_scan_matches_little_endian_aligned():
|
||||||
|
idx = {0x76d80e70: "kernel32.dll!LoadLibraryA", 0x76d7f7f0: "kernel32.dll!GetProcAddress"}
|
||||||
|
mem = bytearray(0x20)
|
||||||
|
struct.pack_into("<I", mem, 0x10, 0x76d80e70)
|
||||||
|
struct.pack_into("<I", mem, 0x14, 0x76d7f7f0)
|
||||||
|
struct.pack_into("<I", mem, 0x18, 0x12345678) # not an export -> no match
|
||||||
|
m = scan_pointer_matches(bytes(mem), 0x400000, idx)
|
||||||
|
check([(r, v) for r, v, _ in m] == [(0x10, 0x76d80e70), (0x14, 0x76d7f7f0)],
|
||||||
|
"scan finds aligned LE pointer matches, skips non-exports")
|
||||||
|
|
||||||
|
|
||||||
|
def test_scan_ignores_unaligned():
|
||||||
|
idx = {0x76d80e70: "kernel32.dll!LoadLibraryA"}
|
||||||
|
mem = bytearray(0x10)
|
||||||
|
struct.pack_into("<I", mem, 0x2, 0x76d80e70) # unaligned -> ignored
|
||||||
|
check(scan_pointer_matches(bytes(mem), 0x400000, idx) == [], "unaligned pointer ignored")
|
||||||
|
|
||||||
|
|
||||||
|
def test_cluster_runs_groups_contiguous():
|
||||||
|
check(cluster_runs([0x100, 0x104, 0x108, 0x200]) == [(0x100, 3), (0x200, 1)],
|
||||||
|
"cluster groups contiguous aligned runs, isolates singleton")
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
test_export_index_canonicalizes()
|
||||||
|
test_export_index_first_name_wins_and_skips_zero()
|
||||||
|
test_scan_matches_little_endian_aligned()
|
||||||
|
test_scan_ignores_unaligned()
|
||||||
|
test_cluster_runs_groups_contiguous()
|
||||||
|
print("FAILURES:", len(FAILS))
|
||||||
|
return 1 if FAILS else 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
Reference in New Issue
Block a user