live-hub-py/scripts/hook_emu_dfp.py

#!/usr/bin/env python3
"""模拟器版 dfpReport 捕获: 验证全新模拟器是否为"新设备身份"。

连接 127.0.0.1:31878 (模拟器 frida 15.2.2), spawn 方式。
只 hook SSL_write, 抓 dfpReport 请求 + 对应 606B 响应(内含 actionV/真或新 hdid)。
若 attach/反调试, 用 bypass 三件套 (可选开关)。
"""
from __future__ import annotations
import json, time
from pathlib import Path
import frida

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_dfp_" + time.strftime("%H%M%S") + ".json")
REMOTE = "127.0.0.1:31878"
USE_BYPASS = True  # 若 spawn 后闪退, 设为 False 再试

JS = r"""
'use strict';
Process.setExceptionHandler(function(d){ return true; });
send({type:'armed'});
var t0 = Date.now();
function hexb(p, n){ try{ return Array.from(new Uint8Array(p.readByteArray(n))).map(function(b){return ('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)return;
      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()-t0,hex:hexb(a[1],len)});
      }
    }});
  });
  send({type:'sslwrite_hooked'});
}catch(e){ send({type:'sslwrite_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 n2=ret.toInt32();
        if(n2<=0||n2>3200)return;
        send({type:'resp',len:n2,hex:hexb(this.buf,n2),t:Date.now()-t0});
      }});
  });
  send({type:'sslread_hooked'});
}catch(e){ send({type:'sslread_err',e:String(e)}); }
"""


def load(session, js, n=5, wait=2):
    for i in range(n):
        try:
            s = session.create_script(js); s.load(); return s
        except Exception as e:
            print(f"[retry {i}] {e}", flush=True); time.sleep(wait)
    return None


def main():
    d = frida.get_device_manager().add_remote_device(REMOTE)
    from subprocess import run
    try:
        pid = d.spawn([PACKAGE])
        print(f"[*] spawn {PACKAGE} pid={pid}", flush=True)
    except Exception as e:
        # fallback attach
        pid = [p for p in d.enumerate_processes() if 'kiwi' in p.name or 'duowan' in p.name]
        pid = pid[0].pid if pid else None
        print(f"[*] spawn failed, attach pid={pid}", flush=True)
    session = d.attach(pid)
    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 == 'sslwrite_hooked':
            print("[*] SSL_write hooked", flush=True)
        elif t == 'sslread_hooked':
            print("[*] SSL_read hooked", flush=True)
        elif t == 'req':
            print(f"[{p.get('t')}ms] [{p.get('n')}] {p.get('cls')} len={p.get('len')}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events))
        elif 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')}ms] [resp] len={p.get('len')}{mark}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events))
        elif t in ('sslwrite_err','sslread_err'):
            print(f"[*] {t}: {p.get('e')}", flush=True)

    if USE_BYPASS:
        print("[*] loading bypass", flush=True)
        load(session, (RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text(), wait=1)
        load(session, (RE / "evidence/scripts/mask_frida_maps_only.js").read_text(), wait=1)
        d.resume(pid)
        time.sleep(9)
        load(session, (RE / "evidence/scripts/patch_guard_block_termination.js").read_text(), wait=1)
        print("[*] loading 主 JS", flush=True)
        script = load(session, JS)
    else:
        d.resume(pid)
        time.sleep(6)
        script = load(session, JS)

    if script:
        script.on('message', on_message)
        print(f"[*] running. OUT={OUT}", flush=True)
    try:
        while True:
            time.sleep(4)
    except KeyboardInterrupt:
        pass
    print(f"[*] done -> {OUT}", flush=True)


if __name__ == "__main__":
    main()