1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
#!/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/<ta>.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 <prefix>_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())
|
