live-hub-py/scripts/diag_egl_strategy.py

#!/usr/bin/env python3
"""挂起策略梯度实验: 确认 attach+maps遮蔽 是否能绕过 msaoaidsec 且不EGL崩.

E0: spawn→attach(空)→resume           = 复现 session gone(静默死)
E1: spawn→attach→load mask脚本→resume  = attach+maps遮蔽
E2: spawn→attach→load mask+sslhook→resume = 最终抓包形态
E3: 纯spawn→resume(不attach)           = B组对照(应活)

每个 case 跑 2 轮, 每轮观察 30s 或死亡, logcat crash 同步。
"""
from pathlib import Path
import frida, time, json, subprocess, sys, re

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_DIR = Path("/Users/yml/codes/douyu_login_py/evidence/diag_egl_strategy")

SSL_HOOK = r"""
'use strict';
send({type:'armed'});
try{
  var r=new ApiResolver('module');
  r.enumerateMatchesSync('exports:*!SSL_write').forEach(function(m){
    Interceptor.attach(m.address,{onEnter:function(a){send({type:'ssl',len:a[2].toInt32(),t:Date.now()});}});
  });
  send({type:'hooked'});
}catch(e){send({type:'err',e:String(e)});}
"""

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

def detach_cb(reason, detail):
    print(f"    [detached] reason={reason} detail={detail}", flush=True)

def run_case(d, case, rnd):
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.2)
    adb("shell", "logcat", "-c")
    pid = None
    session = None
    t_spawn = time.time()
    try:
        if case in ("E4", "E5"):
            # B组: 纯spawn→立即resume→App起来后再attach
            pid = d.spawn([PACKAGE])
            d.resume(pid)
            print(f"[{case}.r{rnd}] spawn pid={pid} 立即resumed", flush=True)
            time.sleep(1.5 if case == "E4" else 5.0)
            t_att = time.time()
            session = d.attach(pid)
            session.on('detached', detach_cb)
            print(f"[{case}.r{rnd}] attach at +{(time.time()-t_spawn)*1000:.0f}ms (attach耗时{(time.time()-t_att)*1000:.0f}ms)", flush=True)
            suspend_ms = (time.time() - t_spawn) * 1000
        elif case in ("E6", "E7"):
            # 极短挂起: 只加载最少 bypass
            pid = d.spawn([PACKAGE])
            suspend_start = time.time()
            session = d.attach(pid)
            session.on('detached', detach_cb)
            if case == "E6":
                sc = session.create_script((RE/"evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()); sc.load()
            else:
                sc = session.create_script((RE/"evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()); sc.load()
                sc2 = session.create_script((RE/"evidence/scripts/mask_frida_maps_only.js").read_text()); sc2.load()
            suspend_ms = (time.time() - suspend_start) * 1000
            d.resume(pid)
            print(f"[{case}.r{rnd}] spawn pid={pid} 挂起{suspend_ms:.0f}ms(1个art_callsite补丁) resumed", flush=True)
        else:
            pid = d.spawn([PACKAGE])
            suspend_start = time.time()
            if case in ("E1", "E2"):
                session = d.attach(pid)
                session.on('detached', detach_cb)
                if case == "E1":
                    sc = session.create_script((RE/"evidence/scripts/mask_frida_maps_only.js").read_text()); sc.load()
                else:
                    sc = session.create_script((RE/"evidence/scripts/mask_frida_maps_only.js").read_text()); sc.load()
                    sc2 = session.create_script(SSL_HOOK); sc2.load()
            elif case == "E0":
                session = d.attach(pid)
                session.on('detached', detach_cb)
            elif case == "E3":
                pass  # 纯spawn不attach
            suspend_ms = (time.time() - suspend_start) * 1000
            d.resume(pid)
            print(f"[{case}.r{rnd}] spawn pid={pid} 挂起{suspend_ms:.0f}ms resumed", flush=True)
    except Exception as e:
        print(f"[{case}.r{rnd}] setup ERR {e}", flush=True)
        return None

    # 观察 30s
    t0 = time.time()
    died_at = None
    pid_changed = None
    while time.time() - t0 < 30:
        time.sleep(2)
        mp = main_pid()
        if mp is None:
            died_at = time.time() - t0
            break
        if mp != pid:
            pid_changed = (mp, time.time() - t0)
            break
    cr = adb("shell", "logcat", "-d", "-b", "crash", "-t", "20").stdout
    egl = "EGL" in cr or "libEGL" in cr
    crash_banner = bool(re.search(r"Fatal|signal \d|Abort", cr))
    try:
        if session: session.detach()
    except Exception: pass
    try: d.kill(pid)
    except Exception: pass
    time.sleep(1)
    result = {
        "case": case, "round": rnd, "pid": pid, "suspend_ms": round(suspend_ms, 1),
        "died_at_s": round(died_at, 1) if died_at else None,
        "pid_changed": pid_changed, "egl_crash": egl,
        "crash_banner": crash_banner,
        "crash_tail": cr[:300],
    }
    print(f"    → {result}", flush=True)
    return result

def main():
    cases = sys.argv[1:] or ["E0", "E1", "E2", "E3", "E4", "E5", "E6", "E7"]
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    d = frida.get_device_manager().add_remote_device(REMOTE)
    results = []
    for c in cases:
        for rnd in (1, 2):
            r = run_case(d, c, rnd)
            if r: results.append(r)
    out = OUT_DIR / "egl_strategy.json"
    out.write_text(json.dumps(results, ensure_ascii=False, indent=1))
    print(f"[*] {len(results)} 结果 -> {out}", flush=True)

if __name__ == "__main__":
    main()