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
180
181
182
183
184
185
186
187
|
#!/usr/bin/env python3
"""Which of the Windows-driver TAs does THIS Navi 31 PSP accept?
`fw/AUDIT-WINDOWS-DRIVER-CORPUS.md` extracted 19 PSP module containers from the Windows
Adrenalin 22.12.1 package, including ten TA types the linux-firmware bundle never ships
(OPM, MERIT, GCP, WSPE, TMZ, AUC, FP, XGMI, RAP, POWERPLAY). Those containers carry
`apu_family_id = 0xffffffff`, so nothing in the file says which ASIC they target.
The PSP answers that for us: `LOAD_TA` authenticates the container against this part's
keys, so the status separates "wrong part / not for this silicon" from "a TA this PSP
will actually accept" — without invoking anything.
Controls, both required:
POS a TA carved from this machine's own `psp_13_0_0_ta.bin` must load (status 0).
Without it, an all-reject result would be indistinguishable from a broken harness.
NEG a junk buffer must be rejected, so "accept" means something.
Any *successful* load allocates a PSP session and this rig has a known
session-exhaustion wall (`hw/loadta_ring_exhaustion.py`), so every success is unloaded
immediately.
usage: PYTHONPATH=/root/psp/tinygrad loadta_windows_sweep.py --dir /root/psp/cont22
"""
import argparse
import glob
import os
import struct
import sys
import time
STATUS = {0x00: 'ACCEPTED', 0x34: 'signature verify failed',
0x01: 'invalid cmd', 0x02: 'invalid params'}
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--dir', default='/root/psp/cont22')
ap.add_argument('--start', type=int, default=0,
help='batch start index; the PSP hits a session wall around 45 '
'loads per boot, so sweep in batches with AM_RESET=1 between')
ap.add_argument('--count', type=int, default=0, help='batch size (0 = all)')
ap.add_argument('--reverse', action='store_true',
help='sweep in reverse order: results must not depend on ordering')
ap.add_argument('--ta-bin', default='/lib/firmware/amdgpu/psp_13_0_0_ta.bin')
args = ap.parse_args()
from tinygrad import Device
import tinygrad.runtime.autogen.am.am as am
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
adev = Device["AMD"].iface.dev_impl
psp = adev.psp
cmd_pa = adev.mm.palloc(0x10000, align=0x1000, zero=True, boot=False)
mcc = adev.paddr2mc(cmd_pa)
app_pa = adev.mm.palloc(0x80000, align=0x1000, zero=True, boot=False)
def load_ta(blob):
n = (len(blob) + 3) & ~3 # USB4 needs 4-byte aligned size
adev.vram.view(app_pa, n)[:] = blob + b'\x00' * (n - len(blob))
adev.gmc.flush_hdp()
mc = adev.paddr2mc(app_pa)
c = am.struct_psp_gfx_cmd_resp(cmd_id=am.GFX_CMD_ID_LOAD_TA)
t = c.cmd.cmd_load_ta
t.app_phy_addr_lo, t.app_phy_addr_hi = mc & 0xFFFFFFFF, mc >> 32
t.app_len = len(blob)
t.cmd_buf_phy_addr_lo, t.cmd_buf_phy_addr_hi = mcc & 0xFFFFFFFF, mcc >> 32
t.cmd_buf_len = 0x10000
try:
r = psp._ring_submit(c)
return r.resp.status & 0xFFFFFFFF, r.resp.session_id
except Exception as e:
try:
return int(str(e).strip().split()[-1]) & 0xFFFFFFFF, None
except Exception:
return -1, None
def unload(sid):
"""Unload and REPORT the status.
The first full-corpus run swallowed this and leaked a session per accepted
TA, hitting the PSP session wall after ~45 loads; every later command then
returned a counter-like value incrementing by 0x10 instead of a status, and
the condition persisted across processes until AM_RESET=1.
"""
c = am.struct_psp_gfx_cmd_resp(cmd_id=am.GFX_CMD_ID_UNLOAD_TA)
c.cmd.cmd_unload_ta.session_id = sid
try:
return psp._ring_submit(c).resp.status & 0xFFFFFFFF
except Exception as e:
try:
return int(str(e).strip().split()[-1]) & 0xFFFFFFFF
except Exception:
return -1
def show(tag, st, sid, extra=''):
name = STATUS.get(st, 'status %#x' % st)
print(' %-34s -> %-24s %s%s'
% (tag, name, ('sid=%#x' % sid) if st == 0 and sid else '', extra),
flush=True)
# ---- POSITIVE control: a TA from this machine's own firmware bundle ----
import subprocess
tab = args.ta_bin
if not os.path.exists(tab) and os.path.exists(tab + '.zst'):
subprocess.run(['zstd', '-d', '-f', '-o', '/tmp/ta.bin', tab + '.zst'],
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
tab = '/tmp/ta.bin'
pos_ok = False
if os.path.exists(tab):
data = open(tab, 'rb').read()
# same carve rules as fw/carve_driver.py
pos = 0
while True:
m = data.find(b'$PS1', pos)
if m < 0:
break
pos = m + 4
cont = m - 0x10
if cont < 0:
continue
n = struct.unpack_from('<I', data, cont + 0x14)[0]
if not (0 < n < 0x400000) or cont + 0x200 + n > len(data):
continue
body = data[cont + 0x100: cont + 0x200 + n]
if body[:4] != b'\x00\x02\x00\x00':
continue
st, sid = load_ta(data[cont:cont + 0x200 + n])
show('POS ctrl: bundle TA @%#x' % cont, st, sid)
if st == 0:
pos_ok = True
if sid:
unload(sid)
break
if not pos_ok:
print(' POSITIVE CONTROL FAILED -- a bundle TA did not load, so a')
print(' sweep of rejects below would prove nothing. Stopping.')
return 1
# ---- NEGATIVE control ----
st, sid = load_ta(b'\xa5' * 0x2000)
show('NEG ctrl: junk 0x2000', st, sid)
if st == 0:
print(' NEGATIVE CONTROL FAILED -- junk accepted. Stopping.')
return 1
# ---- the sweep ----
print('\nWindows-driver containers:')
accepted = []
prev_unknown = [None]
KNOWN_ST = {0x00, 0x34, 0xffff0000, 0xffff000a, 0xffff0006, 0xffff0008}
def degraded(st):
"""Detect the session-wall signature: unknown statuses marching by 0x10."""
if st in KNOWN_ST:
prev_unknown[0] = None
return False
if prev_unknown[0] is not None and st - prev_unknown[0] == 0x10:
return True
prev_unknown[0] = st
return False
files = sorted(glob.glob(os.path.join(args.dir, '*.bin')), reverse=args.reverse)
if args.count:
files = files[args.start:args.start + args.count]
elif args.start:
files = files[args.start:]
for p in files:
blob = open(p, 'rb').read()
st, sid = load_ta(blob)
extra = ''
if st == 0:
us = unload(sid)
extra = ' unload=%#x%s' % (us, '' if us == 0 else ' <-- SESSION LEAK')
accepted.append(os.path.basename(p))
show(os.path.basename(p), st, sid, extra)
if degraded(st):
print('\n ABORT: statuses are marching by 0x10 -- this is the PSP')
print(' session wall, not a verdict. Results past this point would be')
print(' fabricated. Re-run with AM_RESET=1.')
return 2
time.sleep(0.05)
print('\nACCEPTED by this PSP: %d %s' % (len(accepted), accepted))
return 0
if __name__ == '__main__':
raise SystemExit(main())
|
