#!/usr/bin/env python3 """Per-command audit of the DMCUB inbox1 surface: for every implemented `DMUB_CMD__*` type, follow the command frame through the handler closure and report every place a host-supplied byte reaches a dangerous sink. Requires capstone >= 6 (Xtensa): /home/user/.venvs/cs6/bin/python Method ------ 1. Locate the inbox1 dispatcher by its 129-entry jump table (the table whose `jx` index bound is 0x81), read the table, and map type -> case stub -> handler function(s) (the stub either calls handlers or inlines the work). 2. Forward taint, interprocedural, argument-passing only, with **two** taint kinds kept apart (conflating them buries the real hits in reply writes): P = pointer into the host-supplied frame / host-mapped memory V = host-supplied *data* (a value loaded through a P) * the handler's first argument is the command frame, so param 0 starts as P; * a load through P yields V; arithmetic propagates both kinds; * spills are modelled: `s32i aX, a1, off` records the kind of aX for that stack slot and `l32i aY, a1, off` restores it. Without this the analysis misses the known ABM finding entirely -- `0x6002973c` spills the frame pointer to `a1+4` and reloads it before the call that overflows. The `--selftest` mode asserts that hit is still found, as a positive control. * `callN` propagates to the callee's a2..a5 (call8 ABI) with the kind preserved, and the callee is re-analysed for that signature (worklist to a fixpoint, depth-capped); * a store *through* P is a reply write and is deliberately not a finding. Control flow is ignored: this over-approximates on purpose. Every hit is a review item, not a bug. 3. Sinks reported: * `COPY_LEN` -- V is the length argument of the byte-copy routine * `COPY_DST` -- V-derived pointer is the copy destination * `LOOP_CNT` -- V is a `loop*` trip count * `IDX_R` / `IDX_W` -- V (usually scaled by mul16*/mull/slli/addx*) is used as the base of a load / store: an array index taken from host data * `IDX_FRAME` -- V used as an offset into the frame itself (confined to the mailbox window, so low severity, but reported) Each hit records whether the tainted value was compared against something (`bgeu/bltu/bgeui/bltui/bge/blt/bgei/blti`) earlier in the same function, which is what distinguishes a bounded index from an unbounded one. usage: dmcub_cmdaudit.py [--types 64,66] [--verbose] [--closure] """ import argparse import collections import importlib.util import os import re import struct HERE = os.path.dirname(os.path.abspath(__file__)) _s = importlib.util.spec_from_file_location('dmcub_dis', os.path.join(HERE, 'dmcub_dis.py')) dis = importlib.util.module_from_spec(_s) _s.loader.exec_module(dis) ARGREGS = ['a2', 'a3', 'a4', 'a5', 'a6', 'a7'] CALLREGS = ['a10', 'a11', 'a12', 'a13', 'a14', 'a15'] SCALE = ('mul16s', 'mul16u', 'mull', 'slli', 'addx2', 'addx4', 'addx8') ARITH = SCALE + ('add', 'add.n', 'addi', 'addi.n', 'addmi', 'sub', 'and', 'or', 'xor', 'srli', 'sra', 'srai', 'extui', 'moveqz', 'movnez', 'min', 'max', 'minu', 'maxu', 'neg', 'abs', 'sext') CMPS = ('bgeu', 'bgeui', 'bltu', 'bltui', 'bge', 'blt', 'bgei', 'blti') LOADS = ('l8ui', 'l16ui', 'l16si', 'l32i', 'l32i.n') STORES = ('s8i', 's16i', 's32i', 's32i.n') def ops(x): return [p.strip() for p in x.op_str.split(',')] class CmdAudit: def __init__(self, path, abi=None): self.a = dis.Analysis(path) self.copy_fn = self._find_copy() self.names = self._abi_names(abi) self.dispatcher, self.table = self._find_dispatcher() self.frame_reg = self._find_frame_reg() self.icalls = {} for f in self.a.funcs: t = self.a.indirect_targets(f) if t: self.icalls[f] = t # ---- locate things -------------------------------------------------- def _find_copy(self): cands = [] for f, (s, e) in self.a.funcs.items(): ins = self.a.dis(f) pairs = sum(1 for i, x in enumerate(ins[:-1]) if x.mnemonic == 'l8ui' and ins[i + 1].mnemonic == 's8i') if pairs >= 4 and any(x.mnemonic.startswith('loop') for x in ins): cands.append((self.a.calls[f], -(e - s), f)) return max(cands)[2] if cands else None def _find_dispatcher(self): """The function whose `jx` is guarded by a 0x81 bound over a 129-entry table: the inbox1 command dispatch.""" for f in sorted(self.a.funcs): ins = self.a.dis(f) for i, x in enumerate(ins): if x.mnemonic != 'jx': continue base = None for y in range(i - 1, max(-1, i - 20), -1): if ins[y].mnemonic == 'l32r': base = self.a.lit(ins[y]) break if base is None: continue n = 0 while n < 200: v = self.a.img.word(base + 4 * n) if v is None or not self.a.func_of(v): break n += 1 if n >= 120: return f, base return None, None def _find_frame_reg(self): """Which register holds the command-frame pointer at the case stubs. This is **not** stable across builds and must not be hardcoded: 3.2.0, 3.5 and 3.6 keep the frame pointer in `a3`, but the DCN 4.0.1 dispatcher was restructured to load it into `a4` from `a2[0]`. Seeding the wrong register taints nothing, so every command silently reports a closure of one function and zero sinks -- a clean-looking audit of code that was never examined. That is exactly what happened before this was added. Detection follows the dispatch itself: the type byte is `l8ui aT, aF, 0`, aT indexes the jump table via `addx4`, and aF is therefore the frame pointer. Falls back to `a3`. """ if self.dispatcher is None: return 'a3' ins = self.a.dis(self.dispatcher) for i, x in enumerate(ins): if x.mnemonic != 'jx': continue # the register jx jumps through idx = None for y in range(i - 1, max(-1, i - 24), -1): if ins[y].mnemonic == 'addx4': o = ops(ins[y]) if len(o) == 3: idx = o[1] break if idx is None: continue # aF from the l8ui that defines the index at offset 0 for y in range(i - 1, max(-1, i - 40), -1): if ins[y].mnemonic != 'l8ui': continue o = ops(ins[y]) if len(o) == 3 and o[0] == idx and o[2] in ('0', '0x0'): return o[1] return 'a3' @staticmethod def _abi_names(path): if not path or not os.path.exists(path): return {} src = open(path).read() m = re.search(r'enum dmub_cmd_type\s*\{(.*?)\};', src, re.S) if not m: return {} body = re.sub(r'/\*.*?\*/', '', m.group(1), flags=re.S) out, cur = {}, 0 for item in body.split(','): item = item.strip() mm = re.match(r'^(\w+)\s*(?:=\s*(\w+))?$', item) if not mm: continue if mm.group(2): try: cur = int(mm.group(2), 0) except ValueError: pass out[cur] = mm.group(1) cur += 1 return out def handler_map(self): """type -> (case_stub, case_end, [handler functions called by the stub]) The case *end* matters: case bodies sit back-to-back in the dispatcher and each rejoins the ring loop with a backwards `j`. Without an end bound, a taint root placed at one case walks every case above it and every closure looks like the union of all commands. """ ents = [self.a.img.word(self.table + 4 * i) for i in range(129)] default = collections.Counter(ents).most_common(1)[0][0] out = {} dins = self.a.dis(self.dispatcher) for t, v in enumerate(ents): if v == default: continue handlers = [] end = None for x in dins: if x.address < v: continue if x.mnemonic.startswith('call') and not x.mnemonic.startswith('callx'): tgt = self.a.target(x) if tgt in self.a.funcs: handlers.append(tgt) if x.mnemonic in ('retw.n', 'retw') or ( x.mnemonic == 'j' and (self.a.target(x) or 0) < v): end = x.address + x.size break out[t] = (v, end if end is not None else v + 0x40, handlers) return out, default # ---- taint ---------------------------------------------------------- def taint_scan(self, roots, max_funcs=400): """roots: list of (fva, start_va_or_None, [(reg, kind)]), kind in {'P','V'}. `start_va` lets a root begin part-way into a function, which is how the dispatcher's inline command cases are handled: each case is rooted at its own stub address with the frame pointer (a3) tainted, instead of guessing which callee is "the handler" -- that guess made type 1 root at the software-divide routine. """ hits = [] seen = set() work = [(f, sv, frozenset(sig)) for f, sv, sig in roots] closure = set() while work and len(seen) < max_funcs: fva, start_va, sig = work.pop() if (fva, start_va, sig) in seen or fva not in self.a.funcs: continue seen.add((fva, start_va, sig)) closure.add(fva) ins = self.a.dis_reachable(fva) if start_va is not None: lo, hi = start_va if isinstance(start_va, tuple) else (start_va, None) ins = [x for x in ins if x.address >= lo and (hi is None or x.address < hi)] P = set(r for r, k in sig if k == 'P') V = set(r for r, k in sig if k == 'V') scaled, compared = set(), set() slots = {} # stack offset -> (kind, scaled, compared) if ('a1slots', 'P') in sig: # The dispatcher batches a multi-command burst by storing each # frame pointer into a stack array (`addx4 a9, a4, a1; s32i`), and # several cases hand that array (or a slot of it) to the handler # instead of the pointer in a3 -- types 67, 68 and 78 do. Seeding # the slots as frame pointers keeps those handlers in scope. P.add('a1') for k in range(0, 0x40, 4): slots[k] = ('P', False, False) def kill(r): P.discard(r) V.discard(r) scaled.discard(r) for x in ins: p = ops(x) m = x.mnemonic if m in CMPS: for r in p[:2]: if r in V or r in P: compared.add(r) continue if m in LOADS and len(p) == 3 and p[1] == 'a1': dst = p[0] try: off = int(p[2], 0) except ValueError: off = None kill(dst) if off in slots: kind, sc, cm = slots[off] (P if kind == 'P' else V).add(dst) if sc: scaled.add(dst) if cm: compared.add(dst) continue if m in STORES and len(p) == 3 and p[1] == 'a1': try: off = int(p[2], 0) except ValueError: continue src = p[0] if src in P: slots[off] = ('P', src in scaled, src in compared) elif src in V: slots[off] = ('V', src in scaled, src in compared) else: slots.pop(off, None) continue if m in LOADS and len(p) == 3: dst, base = p[0], p[1] if base in V: hits.append(self._hit('IDX_R', fva, x, base in scaled, base in compared)) kill(dst) V.add(dst) elif base in P: kill(dst) V.add(dst) # host data else: kill(dst) continue if m in STORES and len(p) == 3: base = p[1] if base in V: kind = 'IDX_FRAME' if base in P else 'IDX_W' hits.append(self._hit(kind, fva, x, base in scaled, base in compared)) continue # store through P = reply write, benign if m.startswith('loop'): if p and p[0] in V: hits.append(self._hit('LOOP_CNT', fva, x, p[0] in scaled, p[0] in compared)) continue if m.startswith('callx'): # sub-command dispatch through a function-pointer table sub = [(ARGREGS[k], 'P' if r in P else 'V') for k, r in enumerate(CALLREGS[:4]) if r in P or r in V] for cand in self.icalls.get(fva, {}).get(x.address, []): if sub and cand in self.a.funcs: work.append((cand, None, frozenset(sub))) if not sub: hits.append(self._hit('ICALL_NOARG', fva, x, False, False)) for r in CALLREGS: kill(r) continue if m.startswith('call') and not m.startswith('callx'): tgt = self.a.target(x) if tgt == self.copy_fn: if CALLREGS[2] in V: hits.append(self._hit('COPY_LEN', fva, x, CALLREGS[2] in scaled, CALLREGS[2] in compared)) if CALLREGS[0] in V: hits.append(self._hit('COPY_DST', fva, x, CALLREGS[0] in scaled, CALLREGS[0] in compared)) elif tgt in self.a.funcs: sub = [(ARGREGS[k], 'P' if r in P else 'V') for k, r in enumerate(CALLREGS[:4]) if r in P or r in V] if sub: work.append((tgt, None, frozenset(sub))) for r in CALLREGS: kill(r) continue if m in ARITH and len(p) >= 2: srcs = [r for r in p[1:] if re.fullmatch(r'a\d+', r)] anyP = any(r in P for r in srcs) anyV = any(r in V for r in srcs) wasscaled = m in SCALE or any(r in scaled for r in srcs) wascmp = any(r in compared for r in srcs) kill(p[0]) if anyP: P.add(p[0]) if anyV: V.add(p[0]) if (anyP or anyV) and wasscaled: scaled.add(p[0]) if wascmp: compared.add(p[0]) continue if m in ('mov', 'mov.n') and len(p) == 2: src = p[1] wasscaled, wascmp = src in scaled, src in compared inP, inV = src in P, src in V kill(p[0]) if inP: P.add(p[0]) if inV: V.add(p[0]) if wasscaled: scaled.add(p[0]) if wascmp: compared.add(p[0]) continue if p and re.fullmatch(r'a\d+', p[0]): kill(p[0]) return hits, closure, len(seen) >= max_funcs @staticmethod def _hit(kind, fva, x, scaled, compared): return dict(kind=kind, fn=fva, at=x.address, insn='%s %s' % (x.mnemonic, x.op_str), scaled=scaled, compared=compared) # ---- provenance of an index ---------------------------------------- def provenance(self, fva, at, reg=None): """Backward slice of the base register of the load/store at `at`. Returns (frame_offset, scale, bounded) where frame_offset is the byte offset in the command frame the index was loaded from (or None), scale is the multiplier/shift applied to it, and bounded says whether that register met a comparison or a bit-scan-style mask before use. This is what turns "1500 index sinks" into a handful of reviewable facts: most sinks are the same per-instance index reused. """ ins = self.a.dis_reachable(fva) idx = [k for k, x in enumerate(ins) if x.address == at] if not idx: return (None, None, None) i = idx[0] if reg is None: p = ops(ins[i]) reg = p[1] if len(p) >= 2 else None scale = None bounded = False track = {reg} off = None for y in range(i - 1, max(-1, i - 60), -1): x = ins[y] p = ops(x) m = x.mnemonic if m in CMPS and len(p) >= 2 and (p[0] in track or p[1] in track): bounded = True continue if not p or p[0] not in track: continue if m in ('l8ui', 'l16ui', 'l16si') and len(p) == 3: try: off = int(p[2], 0) except ValueError: off = None track.add(p[1]) continue if m in SCALE and len(p) >= 3: try: scale = int(p[2], 0) if not p[2].startswith('a') else scale except ValueError: pass if m in ('mul16s', 'mul16u', 'mull'): for y2 in range(y - 1, max(-1, y - 12), -1): q = ops(ins[y2]) if q and q[0] in (p[1], p[2]) and ins[y2].mnemonic in ('movi', 'movi.n'): try: scale = int(q[1], 0) except ValueError: pass break elif m == 'slli' and len(p) == 3: try: scale = 1 << int(p[2], 0) except ValueError: pass elif m.startswith('addx'): scale = {'addx2': 2, 'addx4': 4, 'addx8': 8}[m] track.update(r for r in p[1:] if re.fullmatch(r'a\d+', r)) continue if m in ARITH or m in ('mov', 'mov.n'): track.update(r for r in p[1:] if re.fullmatch(r'a\d+', r)) if m in ('extui', 'and'): bounded = bounded or True # masked: range-limited by construction continue return (off, scale, bounded) INTERESTING = ('COPY_LEN', 'COPY_DST', 'LOOP_CNT', 'IDX_W', 'IDX_R', 'IDX_FRAME') def main(): ap = argparse.ArgumentParser() ap.add_argument('image') # Prefer the pinned pre-deprecation ABI: upstream master deleted command IDs # 1-4 (REG_SEQ_*/REG_REG_WAIT) from `enum dmub_cmd_type` in commit 9a591ae691b5 # (2026-06-05), so naming from master alone prints "(not in ABI)" for commands # the firmware still dispatches. See fetch_ref.sh. _abi_default = os.path.join(HERE, 'ref', 'dmub_cmd_v6.17.h') if not os.path.exists(_abi_default): _abi_default = os.path.join(HERE, 'ref', 'dmub_cmd.h') ap.add_argument('--abi', default=_abi_default) ap.add_argument('--types', default=None) ap.add_argument('--verbose', action='store_true') ap.add_argument('--closure', action='store_true') ap.add_argument('--index-report', action='store_true', help='dedupe index sinks into (frame offset, stride, bounded) facts') ap.add_argument('--sub-audit', type=lambda s: int(s, 0), default=None, help='audit one command type per sub-command: enumerate the ' "handler's jx sub-table and taint each case separately " '(used for DMUB_CMD__VBIOS, 27 sub-commands)') ap.add_argument('--selftest', action='store_true', help='positive control: the known ABM_INIT_CONFIG COPY_LEN ' 'hit (type 66) must be reported') args = ap.parse_args() if args.selftest: c = CmdAudit(args.image, args.abi) hmap, _ = c.handler_map() stub, end, handlers = hmap[66] hits, closure, _ = c.taint_scan( [(c.dispatcher, (stub, end), [(c.frame_reg, 'P'), ('a1slots', 'P')])]) copylen = [h for h in hits if h['kind'] == 'COPY_LEN'] print('type 66 closure %d fns, COPY_LEN hits: %d' % (len(closure), len(copylen))) for h in copylen: print(' fn 0x%08x @0x%08x %s' % (h['fn'], h['at'], h['insn'])) ok = any(h['fn'] == 0x6003eebc for h in copylen) print('POSITIVE CONTROL %s' % ('PASS' if ok else 'FAIL -- results not trustworthy')) raise SystemExit(0 if ok else 1) c = CmdAudit(args.image, args.abi) print('%s: dispatcher 0x%08x, table 0x%08x, copy routine 0x%08x, frame reg %s' % ( os.path.basename(args.image), c.dispatcher, c.table, c.copy_fn, c.frame_reg)) hmap, default = c.handler_map() print('implemented command types: %d (default stub 0x%08x)\n' % (len(hmap), default)) if args.sub_audit is not None: t = args.sub_audit stub, end, handlers = hmap[t] if not handlers: print('type %d has no called handler' % t) return H = handlers[0] ins = c.a.dis_reachable(H) tbl = guard = None for i, x in enumerate(ins): if x.mnemonic != 'jx': continue for y in range(i - 1, max(-1, i - 20), -1): if ins[y].mnemonic == 'l32r': tbl = c.a.lit(ins[y]) break for y in range(i - 1, max(-1, i - 20), -1): if ins[y].mnemonic in ('bgeu', 'bgeui', 'bltu', 'bltui'): guard = '%s %s' % (ins[y].mnemonic, ins[y].op_str) break break if tbl is None: print('no jx sub-table in handler 0x%08x' % H) return ents = [] while True: v = c.a.img.word(tbl + 4 * len(ents)) if v is None or not c.a.func_of(v): break ents.append(v) print('type %d handler 0x%08x: sub-table 0x%08x, %d entries, guard: %s' % ( t, H, tbl, len(ents), guard)) for i, v in enumerate(ents): f = c.a.func_of(v) nxt = [e for e in sorted(set(ents)) if e > v] case_end = min(nxt) if nxt else c.a.img.va(c.a.funcs[f][1]) hits, closure, _ = c.taint_scan( [(f, (v, case_end), [('a3', 'P'), ('a2', 'P'), ('a1slots', 'P')])]) risky = [h for h in hits if h['kind'] in INTERESTING] kinds = collections.Counter(h['kind'] for h in risky) facts = collections.Counter() for h in risky: if h['kind'] in ('IDX_R', 'IDX_W'): off, scale, bnd = c.provenance(h['fn'], h['at']) if off is not None: facts[(h['fn'], off, scale, bnd)] += 1 print(' sub[%2d] case 0x%08x closure=%-3d sinks=%-4d %s' % ( i, v, len(closure), len(risky), dict(kinds) or '')) for (fn, off, scale, bnd), n in sorted(facts.items(), key=lambda kv: kv[0][3] is True): print(' fn 0x%08x arg+0x%-4x stride %-6s bounded=%-5s x%d' % ( fn, off, hex(scale) if scale else '?', bnd, n)) return want = None if args.types: want = set(int(t, 0) for t in args.types.split(',')) unreached = [] for t in sorted(hmap): if want and t not in want: continue stub, end, handlers = hmap[t] roots = [(c.dispatcher, (stub, end), [(c.frame_reg, 'P'), ('a1slots', 'P')])] hits, closure, capped = c.taint_scan(roots) # Structural control: if a case stub calls handlers but taint reached # none of them, the frame pointer was seeded in the wrong register and # the "0 sinks" below is meaningless. Reported per image, not per type, # because the seeding is global. if handlers and not (set(handlers) & set(closure)): unreached.append(t) risky = [h for h in hits if h['kind'] in INTERESTING] unguarded = [h for h in risky if not h['compared']] print('type %-3d %-36s case 0x%08x..0x%08x handlers %s' % ( t, c.names.get(t, '(not in ABI)'), stub, end, ' '.join('0x%08x' % h for h in handlers) or 'inline')) print(' closure %d fns%s | host-data sinks: %d (%d without a compare)' % ( len(closure), ' (CAPPED)' if capped else '', len(risky), len(unguarded))) counts = collections.Counter(h['kind'] for h in risky) if counts: print(' ' + ' '.join('%s=%d' % kv for kv in sorted(counts.items()))) if args.index_report: facts = collections.Counter() for h in risky: if h['kind'] in ('IDX_R', 'IDX_W', 'IDX_FRAME'): facts[(h['fn'],) + c.provenance(h['fn'], h['at'])] += 1 def _k(kv): fn, off, scale, bounded = kv[0] return (bounded is True, fn, -1 if off is None else off, -1 if scale is None else scale) for (fn, off, scale, bounded), n in sorted(facts.items(), key=_k): print(' idx fn 0x%08x frame+%-6s stride %-6s bounded=%-5s x%d' % ( fn, hex(off) if off is not None else '?', hex(scale) if scale else '?', bounded, n)) for h in risky: if h['kind'] in ('COPY_LEN', 'COPY_DST', 'LOOP_CNT'): print(' %-9s fn 0x%08x @0x%08x %s' % ( h['kind'], h['fn'], h['at'], h['insn'])) print() continue show = risky if args.verbose else [h for h in risky if h['kind'] in ('COPY_LEN', 'LOOP_CNT') or not h['compared']] for h in show[:40]: print(' %-10s fn 0x%08x @0x%08x %-34s scaled=%-5s compared=%s' % ( h['kind'], h['fn'], h['at'], h['insn'], h['scaled'], h['compared'])) if args.closure: print(' closure: ' + ' '.join('0x%08x' % f for f in sorted(closure))) print() if unreached: print('WARNING: %d/%d types call a handler that taint never entered ' '(frame reg %s may be wrong for this build): %s' % ( len(unreached), len(hmap), c.frame_reg, ' '.join(str(t) for t in unreached))) print('Treat every "0 sinks" result in this run as UNVERIFIED.') if __name__ == '__main__': main()