live-hub-py/scripts/hook_emu_spawn_bypass.py

#!/usr/bin/env python3
"""模拟器 spawn + frida 绕过 版 dfpReport 捕获。

时序(模拟器需快速 resume 防 EGL 崩, 但反调试需在 resume 前 hook):
  spawn(挂起) → 快速 load bypass_msaoaid + mask_frida (~1s内)
  → resume → patch_guard → 主 hook (抓 dfpReport+响应 actionV)

bypass 脚本用动态定位(Process.findModuleByName), 偏移为 lib 内偏移,
同一 apk(13.4.22)下与真机一致, 直接复用。
"""
from __future__ import annotations
import frida, time, json, re, subprocess
from pathlib import Path

REMOTE = "127.0.0.1:31878"
PACKAGE = "com.duowan.kiwi"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
OUT = Path("/Users/yml/codes/douyu_login_py/evidence/emu_spawn_bypass_dfp.json")

JS = r"""
'use strict';
send({type:'armed'});
function hexb(p,n){try{return Array.from(new Uint8Array(p.readByteArray(n))).map(b=>('0'+b.toString(16)).slice(-2)).join('');}catch(e){return '';}}
function head(p,n){try{return p.readCString(n);}catch(e){return '';}}
var n=0;
try{
  var r=new ApiResolver('module');
  r.enumerateMatchesSync('exports:*!SSL_write').forEach(function(m){
    Interceptor.attach(m.address,{onEnter:function(a){
      var len=a[2].toInt32();
      if(len<50||len>50000)return;
      var h=head(a[1],Math.min(len,1500));
      if(h.indexOf('dfpReport')>=0||h.indexOf('dckey')>=0){
        n++; send({cls:h.indexOf('dfpReport')>=0?'dfpReport':'dckey',n:n,len:len,t:Date.now(),hex:hexb(a[1],len)});
      }
    }});
  });
  send({type:'hooked'});
}catch(e){send({type:'err',e:String(e)});}
try{
  var r2=new ApiResolver('module');
  r2.enumerateMatchesSync('exports:*!SSL_read').forEach(function(m){
    Interceptor.attach(m.address,{onEnter:function(a){this.buf=a[1];},
      onLeave:function(ret){
        var nn=ret.toInt32();
        if(nn<=0||nn>3200)return;
        send({type:'resp',len:nn,hex:hexb(this.buf,nn),t:Date.now()});
      }});
  });
  send({type:'readhooked'});
}catch(e){send({type:'readerr',e:String(e)});}
"""


def load(session, js, wait=0.2):
    try:
        s = session.create_script(js); s.load(); return s
    except Exception as e:
        print("[load-err]", str(e)[:150], flush=True); return None


def main():
    subprocess.run(["adb", "-s", "127.0.0.1:5555", "shell", "am", "force-stop", PACKAGE], capture_output=True)
    time.sleep(1.5)
    d = frida.get_device_manager().add_remote_device(REMOTE)
    pid = d.spawn([PACKAGE])
    print(f"[*] spawn pid={pid}", flush=True)
    session = d.attach(pid)

    # 1) 挂起时快速加载两个 bypass
    print("[*] load bypass_msaoaid (suspended)", flush=True)
    load(session, (RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text())
    print("[*] load mask_frida (suspended)", flush=True)
    load(session, (RE / "evidence/scripts/mask_frida_maps_only.js").read_text())

    # 2) resume (防 EGL 崩), 快速
    d.resume(pid)
    print("[*] resumed", flush=True)

    # 3) patch_guard + 主 hook
    load(session, (RE / "evidence/scripts/patch_guard_block_termination.js").read_text(), wait=0.1)
    print("[*] load 主 hook JS", flush=True)
    events = []

    def on_message(m, data):
        if m.get('type') == 'error':
            print("[JS-ERR]", str(m)[:200], flush=True); return
        p = m.get('payload') or {}
        t = p.get('type')
        if t == 'armed': print("[*] armed", flush=True)
        elif t == 'hooked': print("[*] SSL_write hooked", flush=True)
        elif t == 'readhooked': print("[*] SSL_read hooked", flush=True)
        elif t in ('err','readerr'): print("[*]", t, p.get('e'), flush=True)
        elif t == 'event': print("[bypass]", p.get('event'), flush=True)
        elif 'cls' in p:
            print(f"[{p.get('t')}] {p.get('cls')} len={p.get('len')}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events))
        elif 'resp' in t or t == 'resp':
            import re
            av = re.search(rb'actionV\(([0-9a-f]{40})', bytes.fromhex(p.get('hex') or ''))
            mark = f" actionV={av.group(1).decode()}" if av else ""
            print(f"[{p.get('t')}] RESP len={p.get('len')}{mark}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events))

    sc = load(session, JS)
    if sc:
        sc.on('message', on_message)
    print(f"[*] running (OUT={OUT})", flush=True)
    t0 = time.time()
    try:
        while time.time() - t0 < 70:
            time.sleep(4)
    except KeyboardInterrupt:
        pass
    print(f"[*] done, {len(events)} events", flush=True)


if __name__ == "__main__":
    main()