live-hub-py/scripts/phone_stable_capture.py

#!/usr/bin/env python3
"""真机稳定抓帧: spawn + art_callsite 补丁 + 长时 hook.

模拟器结论: spawn+脚本必 EGL 崩(约4s), 只能抢 stale 窗口。
真机 M2102J2SC 实测: spawn 挂起加载 art_callsite 补丁后 12s+ 稳定存活
(无 detached, 无 EGL 崩) —— 本脚本验证两点:
  1) 更长存活 (默认 90s), 主进程不被换/不闪退 (watch 逻辑)
  2) hook 能力真实可用: 抓 libudbauthunify.so 的 setSafeDeviceId/createWupDeviceInfo
     (32hex hdid) + SSL_write (hypasswordLogin 帧) + 内存扫描 32hex

用法:
  /Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/.venv/bin/python \
      scripts/phone_stable_capture.py [duration_s] [serial] [remote]
输出:
  evidence/diag_phone/phone_stable_capture.json  (事件 + 存活时间线)
"""
from __future__ import annotations

import json
import re
import subprocess
import sys
import time
from pathlib import Path

import frida

SERIAL = sys.argv[2] if len(sys.argv) > 2 else "5dd8c93f"
REMOTE = sys.argv[3] if len(sys.argv) > 3 else "127.0.0.1:31878"
DURATION = float(sys.argv[1]) if len(sys.argv) > 1 and sys.argv[1].replace(".", "").isdigit() else 90.0
PACKAGE = "com.duowan.kiwi"
REPO = Path("/Users/yml/codes/douyu_login_py")
OUT = REPO / "evidence" / "diag_phone" / "phone_stable_capture.json"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
ART_CALLSITE = RE / "evidence" / "scripts" / "bypass_msaoaid_maps_art_callsite.js"

MAIN_JS = r"""
'use strict';
function readStdString(p){
  if (p.isNull()) return null;
  try{
    var first = p.readU8();
    if ((first & 1) === 0) {
      var len = first >> 1;
      return (len>0 && len<256) ? p.add(1).readUtf8String(len) : '';
    } else {
      var data = p.readPointer();
      var len = p.add(8).readU64();
      return (len>0 && len<1024) ? data.readUtf8String(len) : null;
    }
  }catch(e){ return null; }
}
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 '';}}
function emit(tag,val){ send({type:tag, tag:tag, val:''+(val||''), ts:Date.now()}); }
var hooked=false, scanned=false;

function hookUdb(){
  if (hooked) return;
  var md = Process.findModuleByName('libudbauthunify.so');
  if (!md) return;
  emit('mod-base', ''+md.base);
  try{
    Interceptor.attach(md.base.add(0x26A2E0), {
      onEnter:function(a){ this.sd=readStdString(a[1]); this.hd=readStdString(a[2]); },
      onLeave:function(){
        var hd=(this.hd||''); emit('SETSD', hd+' | sd='+(this.sd||'').slice(0,24));
        if(/^[0-9a-f]{32}$/.test(hd)) emit('FOUND_HDID_32', hd);
      }
    });
  }catch(e){ emit('hook-err','setSd '+e); }
  try{
    Interceptor.attach(md.base.add(0x2746A0), {
      onLeave:function(ret){
        try{ var s=readStdString(ret.add(152)); if(s){ emit('WUPDEV_HDID',''+s); if(/^[0-9a-f]{32}$/.test(s)) emit('FOUND_HDID_32',''+s); } }catch(e){}
      }
    });
  }catch(e){ emit('hook-err','wupdev '+e); }
  emit('ready','all hooked');
  hooked = true;
}

function hookSSL(){
  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,600));
        if(h.indexOf('hypasswordLogin')>=0){
          emit('WUP_LOGIN_FULL', len+'B '+hexb(a[1],len));
        }
      }});
    });
    send({type:'sslwrite-hooked', tag:'ssl'});
  }catch(e){}
}

function scanMem(){
  var found = {};
  try{
    var ranges = Process.enumerateRanges({protection:'r--', coalesce:true});
    var total = ranges.length, scannedB = 0;
    ranges.forEach(function(r, ri){
      try{
        var size = r.size;
        if (size <= 0) return;
        var step = 256*1024;
        var off = 0;
        while (off < size){
          var n = Math.min(step, size - off);
          var buf = Memory.readByteArray(r.base.add(off), n);
          if (buf){
            var sbuf = new Uint8Array(buf);
            var s = '';
            for (var i=0;i<sbuf.length;i++) s += String.fromCharCode(sbuf[i]);
            var re = /[0-9a-f]{32,40}/gi;
            var m;
            while ((m = re.exec(s)) !== null){
              var v = m[0].toLowerCase();
              if (!/^[0-9a-f]{32}$/.test(v)) continue;
              found[v] = (found[v]||0) + 1;
            }
            scannedB += n;
          }
          off += step;
        }
      }catch(e){}
    });
  }catch(e){ emit('scan-err',''+e); }
  var keys = Object.keys(found);
  emit('scan-result', 'ranges='+total+' scannedB='+scannedB+' found32+='+keys.length);
  keys.sort().forEach(function(k, i){
    if (i < 60) emit('H32CAND', k+' x'+found[k]);
  });
  scanned = true;
}

setInterval(hookUdb, 100);
setInterval(hookSSL, 1000);
var scanLaunched = false;
setInterval(function(){
  if (!scanLaunched && hooked && !scanned){
    scanLaunched = true;
    setTimeout(scanMem, 2500);
  }
}, 500);
"""


def adb(*a):
    return subprocess.run(["adb", "-s", SERIAL, *a], capture_output=True, text=True)


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 main():
    OUT.parent.mkdir(parents=True, exist_ok=True)
    adb("shell", "am", "force-stop", PACKAGE)
    time.sleep(1.5)
    adb("shell", "logcat", "-c")

    d = frida.get_device_manager().add_remote_device(REMOTE)
    pid = d.spawn([PACKAGE])
    print(f"[*] spawn pid={pid}", flush=True)
    session = d.attach(pid)
    det = []
    session.on("detached", lambda r, dd: det.append({"reason": r, "detail": str(dd)}) or print(f"[*] detached {r} {dd}", flush=True))

    def load(js, name):
        s = session.create_script(js); s.load()
        print(f"[*] loaded {name}", flush=True)
        return s

    try:
        load(ART_CALLSITE.read_text(), "art_callsite")
    except Exception as e:
        print("[*] bypass load err:", str(e)[:160], flush=True)
    d.resume(pid)
    print("[*] resumed, running", DURATION, "s", 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 == "FOUND_HDID_32":
            print(f"\n*** FOUND_HDID_32 = {p.get('val')} ***", flush=True)
        elif t in ("mod-base", "ready", "scan-result", "sslwrite-hooked", "hook-err"):
            print(f"[{t}] {p.get('val')}", flush=True)
        elif t == "WUP_LOGIN_FULL":
            b = bytes.fromhex(p.get("val", "").split("B ")[-1]) if "B " in p.get("val", "") else b""
            h = re.search(rb"hdid.{0,100}", b)
            print(f"[WUP_LOGIN] len={p.get('val','').split('B')[0]} ctx={(h.group(0)[:110] if h else b[:60])}", flush=True)
        events.append(p)
        OUT.write_text(json.dumps({"events": events, "detached": det}, indent=1), encoding="utf-8")

    main_sc = load(MAIN_JS, "main_hook")
    main_sc.on("message", on_message)

    t0 = time.time()
    timeline = []
    start_pid = pid
    while time.time() - t0 < DURATION:
        cur = main_pid()
        timeline.append({"t": round(time.time() - t0, 1), "proc": cur, "pid_changed": cur != start_pid})
        if cur is None or cur != start_pid:
            print(f"[*] 主进程 pid 变化/消失: {start_pid} -> {cur}", flush=True)
            break
        time.sleep(3)
    print(f"[*] done, {len(events)} events, alive_pid={start_pid == main_pid()}", flush=True)
    OUT.write_text(json.dumps({"events": events, "timeline": timeline, "detached": det}, indent=1), encoding="utf-8")
    print(f"[*] saved {OUT}", flush=True)


if __name__ == "__main__":
    main()