#!/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())