#!/usr/bin/env python3 """Emit a per-TA command-surface dump: every command handler, decompiled and readable. GCP got one of these; the other eleven TAs did not, and the command surface is precisely what an analyst reads first. The obstacle is that `fw/cmdmaps/*.json` records the handler by NAME, captured before the renames, and several entries are `loc_XXXX` labels that are not function starts at all. Both are resolved here: * `sub_XXXX` / `loc_XXXX` -> the address is in the label, so parse it; * any other name -> look it up in fw/libdumps/.json, which pairs the OLD names with addresses, and addresses are stable across renames; * either way, map the address to its CONTAINING function in the live database, because a `loc_` label is usually a block inside the real handler. Every emitted handler is checked to decompile; failures are listed in the file rather than quietly omitted, so the artifact never overstates coverage. Usage: ./ta_surface.py --prefix opm (or --all) """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path HERE = Path(__file__).resolve().parent REPO = HERE.parent sys.path.insert(0, str(Path.home() / "dev" / "ida-tui-maybe")) def resolve_targets(prefix: str) -> dict[str, int]: """command -> raw target address, from the recorded map. Prefer _ea.json when it exists: that map was decoded straight from the dispatcher's jump tables and stores ADDRESSES, so it is authoritative. GCP has one, and using the name-keyed map for GCP instead resolved only 4 of its 17 handlers -- which silently replaced a good artifact with a worse one on the first run. """ ea_map = HERE / "cmdmaps" / ("%s_ea.json" % prefix) if ea_map.exists(): return {k: int(v, 16) for k, v in json.loads(ea_map.read_text()).items()} cm = json.loads((HERE / "cmdmaps" / ("%s.json" % prefix)).read_text()) dump = json.loads((HERE / "libdumps" / ("%s.json" % prefix)).read_text()) by_name = {f["name"]: f["ea"] for f in dump} out: dict[str, int] = {} for cmd, target in cm.items(): m = re.match(r"^(?:sub|loc|off)_([0-9A-Fa-f]+)$", str(target)) if m: out[cmd] = int(m.group(1), 16) elif target in by_name: out[cmd] = by_name[target] return out def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--prefix") ap.add_argument("--all", action="store_true") args = ap.parse_args() from idatui import remote_ops from idatui.nexus_client import NexusClient idx = json.loads((HERE / "syms" / "index.json").read_text()) prefixes = sorted(idx) if args.all else [args.prefix] summary = [] for prefix in prefixes: targets = resolve_targets(prefix) if not targets: summary.append((prefix, 0, 0, "no command map")) continue c = NexusClient(str(REPO / idx[prefix]["elf"])) c.connect(timeout=600) # map each recorded target to the function that CONTAINS it # An exact hit on a function START is a real handler. A target that merely falls # INSIDE some function is not: rap's recorded loc_401000 lands inside `start`, and # emitting that as "the cmd 1 handler" would assert something false. Keep the two # classes apart. handlers: dict[int, list[str]] = {} uncertain: list[tuple[str, int, str]] = [] unresolved = [] for cmd, ea in sorted(targets.items(), key=lambda kv: int(kv[0], 16)): r = c.call(remote_ops.lookup_funcs, queries=["%x" % ea]) fn = ((r.get("result") or [{}])[0] or {}).get("fn") if not fn: unresolved.append((cmd, ea)) continue start = int(str(fn["addr"]), 16) if start == ea: handlers.setdefault(start, []).append(cmd) else: uncertain.append((cmd, ea, "%s+%#x" % (fn.get("name"), ea - start))) lines = [ "// %s TA command surface, decompiled with restored names, prototypes and types." % prefix.upper(), "// Source: %s" % json.loads((HERE / "syms" / ("%s.meta.json" % prefix)).read_text())["path"], "// Handlers resolved from fw/cmdmaps/%s.json; loc_* labels mapped to their" % prefix, "// containing function. Generated by fw/ta_surface.py.", "", ] ok = 0 for ea, cmds in sorted(handlers.items()): r = c.call(remote_ops.decompile, addr=hex(ea), include_addresses=False) code = (r.get("code") or "") if isinstance(r, dict) else "" tag = ", ".join(cmds) if not code or (isinstance(r, dict) and r.get("failed")): lines += ["// cmd %s @ %s -- DECOMPILATION FAILED: %s" % (tag, hex(ea), (r or {}).get("error")), ""] continue lines += ["// ================ cmd %s @ %s ================" % (tag, hex(ea)), code, ""] ok += 1 if uncertain: lines += ["", "// ---------------------------------------------------------------", "// COMMANDS ENTERED AT AN INTERNAL LABEL, NOT AT A FUNCTION ENTRY.", "// These dispatch into the middle of a larger function, which is what a", "// switch-style dispatcher produces. IDA refuses to split them out, and", "// undefining the owner to force a split was not worth the risk, so the", "// containing function is reproduced below IN FULL. Read the command's", "// code at the stated offset -- the function as a whole is NOT the handler.", "// ---------------------------------------------------------------"] lines += ["// cmd %-6s -> %s (enters %s)" % (a, hex(b), t) for a, b, t in uncertain] shown = set() for a, b, t in uncertain: owner = t.split("+")[0] rr = c.call(remote_ops.lookup_funcs, queries=["%x" % b]) fnn = ((rr.get("result") or [{}])[0] or {}).get("fn") or {} start = fnn.get("addr") if not start or start in shown: continue shown.add(start) d = c.call(remote_ops.decompile, addr=start, include_addresses=False) cc = (d.get("code") or "") if isinstance(d, dict) else "" if cc: lines += ["", "// ---- containing function %s @ %s (entered mid-body by the" " commands above) ----" % (owner, start), cc] if unresolved: lines += ["", "// unresolved recorded targets (no function contains them):"] lines += ["// cmd %s -> %s" % (a, hex(b)) for a, b in unresolved] path = REPO / "fw" / "pseudo" / ("%s_command_surface.c" % prefix) path.write_text("\n".join(lines)) summary.append((prefix, len(handlers), ok, "%d uncertain, %d unresolved" % (len(uncertain), len(unresolved)))) c.close() print("%-10s %9s %9s %s" % ("TA", "handlers", "decompiled", "notes")) for row in summary: print("%-10s %9s %9s %s" % row) return 0 if __name__ == "__main__": raise SystemExit(main())