live-hub-py/scripts/emu_spawn_captcha_test.py

#!/usr/bin/env python3
"""frida 存活期间验证码加载对照实验 (模拟器, spawn+bypass 三件套).

复用 hook_emu_stable.py 的加载顺序 (art_callsite 版 bypass → 立即 resume → patch_guard),
主 hook 只挂轻量 SSL hook (不干扰 UI), 然后用 adb UI 自动化走到登录→验证码,
验证: spawn+bypass 存活时, 极验验证码能否正常加载。
"""
from pathlib import Path
import frida, time, json, subprocess, re, 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")
OUT = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("/tmp/emu_spawn_captcha.json")

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

MAIN_JS = """
'use strict';
send({type:'armed'});
/* 轻量: 只投递存活心跳, 不碰 UI 线程逻辑 */
setInterval(function(){ send({type:'alive', t:Date.now()}); }, 5000);
"""

def ui_dump():
    adb("shell", "uiautomator", "dump", "/data/local/tmp/ui.xml")
    return adb("shell", "cat", "/data/local/tmp/ui.xml").stdout

def find_bounds(xml, text):
    m = re.search(r'<node[^>]*text="' + re.escape(text) + r'"[^>]*bounds="\[(\d+),(\d+)\]\[(\d+),(\d+)\]"', xml)
    if not m: return None
    x1, y1, x2, y2 = map(int, m.groups())
    return (x1 + x2) // 2, (y1 + y2) // 2

def tap(x, y): adb("shell", "input", "tap", str(x), str(y))
def type_text(s): adb("shell", "input", "text", s)

def go_login(force=False):
    xml = ui_dump()
    b = find_bounds(xml, "我的")
    if not b: return False
    tap(*b); time.sleep(2.5)
    xml = ui_dump()
    b = find_bounds(xml, "立即登录")
    if not b: return False
    tap(*b); time.sleep(3)
    xml = ui_dump()
    b = find_bounds(xml, "账号密码登录")
    if b:
        tap(*b); time.sleep(2)
    return True

def fill_submit():
    xml = ui_dump()
    b = find_bounds(xml, "手机号/虎牙号") or find_bounds(xml, "请填写手机号码")
    if not b: return False
    tap(*b); time.sleep(0.4); type_text("13800138000")
    xml = ui_dump()
    b = find_bounds(xml, "密码")
    if b:
        tap(*b); time.sleep(0.4); type_text("test12345678")
    time.sleep(0.6)
    xml = ui_dump()
    b = find_bounds(xml, "立即登录")
    if not b: return False
    tap(*b); time.sleep(2)
    xml = ui_dump()
    b = find_bounds(xml, "同意并继续")
    if b:
        tap(*b); time.sleep(4)
    return True

def captcha_status():
    xml = ui_dump()
    joined = xml
    webview = "android.webkit.WebView" in joined
    slider = any(k in joined for k in ("滑块", "拼图", "安全验证"))
    return {"webview": webview, "slider": slider,
            "focus": adb("shell", "dumpsys window").stdout.count("OakVerifyActivity") > 0}

def main():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    result = {}

    # ===== 阶段1: 无 frida 基线 =====
    print("== 阶段1: 无frida 基线 ==", flush=True)
    adb("shell", "am", "force-stop", PACKAGE); time.sleep(1.2)
    adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")
    time.sleep(8)
    if go_login() and fill_submit():
        time.sleep(5)
        result["baseline_no_frida"] = captcha_status()
        print("  验证码:", result["baseline_no_frida"], flush=True)
    else:
        print("  基线流程失败", flush=True)

    # ===== 阶段2: spawn+bypass 存活时 =====
    print("== 阶段2: spawn+bypass ==", flush=True)
    adb("shell", "am", "force-stop", PACKAGE); time.sleep(1.2)
    pid = d.spawn([PACKAGE])
    print(f"  spawn pid={pid}", flush=True)
    session = d.attach(pid)
    def fast_load(path):
        try:
            s = session.create_script(path.read_text()); s.load(); return True
        except Exception as e:
            print(f"  [load-err] {e}", flush=True); return False
    fast_load(RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js")
    fast_load(RE / "evidence/scripts/mask_frida_maps_only.js")
    d.resume(pid)
    print("  resumed", flush=True)
    try:
        sg = session.create_script((RE/"evidence/scripts/patch_guard_block_termination.js").read_text()); sg.load()
    except Exception as e:
        print(f"  [guard-err] {e}", flush=True)

    alive_events = []
    def on_main(m, dta):
        if m.get('type') == 'send':
            p = m.get('payload') or {}
            if p.get('type') == 'alive':
                alive_events.append(p)
        elif m.get('type') == 'error':
            print("  [JS-ERR]", str(m)[:120], flush=True)
    sc = session.create_script(MAIN_JS)
    sc.on('message', on_main)
    sc.load()

    # 等 UI 起来
    time.sleep(8)
    alive = [p for p in d.enumerate_processes() if p.pid == pid]
    result["spawn_alive_at_8s"] = bool(alive)
    print(f"  8s 存活: {bool(alive)}", flush=True)
    if alive and go_login() and fill_submit():
        time.sleep(5)
        result["captcha_with_frida_alive"] = captcha_status()
        print("  frida存活时验证码:", result["captcha_with_frida_alive"], flush=True)
    alive2 = [p for p in d.enumerate_processes() if p.pid == pid]
    result["spawn_alive_at_end"] = bool(alive2)
    result["alive_heartbeats"] = len(alive_events)
    print(f"  最终存活: {bool(alive2)}, 心跳数: {len(alive_events)}", flush=True)
    try: d.kill(pid)
    except: pass
    OUT.write_text(json.dumps(result, ensure_ascii=False, indent=1))
    print("[*] done ->", OUT, flush=True)

if __name__ == "__main__":
    main()