live-hub-py/scripts/emu_msaoaid_trace.py

#!/usr/bin/env python3
"""msaoaidsec 检测面精确归因 v2 (修正误判版).

上次 attach_trace_survival 用 main_pid() is None 判断死亡 —— 错!
KeepAlive 重启会换 pid, main_pid() 返回新 pid, 误报"存活"。
本版: 死亡判定 = pid 变化 或 无主进程 (均算死), 并采集:
  1. trace_msaoaid_thread_strstr_plt.js — 线程名检测器 strstr 命中 (needle=?)
  2. trace_msaoaid_exit_sources.js (API兼容修复) — _exit PLT slot 观察, backtrace 帧
每组 spawn+attach (挂起中加载脚本, 立即resume), 观察 12s。
"""
from pathlib import Path
import frida, time, json, subprocess, sys

REMOTE = "127.0.0.1:31878"
ADB = ["adb", "-s", "127.0.0.1:5555"]
PACKAGE = "com.duowan.kiwi"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")

def sh(*a): return subprocess.run(a, capture_output=True, text=True)
def adb(*a): return sh(*ADB, *a)

def main_pid():
    r = adb("shell", "pidof", PACKAGE)
    for p in r.stdout.strip().split():
        rr = adb("shell", "cat", f"/proc/{p}/cmdline")
        if rr.stdout.strip("\x00").strip() == PACKAGE:
            return int(p)
    return None

# frida 15 兼容修复: getGlobalExportByName -> findExportByName
def fix_script(text):
    return text.replace("Module.getGlobalExportByName(", "Module.findExportByName(null, ")

def run_case(d, rnd, scripts):
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.2)
    adb("shell", "logcat", "-c")
    pid = d.spawn([PACKAGE])
    session = d.attach(pid)
    t0 = time.time()
    events = []
    def on(m, dta):
        if m.get('type') != 'send':
            if m.get('type') == 'error':
                print(f"  [JS-ERR] {str(m)[:100]}", flush=True)
            return
        p = m.get('payload') or {}
        ev = p.get('event', '?')
        line = f"[r{rnd} +{time.time()-t0:6.2f}s] {ev}"
        if ev == 'msaoaid-thread-strstr':
            line += f" needle={p.get('needle')} hay={p.get('haystack','')[:30]!r}"
        elif ev == 'msaoaid-thread-strstr-return':
            line += f" needle={p.get('needle')} hit={p.get('returnedNonNull')} ret={p.get('returnPointer')}"
        elif ev == 'msaoaid-exit-source':
            line += f" status={p.get('status')} tid={p.get('tid')} frames={p.get('frames')}"
        elif ev in ('msaoaid-exit-got-thread-exit', 'msaoaid-exit-got-thread-exit-installed'):
            line += f" data={p.get('status') or p.get('slot')}"
        print(line, flush=True)
        events.append(p)
    for name in scripts:
        try:
            text = (RE / "evidence/scripts" / name).read_text()
            sc = session.create_script(fix_script(text))
            sc.on('message', on)
            sc.load()
            print(f"[r{rnd}] loaded {name} +{(time.time()-t0)*1000:.0f}ms", flush=True)
        except Exception as e:
            print(f"[r{rnd}] load-err {name}: {e}", flush=True)
    d.resume(pid)
    print(f"[r{rnd}] resumed +{(time.time()-t0)*1000:.0f}ms", flush=True)
    # 精确死亡判定: pid变化或无主进程
    died = None
    change = None
    while time.time() - t0 < 12:
        time.sleep(1.5)
        mp = main_pid()
        if mp is None:
            died = time.time() - t0
            print(f"[r{rnd}] DEAD(no main) at +{died:.1f}s", flush=True)
            break
        if mp != pid:
            change = (mp, round(time.time() - t0, 1))
            print(f"[r{rnd}] PID CHANGED {pid}->{mp} at +{change[1]}s (主进程死/重启)", flush=True)
            break
    if died is None and change is None:
        print(f"[r{rnd}] 存活满12s (pid={pid} 未变)", flush=True)
    cr = adb("shell", "logcat", "-d", "-b", "crash", "-t", "6").stdout
    egl = "EGL" in cr
    print(f"[r{rnd}] EGL崩={egl}", flush=True)
    try: session.detach()
    except Exception: pass
    try: d.kill(pid)
    except Exception: pass
    time.sleep(1)
    return {"round": rnd, "scripts": scripts, "died_at": died, "pid_change": change, "egl": egl, "events": events}

def main():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    # 组1: 只读线程名探针 (不patch不mask)
    r1 = run_case(d, 1, ["trace_msaoaid_thread_strstr_plt.js"])
    # 组2: 只读线程名探针 + exit源观察
    r2 = run_case(d, 2, ["trace_msaoaid_thread_strstr_plt.js", "trace_msaoaid_exit_sources.js"])
    # 组3: 只读 exit源观察 (无线程名探针, 减少干扰)
    r3 = run_case(d, 3, ["trace_msaoaid_exit_sources.js"])
    out = Path("/tmp/msaoaid_precise.json")
    out.write_text(json.dumps([r1, r2, r3], ensure_ascii=False, indent=1))
    print(f"[*] -> {out}", flush=True)

if __name__ == "__main__":
    main()