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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
|
#!/usr/bin/env python3
"""Hunt memory-corruption primitives: host data reaching a length or an index unbounded.
Everything this round has produced is state confusion, not memory corruption. The shapes
that would be memory corruption are:
L a call whose LENGTH argument derives from the host request, with no bound compare on
that value anywhere in the function (the HDCP subcmd12 / MERIT cmd05 shape)
X an array or pointer INDEX derived from the host request, scaled and dereferenced, with
no bound compare (the powerplay 0x109a shape, but host-side)
S a host-derived value used as a size for a stack buffer or a loop trip count that walks
a fixed-size local
Earlier copy audits worked on raw disassembly and tracked R2. This one works on decompiled
pseudocode, which is only possible now that every function in every TA decompiles and
carries a prototype with `params` typed -- so "derives from the host request" is a textual
fact (`params->...`) rather than an inference about a register.
CONTROLS: the detector must rediscover the two known unbounded-length leads --
`merit_cmd05` (memcmp length from params+0x4c) and `gcp_cmdFE09` (unwrap length from the
request). A detector that cannot find those says nothing trustworthy about the rest.
Usage:
./ta_memsafety_audit.py --selftest
./ta_memsafety_audit.py --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"))
# a value that came from the host request
# The host request is NOT always the second argument. In MERIT it arrives FIRST: cmd05
# reads its length from *((_DWORD *)ctx + 19) and its data from ctx + 80, and even passes
# ctx + 20 as a helper's `params`. My role heuristic had typed those handlers
# (void *ctx, struct ta_params *params), which hid the known unbounded length from the
# first version of this detector. So taint BOTH argument pointers and label the source;
# triage separates a genuine host buffer from an internal context by reading.
HOST_SRC = re.compile(
r"(?:params->\w+|params->data20\[\d+\]|\*\(_\w+ \*\)&?params(?:->\w+)?(?:\[\d+\])?"
r"|\*\(_\w+ \*\)\(\s*a[12]\s*\+\s*\d+\s*\)|a[12]\[\d+\]"
r"|\*\(\(_\w+ \*\)(?:ctx|params|a1|a2) \+ \d+\)"
r"|\((?:int|char \*|unsigned int)\)(?:ctx|params) \+ \d+)")
ASSIGN = re.compile(r"^\s*(\w+)\s*=\s*(.+?);\s*$")
CALL = re.compile(r"(\w+)\s*\(([^;]*)\)")
BOUND = re.compile(r"(?:if|while)\s*\(.*?\b%s\b\s*(?:<=|<|>|>=|==|!=)\s*(?:0x[0-9a-f]+|\d+)")
BOUND_REV = re.compile(r"(?:if|while)\s*\(.*?(?:0x[0-9a-f]+|\d+)\s*(?:<=|<|>|>=)\s*\b%s\b")
INDEX = re.compile(r"(\w+)\s*\[\s*(\w+)\s*\]")
# calls that are not memory operations: logging dominates these images
NOISE = re.compile(r"_log\b|log\w*\(|printf|zero_fill|snprintf", re.I)
def tainted_vars(lines: list[str]) -> dict[str, str]:
"""Variables carrying host-request data, with the line that made them so."""
taint: dict[str, str] = {}
for _ in range(3): # cheap fixpoint for one/two-hop propagation
for l in lines:
m = ASSIGN.match(l)
if not m:
continue
dst, rhs = m.group(1), m.group(2)
if dst in taint:
continue
if HOST_SRC.search(rhs):
taint[dst] = l.strip()
else:
for t in list(taint):
if re.search(r"\b%s\b" % re.escape(t), rhs):
taint[dst] = l.strip()
break
return taint
def is_bounded(body: str, var: str) -> bool:
v = re.escape(var)
return bool(re.search(BOUND.pattern % v, body) or re.search(BOUND_REV.pattern % v, body))
def analyse(code: str) -> list[str]:
body = "\n".join(l for l in code.splitlines() if not l.strip().startswith("//"))
lines = body.splitlines()
taint = tainted_vars(lines)
hits = []
# ---- F: host-derived length together with a FIXED-SIZE LOCAL buffer --------------
# This is the actual memory-corruption shape, and it cuts the noise that positional
# length-guessing produces: a call that receives both a stack array of known size AND
# a host-derived length that nothing bounds. Sizes come from the decompiler's own
# declarations (`_DWORD v19[4]; // [sp+...]`).
locals_sz = {m.group(2): int(m.group(3))
for m in re.finditer(r"(_[A-Z]+|char|int|unsigned \w+)\s+(\w+)\[(\d+)\]", body)}
for l in lines:
if NOISE.search(l):
continue
for fname, args in CALL.findall(l):
if fname in ("if", "while", "for", "switch", "return", "sizeof", "memset") \
or NOISE.search(fname):
continue
parts = [re.sub(r"^a\d+:\s*", "", a.strip()) for a in args.split(",")]
if len(parts) < 2:
continue
buf = next((re.sub(r"[^\w]", "", p2) for p2 in parts
if re.sub(r"[^\w]", "", p2) in locals_sz), None)
lastm = re.fullmatch(r"\(?\w*\)?\s*(\w+)", parts[-1])
lenv = lastm.group(1) if lastm else None
if buf and lenv and lenv in taint and not is_bounded(body, lenv):
hits.append("F:*** %s(%s[%d bytes], ..., len=%s) host length, UNBOUNDED ***"
% (fname, buf, locals_sz[buf] * 4, lenv))
# ---- L: host-derived length in a call argument -----------------------------------
for l in lines:
if NOISE.search(l):
continue
for fname, args in CALL.findall(l):
if NOISE.search(fname) or fname in ("if", "while", "for", "switch", "return",
"sizeof", "memset"):
continue
parts = [a.strip() for a in args.split(",")]
for pos, a in enumerate(parts):
# a length is normally the 2nd or later argument
if pos == 0:
continue
bare = re.sub(r"^a\d+:\s*", "", a)
m = re.fullmatch(r"\(?\w+\)?\s*(\w+)", bare) or re.fullmatch(r"(\w+)", bare)
v = m.group(1) if m else None
direct = HOST_SRC.search(bare)
if (v in taint or direct) and not (v and is_bounded(body, v)):
src = taint.get(v, "direct from request")
hits.append("L:len-arg%d of %s <- %s (unbounded)" % (pos + 1, fname, v or bare))
break
# ---- X: host-derived index into an array -----------------------------------------
for arr, idx in INDEX.findall(body):
if idx in taint and not is_bounded(body, idx):
hits.append("X:index %s[%s] host-derived, unbounded" % (arr, idx))
return sorted(set(hits))
CONTROL = ("merit_cmd05", "gcp_cmdFE09")
def scan(prefix: str, elf: str, want: str | None = None):
from idatui import remote_ops
from idatui.nexus_client import NexusClient
c = NexusClient(str(REPO / elf))
c.connect(timeout=900)
raw, off = [], 0
while True:
rr = c.call(remote_ops.list_funcs, queries=[{"count": 500, "offset": off}])
blk = (rr.get("result") or [{}])[0]
raw += blk.get("data") or []
total = int(blk.get("total") or 0)
off = int(blk.get("next_offset") or 0)
if not off or len(raw) >= total:
break
out = {}
for it in raw:
if want and want not in it["name"]:
continue
r = c.call(remote_ops.decompile, addr=it["addr"], include_addresses=False)
code = (r.get("code") or "") if isinstance(r, dict) else ""
if not code:
continue
h = analyse(code)
if h:
out[it["name"]] = h
c.close()
return out
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--selftest", action="store_true")
ap.add_argument("--all", action="store_true")
ap.add_argument("--prefix")
args = ap.parse_args()
idx = json.loads((HERE / "syms" / "index.json").read_text())
if args.selftest:
ok = True
print("SELFTEST -- must rediscover the two known unbounded-length leads:")
for fn in CONTROL:
ta = "merit" if fn.startswith("merit") else "gcp"
found = scan(ta, idx[ta]["elf"], want=fn.split("_cmd")[0] + "_cmd")
got = {k: v for k, v in found.items() if fn in k}
print(" %-16s %s" % (fn, ("FOUND: %s" % list(got.values())[0][:2]) if got else "MISSED"))
ok &= bool(got)
print("\nverdict: %s" % ("TRUSTWORTHY" if ok else "NOT trustworthy"))
return 0 if ok else 1
for p in (sorted(idx) if args.all else [args.prefix]):
try:
res = scan(p, idx[p]["elf"])
except Exception as exc: # noqa: BLE001
print("%-10s SKIP (%s)" % (p, str(exc)[:60]))
continue
print("\n=== %s: %d function(s)" % (p, len(res)))
for fn, h in sorted(res.items()):
print(" %-38s %s" % (fn, "; ".join(h[:3])))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|
