live-hub-py/scripts/emu_attach_captcha_test.py

#!/usr/bin/env python3
"""模拟器 attach+三件套 对照实验:
1) 干净启动 App → 自动走到账号密码登录页 → 填表 → 登录 → 观察验证码是否加载(基线)
2) attach frida 三件套(bypass_msaoaid_maps_skip_cleanup + mask_frida_maps_only + patch_guard)
   → 观察存活 → 重新走登录 → 观察验证码是否加载(对照)

用 RE 仓库 .venv 环境与 RE 仓库 evidence/scripts 三件套。"""
from pathlib import Path
import frida, time, subprocess, json, 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_attach_captcha.json")
TRIO = [
    (RE / "evidence/scripts/bypass_msaoaid_maps_skip_cleanup.js", "bypass_main"),
    (RE / "evidence/scripts/mask_frida_maps_only.js", "mask_frida"),
    (RE / "evidence/scripts/patch_guard_block_termination.js", "patch_guard"),
]

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

def adb(*args):
    return sh(*ADB, *args)

def force_stop():
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)

def launch():
    adb("shell", "monkey", "-p", PACKAGE, "-c", "android.intent.category.LAUNCHER", "1")
    time.sleep(6)

def wait_pid(timeout=20):
    for _ in range(timeout):
        r = adb("shell", "pidof", PACKAGE)
        if r.stdout.strip():
            return int(r.stdout.strip().split()[0])
        time.sleep(1)
    return None

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

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

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

def input_text(s):
    adb("shell", "input", "text", s)

def go_login_page():
    """从首页走到账号密码登录页. 返回 True 若成功."""
    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 not b:
        # 已经是验证码登录页, 尝试切换
        print("  [!] 未找到账号密码登录tab", flush=True)
        return False
    tap(*b); time.sleep(2)
    return True

def fill_and_submit():
    xml = ui_dump()
    b = find_bounds(xml, "手机号/虎牙号")
    if not b:
        b = find_bounds(xml, "请填写手机号码")
        if not b:
            return False
    tap(*b); time.sleep(0.4); input_text("13800138000")
    xml = ui_dump()
    b = find_bounds(xml, "密码")
    if b:
        tap(*b); time.sleep(0.4); input_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()
    nodes = xml.replace("></", ">\n</")
    texts = [m for m in nodes.splitlines() if 'text="' in m]
    joined = " | ".join(t for t in texts if 'text="' in t)
    has_slider = "滑块" in joined or "拼图" in joined or "安全验证" in joined
    has_webview = "android.webkit.WebView" in joined
    # 统计关键文本
    keys = [k for k in ("安全验证", "滑块", "拼图", "验证码", "网络异常", "加载失败", "请稍后") if k in joined]
    return {"ok": has_slider and has_webview, "webview": has_webview, "slider": has_slider,
            "keys": keys, "texts": [l.strip() for l in nodes.splitlines() if 'text="' in l and l.strip().startswith('<node')][:12]}

def attach_trio(d, pid):
    session = d.attach(pid)
    print(f"[*] attach pid={pid}", flush=True)
    loaded = []
    for js_path, name in TRIO:
        try:
            sc = session.create_script(js_path.read_text())
            sc.on('message', lambda m, dd, n=name: print(f"  [{n}] {m.get('payload') if m.get('type')=='send' else m}", flush=True))
            sc.load()
            loaded.append(name)
            time.sleep(0.15)
        except Exception as e:
            print(f"  [!] {name} 加载失败: {e}", flush=True)
    print(f"[*] 三件套已加载: {loaded}", flush=True)
    return session

def main():
    results = {"baseline": None, "with_frida": None, "captcha_baseline": None, "captcha_with_frida": None}
    d = frida.get_device_manager().add_remote_device(REMOTE)

    # ---- 基线: 无 frida 走登录 ----
    print("===== 基线: 无 frida =====", flush=True)
    force_stop(); launch()
    pid = wait_pid()
    results["baseline"] = pid
    print(f"[*] app pid={pid}", flush=True)
    if pid and go_login_page() and fill_and_submit():
        time.sleep(5)
        results["captcha_baseline"] = captcha_status()
        print(f"[*] 基线验证码: {json.dumps(results['captcha_baseline'], ensure_ascii=False)}", flush=True)

    # ---- 对照: attach 三件套 ----
    print("===== 对照: attach + 三件套 =====", flush=True)
    pid = wait_pid()
    if not pid:
        print("[!] app 未运行", flush=True)
        return
    session = attach_trio(d, pid)
    results["with_frida"] = pid
    # 观察存活
    alive = True
    t0 = time.time()
    while time.time() - t0 < 40:
        time.sleep(2)
        try:
            cur = [p for p in d.enumerate_processes() if p.pid == pid]
        except Exception:
            cur = []
        if not cur:
            alive = False
            print(f"[!] attach 后 App 死亡于 +{time.time()-t0:.0f}s", flush=True)
            session = None
            break
    results["frida_survival_s"] = int(time.time() - t0) if alive else int(time.time() - t0)
    print(f"[*] attach 后存活: {'YES' if alive else 'NO'} ({results['frida_survival_s']}s)", flush=True)

    if alive:
        # 回到登录页重新走
        if go_login_page() and fill_and_submit():
            time.sleep(5)
            results["captcha_with_frida"] = captcha_status()
            print(f"[*] frida 下验证码: {json.dumps(results['captcha_with_frida'], ensure_ascii=False)}", flush=True)
    else:
        results["captcha_with_frida"] = "app died"

    OUT.write_text(json.dumps(results, ensure_ascii=False, indent=1))
    print(f"[*] 结果写入 {OUT}", flush=True)
    try:
        if session: session.detach()
    except Exception:
        pass

if __name__ == "__main__":
    main()