live-hub-py/tools/frida/run_capture.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""虎牙 OTP 定向捕获运行器。

流程:
  1) adb 启动手机端 frida-server (re.frida.server/fs152, 监听 127.0.0.1:31878)
  2) 远程连接 127.0.0.1:31878, spawn com.duowan.kiwi
  3) 依次注入: frida_bypass.js (maps 伪装) -> frida_java_exit.js (Java kill 拦截)
              -> hook_otp_capture.js (OTP 六元组 + setSafeDeviceId 捕获)
  4) resume, 收集 send() 事件 -> JSONL 落盘 + 实时打印
用法:
  python3 run_capture.py                 # 默认输出 ./capture_<ts>.jsonl
  CAP_OUT=/tmp/otp.jsonl python3 run_capture.py
"""
import json
import os
import subprocess
import sys
import time
from datetime import datetime

HERE = os.path.dirname(os.path.abspath(__file__))
OUT_DIR = os.environ.get("CAP_OUT_DIR", HERE)
PKG = "com.duowan.kiwi"
SERVER_PATH = "/data/local/tmp/re.frida.server/fs152"
SERVER_PORT = "31878"
BYpass_RE = "/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/evidence/scripts/bypass_msaoaid_maps_art_callsite.js"
JAVA_EXIT_RE = ""   # 真机通道: 单一 bypass 脚本最稳, 不加载 java-exit
HOOK_LOCAL = os.path.join(HERE, "hook_otp_capture.js")


def adb(*args):
    return subprocess.run(["adb"] + list(args), capture_output=True, text=True)


def main():
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    out_path = os.path.join(OUT_DIR, f"capture_{ts}.jsonl")
    fout = open(out_path, "w", encoding="utf-8")
    print(f"[runner] output -> {out_path}", flush=True)

    # 1) 启动手机端 frida-server (root, 后台)
    print("[runner] starting frida-server on device ...", flush=True)
    adb("shell", "su", "-c",
        f"nohup {SERVER_PATH} -l 127.0.0.1:{SERVER_PORT} >/data/local/tmp/re.frida.server/fs152.log 2>&1 &")
    time.sleep(2.0)

    import frida

    dev = None
    for i in range(6):
        try:
            dev = frida.get_device_manager().add_remote_device(f"127.0.0.1:{SERVER_PORT}")
            apps = dev.enumerate_processes()
            print(f"[runner] frida-server OK ({len(apps)} procs)", flush=True)
            break
        except Exception as e:
            print(f"[runner] wait frida-server [{i}] {str(e)[:80]}", flush=True)
            time.sleep(2.0)
    if dev is None:
        print("[runner] FATAL: frida-server unreachable", flush=True)
        sys.exit(1)

    # 2) 确保旧进程关闭, 然后 spawn
    adb("shell", "am", "force-stop", PKG)
    time.sleep(1.0)
    try:
        pid = dev.spawn([PKG])
        print(f"[runner] spawned pid={pid}", flush=True)
    except Exception as e:
        print(f"[runner] spawn failed: {e}", flush=True)
        sys.exit(1)

    session = dev.attach(pid)

    def on_message(msg, data):
        if msg.get("type") == "send":
            payload = msg.get("payload")
            if payload is not None:
                line = json.dumps(payload, ensure_ascii=False)
                fout.write(line + "\n")
                fout.flush()
                ev = payload.get("event", "")
                if ev in ("crypto-otp", "setdi", "gethdid", "hook-missing", "hook-installed"):
                    print(f"[cap] {line[:400]}", flush=True)
        elif msg.get("type") == "error":
            print(f"[cap-err] {msg.get('stack', msg)}", flush=True)
        elif msg.get("type") == "device":
            print(f"[cap-dev] {msg.get('payload')}", flush=True)

    scripts = []
    # 阶段1: 只注入 bypass (STATUS.md 验证的 90s 稳定配方, 避免启动期注入竞争)
    with open(BYpass_RE, "r", encoding="utf-8") as f:
        sc = session.create_script(f.read())
    sc.on("message", on_message)
    sc.load()
    scripts.append(sc)
    print("[runner] loaded bypass (phase-1)", flush=True)

    dev.resume(pid)
    print("[runner] RESUMED (phase-1). 等待 app 启动 ...", flush=True)

    # 阶段2: app 启动后延迟注入 otp 钩子 (避开 msaoaid 启动扫描 + EGL 竞争)
    def inject_phase2():
        time.sleep(8)
        try:
            with open(HOOK_LOCAL, "r", encoding="utf-8") as f:
                src = f.read()
            sc2 = session.create_script(src)
            sc2.on("message", on_message)
            sc2.load()
            scripts.append(sc2)
            print("[runner] loaded otp-hook (phase-2)", flush=True)
        except Exception as e:
            print(f"[runner] phase-2 inject failed: {e}", flush=True)
    import threading
    threading.Thread(target=inject_phase2, daemon=True).start()

    # 稳定窗口 ~120s 后自动收工
    def auto_stop():
        time.sleep(120)
        print("[runner] auto-stop", flush=True)
        os._exit(0)
    threading.Thread(target=auto_stop, daemon=True).start()

    try:
        while True:
            time.sleep(1.0)
    except KeyboardInterrupt:
        pass

    fout.close()
    print(f"[runner] done -> {out_path}", flush=True)


if __name__ == "__main__":
    main()