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

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""虎牙 OTP / setDeviceInfo 定向捕获运行器 (自愈多轮版)。

稳定性设计 (2026-08-29 第二轮):
  - 两阶段注入: spawn 挂起只注入 bypass_msaoaid_maps_art_callsite.js (真机唯一稳定配方)
    -> resume -> 等 app 启动后 (+6s) 再注入 OTP/设备信息钩子 (避开 msaoaid 启动扫描竞争)
  - 自愈: 每轮 app 被杀/窗口结束自动 force-stop 重 spawn, 最多 N 轮, 事件汇总到同一 JSONL
  - 单轮窗口 150s (真机稳定绿区 ~90s+, 加上两阶段补偿)

用法:
  python3 run_capture.py
  CAP_MAX_ATTEMPTS=12 CAP_WINDOW=160 python3 run_capture.py
"""
import json
import os
import subprocess
import sys
import time
import threading
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/fs152"
SERVER_PORT = "31878"
BYpass_RE = "/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/evidence/scripts/bypass_msaoaid_maps_art_callsite.js"
HOOK_LOCAL = os.environ.get("CAP_HOOK")

MAX_ATTEMPTS = int(os.environ.get("CAP_MAX_ATTEMPTS", "8"))
WINDOW_SEC = int(os.environ.get("CAP_WINDOW", "150"))
HOOK_DELAY = float(os.environ.get("CAP_HOOK_DELAY", "6"))

_count = {"events": 0}


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


def clear_turing_state():
    """清 turing 状态 -> 自动重注册触发报告 (文档配方 §11.49)."""
    try:
        subprocess.run(["adb","shell","am","force-stop","com.duowan.kiwi"], capture_output=True, text=True)
        time.sleep(1)
        for path in ["app_turingdfp","app_turingfd","resinfo","app_turingdmp","cocos","turing_files"]:
            r = subprocess.run(["adb","shell","run-as","com.duowan.kiwi","rm","-rf",f"/data/data/com.duowan.kiwi/{path}"], capture_output=True, text=True)
            if r.returncode == 0:
                print(f"[runner] cleared {path}", flush=True)
        print("[runner] turing state cleared -> expect auto re-register report", flush=True)
    except Exception as e:
        print(f"[runner] clear-EXC {e}", flush=True)

def run_once(fout, out_path):
    """单次 spawn 捕获一轮; 返回该轮事件数 (0 = 该轮无捕获)."""
    import frida

    # 1) frida-server 存活
    adb("shell", "su", "-c",
        f"if ! pgrep -f {SERVER_PATH}; then nohup {SERVER_PATH} -l 127.0.0.1:{SERVER_PORT} >/data/local/tmp/fs152.log 2>&1 & fi")
    time.sleep(1.5)

    dev = None
    for i in range(6):
        try:
            dev = frida.get_device_manager().add_remote_device(f"127.0.0.1:{SERVER_PORT}")
            n = len(dev.enumerate_processes())
            print(f"[runner] frida-server OK ({n} 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)
        return 0

    adb("shell", "am", "force-stop", PKG)
    time.sleep(1.0)
    try:
        clear_turing_state()
        pid = dev.spawn([PKG])
        print(f"[runner] spawned pid={pid}", flush=True)
    except Exception as e:
        print(f"[runner] spawn failed: {e}", flush=True)
        return 0

    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()
                _count["events"] += 1
                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)[:300]}", flush=True)
        elif msg.get("type") == "device":
            print(f"[cap-dev] {msg.get('payload')}", flush=True)

    # 阶段1: 仅注入 bypass
    try:
        with open(BYpass_RE, "r", encoding="utf-8") as f:
            sc = session.create_script(f.read())
        sc.on("message", on_message)
        sc.load()
        print("[runner] loaded bypass (phase-1)", flush=True)
    except Exception as e:
        print(f"[runner] bypass load failed: {e}", flush=True)
        session.detach()
        return 0

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

    # 阶段2: 延迟注入 otp 钩子
    def inject_phase2():
        time.sleep(HOOK_DELAY)
        try:
            with open(HOOK_LOCAL, "r", encoding="utf-8") as f:
                sc2 = session.create_script(f.read())
            sc2.on("message", on_message)
            sc2.load()
            print("[runner] loaded otp-hook (phase-2)", flush=True)
        except Exception as e:
            print(f"[runner] phase-2 inject failed: {e[:200]}", flush=True)
    threading.Thread(target=inject_phase2, daemon=True).start()

    # 窗口等待 + 进程死亡检测 (自愈: 死了立刻重试)
    start = time.time()
    while time.time() - start < WINDOW_SEC:
        time.sleep(1.0)
        out = adb("shell", "su", "-c", f"kill -0 {pid} 2>/dev/null && echo alive || echo dead")
        if "dead" in out.stdout:
            print(f"[runner] process {pid} died at {int(time.time()-start)}s", flush=True)
            break
    try:
        session.detach()
    except Exception:
        pass
    adb("shell", "am", "force-stop", PKG)
    print(f"[runner] round done: {out_path}", flush=True)
    return _count["events"]


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}  attempts<={MAX_ATTEMPTS} window={WINDOW_SEC}s", flush=True)

    attempts = 0
    while attempts < MAX_ATTEMPTS:
        attempts += 1
        before = _count["events"]
        try:
            run_once(fout, out_path)
        except Exception as e:
            print(f"[runner] round {attempts} exception: {str(e)[:150]}", flush=True)
        if _count["events"] > before:
            print(f"[runner] attempts={attempts} events={_count['events']} — stopping (capture got data)", flush=True)
            break
        time.sleep(3)

    fout.close()
    print(f"[runner] FINAL attempts={attempts} events={_count['events']} -> {out_path}", flush=True)


if __name__ == "__main__":
    main()