#!/usr/bin/env python3 """wine_ptrace_trace.py — точный пер-оп захват FIR-цепи soothe2 через ptrace. Запускает reaper -renderproject как ребёнок (=> ptrace разрешён при любом yama scope), находит wine-хост yabridge (soothe2 в maps), прицепляется ко всем тредам, ставит int3 на входах EXP/DF0 ядра, на хитах читает регистры (PTRACE_GETREGS) и буферы через /proc/tid/mem; между хитами CONT. Брейкпоинты: EXP 0x1803831c0 rcx=buf, r8d=count(float) DF0 0x18000b3c0 rcx=FIR, rdx=track, r8d=n(пар) Плюс слоты контекста тем же мгновением (scr/trk/cur/FIR@540668). Запуск: python3 scripts/wine_ptrace_trace.py [n_hits] [outdir] """ import ctypes import os import pickle import signal import struct import subprocess import sys import time import numpy as np BP_EXP = 0x1803831c0 BP_DF0 = 0x18000b3c0 BP_COPY = 0x1800136e0 BP_DF0RET = 0x18052b898 BP_TRACKSAVE = 0x18052b574 BP_DIV = 0x1803a06a0 BP_DC40 = 0x1800dc40 BP_EXPVAR = 0x1802dc0e0 BP_FN = 0x180529fe0 CTX_SLOTS = {'scr': 0x540628, 'trk': 0x540688, 'cur': 0x540678, 'fir_ptr': 0x540668} libc = ctypes.CDLL('libc.so.6', use_errno=True) PTRACE_ATTACH = 16 PTRACE_DETACH = 17 PTRACE_CONT = 7 PTRACE_SINGLESTEP = 9 PTRACE_PEEKDATA = 2 PTRACE_POKEDATA = 5 PTRACE_GETREGS = 12 PTRACE_SETOPTIONS = 0x4200 PTRACE_O_TRACECLONE = 1 << 22 __WALL = 0x40000000 libc.ptrace.restype = ctypes.c_long libc.ptrace.argtypes = [ctypes.c_long, ctypes.c_long, ctypes.c_void_p, ctypes.c_void_p] class UserRegs(ctypes.Structure): _fields_ = [('r15', ctypes.c_uint64), ('r14', ctypes.c_uint64), ('r13', ctypes.c_uint64), ('r12', ctypes.c_uint64), ('rbp', ctypes.c_uint64), ('rbx', ctypes.c_uint64), ('r11', ctypes.c_uint64), ('r10', ctypes.c_uint64), ('r9', ctypes.c_uint64), ('r8', ctypes.c_uint64), ('rax', ctypes.c_uint64), ('rcx', ctypes.c_uint64), ('rdx', ctypes.c_uint64), ('rsi', ctypes.c_uint64), ('rdi', ctypes.c_uint64), ('orig_rax', ctypes.c_uint64), ('rip', ctypes.c_uint64), ('cs', ctypes.c_uint64), ('eflags', ctypes.c_uint64), ('rsp', ctypes.c_uint64), ('ss', ctypes.c_uint64), ('fs_base', ctypes.c_uint64), ('gs_base', ctypes.c_uint64), ('ds', ctypes.c_uint64), ('es', ctypes.c_uint64), ('fs', ctypes.c_uint64), ('gs', ctypes.c_uint64)] def pt(req, pid, addr=0, data=0): if not isinstance(data, int): data = ctypes.cast(data, ctypes.c_void_p) else: data = ctypes.c_void_p(data) return libc.ptrace(req, pid, ctypes.c_void_p(addr), data) def getregs(tid): r = UserRegs() if pt(PTRACE_GETREGS, tid, 0, ctypes.byref(r)) != 0: raise OSError('GETREGS tid=%d' % tid) return r def setregs(tid, r): if pt(PTRACE_SETREGS := 13, tid, 0, ctypes.byref(r)) != 0: raise OSError('SETREGS tid=%d' % tid) def peek(tid, addr): v = pt(PTRACE_PEEKDATA, tid, addr, 0) if v == -1: e = ctypes.get_errno() if e != 0: raise OSError(e) return v & 0xFFFFFFFFFFFFFFFF def poke(tid, addr, val): if pt(PTRACE_POKEDATA, tid, addr, val) == -1 and ctypes.get_errno(): raise OSError('POKEDATA %#x tid=%d: %d' % (addr, tid, ctypes.get_errno())) def find_host(): import glob 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 find_ctx(fd, pid): vt = struct.pack('= 0: cand = a + j sb = os.pread(fd, 4, cand + 0x540870) if sb and struct.unpack(' 100: return cand j = d.find(vt, j + 1) j = d.find(m48) while j >= 0: cand = a + j - 0x24 try: sb = os.pread(fd, 4, cand + 0x540870) if sb and struct.unpack(' 100: return cand except OSError: pass j = d.find(m48, j + 1) a += n return None def find_ctx_candidates(fd, pid, fir_ptr): """Все адреса X (кратные 8), где [X+0x540668]==fir_ptr => кандидат X.""" val = struct.pack('= 0: if j % 8 == 0: out.append(a + j - 0x540668) j = d.find(val, j + 1) a += n return out def main(): rpp = sys.argv[1] if len(sys.argv) > 1 else '/home/m/soothe-bt/dual_b1q_0.5.rpp' n_target = int(sys.argv[2]) if len(sys.argv) > 2 else 80 outdir = sys.argv[3] if len(sys.argv) > 3 else '/tmp/opencode/winetrace' os.makedirs(outdir, exist_ok=True) wav = None for ln in open(rpp, errors='replace'): if 'RENDER_FILE' in ln and '"' in ln: wav = ln.split('"')[1] break if wav and os.path.exists(wav): os.remove(wav) subprocess.run("pkill -9 -x reaper; pkill -9 -f '[y]abridge'; " "rm -rf /run/user/1000/yabridge-soothe2_x64-*; sleep 1", shell=True) proc = subprocess.Popen(['/usr/bin/reaper', '-nosplash', '-ignoreerrors', '-renderproject', rpp], stdout=open('/dev/null', 'w'), stderr=subprocess.STDOUT) t0 = time.time() host = None ctx_fd = None ctx = None # Фаза 1: ждём появления хоста и контекста ЧИТАЮЧЕЙ памятью (без ptrace), # чтобы не мешать загрузке плагина while time.time() - t0 < 25: if host is None: host = find_host() if host: try: ctx_fd = os.open(f'/proc/{host}/mem', os.O_RDONLY) print('host %d (+%.3fs)' % (host, time.time()-t0), flush=True) except OSError: host = None time.sleep(0.001) continue if host is not None: try: ctx = find_ctx(ctx_fd, host) except (ProcessLookupError, OSError): ctx = None host = None time.sleep(0.001) continue if ctx: break time.sleep(0.002) if not host or not ctx: print('NO HOST/CTX (host=%s ctx=%s)' % (host, ctx)) return 1 print('ctx %#x (+%.3fs)' % (ctx, time.time()-t0), flush=True) fd = ctx_fd def rd(a, n): return os.pread(fd, n, a) def rd_f32(a, n): return np.frombuffer(rd(a, 4*n), dtype=' 0x10000: rec_a[nm] = rd_f32(p, cnt) except OSError as e: rec_a['err'] = str(e) samples.append(rec_a) hits[kind] += 1 poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) continue if kind == 'DF0': track_by_tid[pid] = regs.rdx if kind == 'DF0RET': tp = track_by_tid.get(pid) rec_r = {'kind': 'DF0RET', 'tid': pid, 't': round(time.time()-t_start, 4)} try: if tp and tp > 0x10000: rec_r['track'] = rd_f32(tp, 2049*2) samples.append(rec_r) hits['DF0RET'] += 1 except OSError as e: rec_r['err'] = str(e) samples.append(rec_r) poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) continue if kind == 'COPY': try: cnt = min(regs.r9 & 0xFFFFFFFF, 2049) rec_c = {'kind': 'COPY', 'tid': pid, 't': round(time.time()-t_start, 4), 'src': rd_f32(regs.rcx, cnt), 'dst': regs.r8} # снять int3/step/restore как у остальных — общий код ниже poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) samples.append(rec_c) hits['COPY'] += 1 continue except OSError as e: print('copy err', e, flush=True) continue # ctx по фактическому указателю FIR из хита + валидация # инварианта trk==exp(scr) (24mm3), строгая if kind == 'DF0': good = None cands = find_ctx_candidates(fd, host, regs.rcx) for cand in cands: if cand <= 0x10000: continue try: v_sc = rd_f32(cand+CTX_SLOTS['scr'], 2049) v_tr = rd_f32(cand+CTX_SLOTS['trk'], 2049) except OSError: continue if not (np.isfinite(v_sc).all() and np.isfinite(v_tr).all()): continue if np.abs(v_sc).max() > 40: continue if np.allclose(v_tr, np.exp(v_sc), rtol=1e-3, atol=1e-9): good = cand break if pc_dbg := True: for cand in cands[:4]: try: vs = rd_f32(cand+CTX_SLOTS['scr'], 2049) vt = rd_f32(cand+CTX_SLOTS['trk'], 2049) except OSError: continue dmax = np.abs(vt-np.exp(np.clip(vs,-80,80))).max() print(' cand %#x: |scr|=%.4g |trk|=%.4g maxdiff=%.4g' % (cand, np.abs(vs).max(), np.abs(vt).max(), dmax), flush=True) print('cands=%d good=%s' % (len(cands), hex(good) if good else '-'), flush=True) if good: ctx = good if kind == 'FN': ra = rd_q(regs.rsp) rec_f = {'kind':'FN','tid':pid, 'rcx':regs.rcx,'rdx':regs.rdx,'r8':regs.r8,'r9':regs.r9, 'ret':ra,'t':round(time.time()-t_start,4)} samples.append(rec_f); hits['FN'] += 1 if hits['FN'] <= 3: print('FN: rcx=%#x rdx=%#x r8=%#x r9=%#x ret=%#x'%( regs.rcx,regs.rdx,regs.r8,regs.r9,ra), flush=True) poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) continue if kind in ('DIV','DC40','EXPVAR'): rec_a = {'kind': kind, 'tid': pid, 't': round(time.time()-t_start, 4), 'rcx': regs.rcx, 'rdx': regs.rdx, 'r8': regs.r8, 'r9': regs.r9} try: for nm, p, cnt in (('a', regs.rcx, 2050), ('b', regs.rdx, 2050), ('c', regs.r8, 2050)): if p > 0x10000: rec_a[nm] = rd_f32(p, cnt) except OSError as e: rec_a['err'] = str(e) samples.append(rec_a) hits[kind] += 1 poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) continue if kind == 'DF0': track_by_tid[pid] = regs.rdx rec = {'kind': kind, 'tid': pid, 'rcx': regs.rcx, 'rdx': regs.rdx, 'r8': regs.r8 & 0xFFFFFFFF, 't': round(time.time()-t_start, 4)} try: if kind == 'EXP': rec['buf'] = rd_f32(regs.rcx, 4098) rec['count'] = rec['r8'] else: rec['fir'] = rd_f32(regs.rcx, 4098) if regs.rdx > 0x10000: rec['track'] = rd_f32(regs.rdx, 2049*2) if ctx: snapshot_slots(rec) if kind == 'DF0': rec['fir_via_ctx'] = rec.get('fir_via_ctx') except OSError as e: rec['err'] = str(e) samples.append(rec) hits[kind] += 1 # снять int3 -> шаг назад -> singlestep -> вернуть int3 -> cont poke(pid, site, (peek(pid, site) & ~0xFF) | obyte) regs.rip = site setregs(pid, regs) pt(PTRACE_SINGLESTEP, pid, 0, 0) os.waitpid(pid, __WALL) poke(pid, site, (peek(pid, site) & ~0xFF) | 0xCC) pt(PTRACE_CONT, pid, 0, 0) if sum(hits.values()) % 10 == 0: print('hits:', hits, flush=True) elif sig in (signal.SIGSTOP, signal.SIGCHLD, signal.SIGWINCH): pt(PTRACE_CONT, pid, 0, 0) else: # посторонний сигнал — доставить pt(PTRACE_CONT, pid, 0, sig) finally: # снять int3 и отсоединиться for addr, (name, obyte) in bps.items(): try: poke(host, addr, (peek(host, addr) & ~0xFF) | obyte) except OSError: pass for tid in list(attached): try: pt(PTRACE_DETACH, tid, 0, 0) except OSError: pass print('сбор завершён:', hits, flush=True) snap_ptrs, snap_arr = {}, {} for nm, off in CTX_SLOTS.items(): p = rd_q(ctx+off) if p > 0x10000: snap_ptrs[nm] = p snap_arr[nm] = rd_f32(p, 4100) with open(os.path.join(outdir, 'chain_samples.pkl'), 'wb') as f: pickle.dump({'samples': samples, 'snap_ptrs': snap_ptrs, 'ctx': ctx}, f) np.savez_compressed(os.path.join(outdir, 'ctx_snap.npz'), **snap_arr) print('saved %d -> %s' % (len(samples), outdir), flush=True) for _ in range(600): if proc.poll() is not None: break time.sleep(0.1) print('reaper_rc=%s wav=%s' % (proc.poll(), os.path.getsize(wav) if wav and os.path.exists(wav) else 'NONE'), flush=True) return 0 if __name__ == '__main__': sys.exit(main())