P1.5 SOLVED: live ctx (0x2370040, +0x24==48000) + level-tracker A[] + mask scalars captured via realtime playback; method play.lua + rtctx_rt.py; prior 'unreachable' was offline-engine artifact
This commit is contained in:
+193
@@ -0,0 +1,193 @@
|
||||
#!/usr/bin/env python3
|
||||
"""rtctx_rt.py — P1.5: find live ctx via REAPER realtime transport playback.
|
||||
|
||||
Launches REAPER with render_long.rpp + play.lua (realtime audio playback, host
|
||||
audio callback alive), then snapshots the yabridge-host memory during playback
|
||||
and searches for the DSP ctx by FUN_180535ae0 constructor markers.
|
||||
|
||||
Usage: rtctx_rt.py
|
||||
"""
|
||||
import subprocess, os, glob, time, struct, sys, json
|
||||
|
||||
SNAP = '/tmp/snap_rt.bin'
|
||||
|
||||
|
||||
def mem_open(pid):
|
||||
return os.open(f'/proc/{pid}/mem', os.O_RDONLY)
|
||||
|
||||
|
||||
def find_host():
|
||||
for p in glob.glob('/proc/[0-9]*'):
|
||||
pid = int(os.path.basename(p))
|
||||
try:
|
||||
cmd = open(f'/proc/{pid}/cmdline', 'rb').read().replace(b'\0', b' ').decode('utf8', 'replace')
|
||||
maps = open(f'/proc/{pid}/maps').read()
|
||||
except Exception:
|
||||
continue
|
||||
if 'soothe2' in maps and 'reaper' not in cmd:
|
||||
return pid
|
||||
return None
|
||||
|
||||
|
||||
def snapshot(host):
|
||||
fd = mem_open(host)
|
||||
out = open(SNAP, 'wb')
|
||||
nreg = 0; nbytes = 0; t0 = time.time()
|
||||
for line in open(f'/proc/{host}/maps').read().splitlines():
|
||||
p = line.split()
|
||||
if len(p) < 2: continue
|
||||
lo, hi = (int(x, 16) for x in p[0].split('-'))
|
||||
if 'r' not in p[1]: continue
|
||||
a = lo
|
||||
while a < hi:
|
||||
n = min(hi - a, 8 * 1024 * 1024)
|
||||
try:
|
||||
d = os.pread(fd, n, a)
|
||||
except Exception:
|
||||
a += n; continue
|
||||
if not d:
|
||||
a += n; continue
|
||||
out.write(struct.pack('<QQ', a, len(d))); out.write(d)
|
||||
nreg += 1; nbytes += len(d)
|
||||
a += n
|
||||
out.close()
|
||||
os.close(fd)
|
||||
return nreg, nbytes
|
||||
|
||||
|
||||
def parse_snap():
|
||||
data = open(SNAP, 'rb').read()
|
||||
regs = []
|
||||
i = 0
|
||||
while i + 16 <= len(data):
|
||||
lo, sz = struct.unpack_from('<QQ', data, i)
|
||||
regs.append((lo, data[i + 16:i + 16 + sz]))
|
||||
i += 16 + sz
|
||||
return regs
|
||||
|
||||
|
||||
def readabs(regs, addr, n):
|
||||
for lo, body in regs:
|
||||
if lo <= addr < lo + len(body) and addr - lo + n <= len(body):
|
||||
return body[addr - lo:addr - lo + n]
|
||||
return None
|
||||
|
||||
|
||||
def find_ctx(regs):
|
||||
"""Search for DSP ctx by FUN_180535ae0 ctor markers:
|
||||
+0x24==40000.0f, +0x540874==1.0, +0x54087c==0.5, +0x54088c==1.0."""
|
||||
sig24 = struct.pack('<I', 0x472c4400) # 40000.0f
|
||||
f1 = struct.pack('<f', 1.0)
|
||||
fhalf = struct.pack('<f', 0.5)
|
||||
cands = []
|
||||
for lo, body in regs:
|
||||
j = 0
|
||||
while True:
|
||||
j = body.find(sig24, j)
|
||||
if j < 0:
|
||||
break
|
||||
base = lo + j - 0x24
|
||||
if (readabs(regs, base + 0x540874, 4) == f1 and
|
||||
readabs(regs, base + 0x54087c, 4) == fhalf and
|
||||
readabs(regs, base + 0x54088c, 4) == f1):
|
||||
cands.append(base)
|
||||
j += 1
|
||||
return cands
|
||||
|
||||
|
||||
def find_registry(regs):
|
||||
okc = {8193, 2049, 2048, 32768, 16384, 1024, 4096}
|
||||
hits = []
|
||||
for lo, body in regs:
|
||||
if len(body) < 0x2000: continue
|
||||
for off in range(0, len(body) - 64, 8):
|
||||
good = True
|
||||
for k in range(4):
|
||||
if off + 16 + 8 > len(body): good = False; break
|
||||
c = struct.unpack_from('<Q', body, off + k * 16)[0]
|
||||
if c not in okc: good = False; break
|
||||
p = struct.unpack_from('<Q', body, off + k * 16 + 8)[0]
|
||||
if readabs(regs, p, 4) is None: good = False; break
|
||||
if good: hits.append(lo + off)
|
||||
hits = sorted(set(hits))
|
||||
dedup = []
|
||||
for h in hits:
|
||||
if not dedup or h - dedup[-1] > 0x40:
|
||||
dedup.append(h)
|
||||
return dedup
|
||||
|
||||
|
||||
def dump_targets(regs, ctx):
|
||||
out = {}
|
||||
def f32(addr):
|
||||
b = readabs(regs, addr, 4)
|
||||
return struct.unpack('<f', b)[0] if b else None
|
||||
def f64arr(addr, n):
|
||||
b = readabs(regs, addr, 8 * n)
|
||||
return list(struct.unpack('<%dd' % n, b)) if b and len(b) >= 8 * n else None
|
||||
out['scalars'] = {
|
||||
'0x24': f32(ctx + 0x24),
|
||||
'0x540870': f32(ctx + 0x540870),
|
||||
'0x540874': f32(ctx + 0x540874),
|
||||
'0x54087c': f32(ctx + 0x54087c),
|
||||
'0x540888': f32(ctx + 0x540888),
|
||||
'0x54088c': f32(ctx + 0x54088c),
|
||||
}
|
||||
out['bandconfig0'] = {
|
||||
'A': f32(ctx + 0x188),
|
||||
'B': f32(ctx + 0x188 + 0x04),
|
||||
'gamma': f32(ctx + 0x188 + 0x0c),
|
||||
'flag': readabs(regs, ctx + 0x188 + 0x10, 1),
|
||||
}
|
||||
for name, off, cnt in [('0x4c0528', 0x4c0528, 32),
|
||||
('0x3c0510', 0x3c0510, 32),
|
||||
('0x2c04f8', 0x2c04f8, 32)]:
|
||||
a = f64arr(ctx + off, cnt)
|
||||
if a is not None:
|
||||
out[name] = a
|
||||
return out
|
||||
|
||||
|
||||
def main():
|
||||
os.system("pkill -9 -x reaper 2>/dev/null; pkill -9 -f '[y]abridge' 2>/dev/null; sleep 1")
|
||||
proc = subprocess.Popen(
|
||||
['/usr/bin/reaper', '-nosplash', '-ignoreerrors',
|
||||
'/home/m/soothe-bt/render_long.rpp', '/home/m/re-tools/play.lua'],
|
||||
stdout=open('/dev/null', 'w'), stderr=subprocess.STDOUT)
|
||||
t0 = time.time(); host = None
|
||||
while time.time() - t0 < 60 and not host:
|
||||
host = find_host(); time.sleep(0.2)
|
||||
if not host:
|
||||
print('NO HOST'); return 1
|
||||
print('host', host, 'at %.1fs' % (time.time() - t0))
|
||||
time.sleep(8)
|
||||
for attempt in range(3):
|
||||
nreg, nbytes = snapshot(host)
|
||||
print('snapshot #%d: %d regs %.1f MB' % (attempt, nreg, nbytes / 1e6))
|
||||
if nbytes > 50e6:
|
||||
break
|
||||
time.sleep(3)
|
||||
if proc.poll() is None:
|
||||
proc.kill()
|
||||
regs = parse_snap()
|
||||
hits = find_registry(regs)
|
||||
rb = None
|
||||
for h in hits:
|
||||
b = readabs(regs, h, 8)
|
||||
if b and struct.unpack('<Q', b)[0] == 8193:
|
||||
rb = h; break
|
||||
print('registry base:', ('0x%x' % rb) if rb else 'NONE')
|
||||
cands = find_ctx(regs)
|
||||
print('ctx candidates:', [('0x%x' % c) for c in cands])
|
||||
if cands:
|
||||
ctx = cands[0]
|
||||
dump = dump_targets(regs, ctx)
|
||||
print(json.dumps(dump, indent=1, default=str))
|
||||
with open('/tmp/rtctx_live.json', 'w') as f:
|
||||
json.dump(dump, f, indent=1, default=str)
|
||||
print('saved /tmp/rtctx_live.json')
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user