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
|
#!/usr/bin/env python3
"""Hunt GCP's three lifetime defects across every TA, on decompiled pseudocode.
All three defects this round produced share one habit: object state is maintained separately
from the operations that consume it. Each has a signature that survives decompilation:
A destroy-clears-only-a-flag
A destroy path writes 0 to ONE field of a slot while the create path zero-fills the
whole slot, so the identity field (the handle) survives the destroy.
-> GCP FE0D: clears the state word, leaves slot+0.
B lookup-without-liveness
A scan over a fixed-count array compares one field and returns another, without ever
testing the slot's state/liveness field.
-> GCP FE0F: memcmp on slot+0x18, no state test, returns slot+0.
C counter-committed-before-a-failing-acquire
A counter is incremented (or decremented) and then a call is made whose failure path
returns WITHOUT restoring the counter.
-> GCP FE0C: ++refcount, then asd_rpc_305 fails, error returned with no rollback.
-> GCP FE0D: decrement taken on a path where the handle was already rejected.
Detection works on pseudocode text because every function in every TA now decompiles and
carries a prototype -- before that work this analysis was not possible.
VALIDATION: --selftest requires the detector to rediscover all three known GCP defects. A
detector that cannot find the bugs it was modelled on reports nothing trustworthy about the
TAs where the answer is unknown.
Usage:
./ta_lifetime_audit.py --selftest
./ta_lifetime_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"))
ZERO_FILL = re.compile(r"zero_fill\w*\s*\(", re.I)
# field writes, in every form Hex-Rays emits for these images
CLEAR = re.compile(r"(?:\*\(\(_DWORD \*\)(\w+) \+ (\d+)\)|\*\(_DWORD \*\)\((\w+) \+ (\d+)\)|(\w+)->(\w+))\s*=\s*0\s*;")
# identity comparison against a caller-supplied value
IDENT_CMP = re.compile(r"(?:\*\(\(_DWORD \*\)\w+ \+ \d+\)|\*\(_DWORD \*\)\(\w+ \+ \d+\)|\w+->\w+)\s*==\s*\w+")
PTR_STEP = re.compile(r"\+\+(\w+)\s*;|(\w+)\s*\+=\s*\d+\s*;")
FIELD_OF = re.compile(r"(\w+)->(\w+)")
STATE_WORDS = ("state", "status", "live", "active", "inuse", "flag", "valid")
def analyse(name: str, code: str) -> list[str]:
"""Signatures of the three GCP lifetime defects, matched against real Hex-Rays text."""
hits = []
body = "\n".join(l for l in code.splitlines() if not l.strip().startswith("//"))
lines = [l.strip() for l in body.splitlines()]
low = body.lower()
# ---- A: clears one field, never zero-fills, yet validates an identity -----------
clears = CLEAR.findall(body)
if clears and not ZERO_FILL.search(body) and IDENT_CMP.search(body):
hits.append("A:destroy-clears-%d-field(s)-no-zero-fill-identity-checked" % len(clears))
# ---- B: pointer-stepping scan that never consults a liveness field --------------
stepped = {m[0] or m[1] for m in PTR_STEP.findall(body)}
in_loop = any(re.match(r"(for|while)\s*\(", l) for l in lines)
if in_loop and stepped:
# does the function read a field off the stepped pointer (i.e. it is a struct scan)?
touched = {v for v, f in FIELD_OF.findall(body) if v in stepped}
if touched and not any(w in low for w in STATE_WORDS):
hits.append("B:scan-over-%s-without-liveness-test" % ",".join(sorted(touched)))
# ---- C: PERSISTENT counter changed, then an early return with no rollback -------
# The first version matched any lvalue and drowned in loop induction variables --
# ~50 hits, nearly all `++i` in library code. A refcount is persistent state: a
# dereference, an indexed slot, or a struct field. Bare locals are excluded.
PERSIST = re.compile(r"(\*\s*\w+|\w+->\w+|\w+\[\d+\])\s*=|\+\+\s*(\*\s*\w+|\w+->\w+)|--\s*(\*\s*\w+|\w+->\w+)")
for i, l in enumerate(lines):
m = PERSIST.search(l)
if not m:
continue
lv = (m.group(1) or m.group(2) or m.group(3) or "").replace(" ", "")
if not lv or not re.search(r"\+\+|--|\+\s*1\b|-\s*1\b", l):
continue
direction = "down" if ("--" in l or re.search(r"-\s*1\b", l)) else "up"
tail = lines[i + 1: i + 16]
called = any(re.search(r"\w+\s*\(", t) for t in tail)
def is_rollback(t: str) -> bool:
t2 = t.replace(" ", "")
if lv not in t2:
return False
return ("--" + lv in t2 or lv + "-1" in t2) if direction == "up" \
else ("++" + lv in t2 or lv + "+1" in t2)
restored = any(is_rollback(t) for t in tail)
returns = any(t.startswith("return") or "goto LABEL" in t for t in tail)
if called and returns and not restored:
hits.append("C:persistent-counter-'%s'-%s-then-early-return-without-rollback" % (lv, direction))
break
return hits
# the three known defects the detector must rediscover
CONTROL = {
"gcp_cmdFE0D_DestroyCryptoSession": ("A", "C"),
"gcp_cmdFE0F_GetCryptoSessionForDevice": ("B",),
"gcp_cmdFE0C_CreateCryptoSession": ("C",),
}
def scan_ta(prefix: str, elf: str):
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:
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(it["name"], 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:
found = scan_ta("gcp", idx["gcp"]["elf"])
ok = True
print("SELFTEST -- must rediscover the three known GCP defects:")
for fn, classes in CONTROL.items():
got = found.get(fn, [])
got_cls = {h[0] for h in got}
missing = [c for c in classes if c not in got_cls]
print(" %-44s wanted %s got %s %s"
% (fn, list(classes), sorted(got_cls) or "none",
"OK" if not missing else "MISSED %s" % missing))
ok &= not missing
print("\nverdict: %s" % ("TRUSTWORTHY" if ok else "NOT trustworthy -- do not report findings"))
return 0 if ok else 1
targets = sorted(idx) if args.all else [args.prefix]
for p in targets:
try:
res = scan_ta(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) with a lifetime signature" % (p, len(res)))
for fn, h in sorted(res.items()):
print(" %-40s %s" % (fn, "; ".join(h)))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|
