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
|
#!/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('<Q', pos[0])
adev.gmc.flush_hdp()
adev.doorbell64[doorbell] = pos[0]
t0 = time.time()
while time.time() - t0 < timeout_s:
if struct.unpack('<Q', bytes(adev.vram.view(rptr, 8)[:]))[0] >= 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())
|
