live-hub-py/scripts/phone_probe_hdid3.py

#!/usr/bin/env python3
"""真机探针 v3: 等 so 加载后 hook, 抓 32hex hdid 生成/使用链路 + 登录触发.

修正 v1/v2: 挂起时 libhydeviceid.so 尚未加载 -> 用 setInterval 轮询,
so 一加载即 hook:
  - libhydeviceid.so: 枚举非 runtime 导出 + hook JNI_OnLoad(0x22c4c8)
  - libudbauthunify.so: hook BusinessCfg::getHdid(0x26a484) / setSafeDeviceId(0x26a2e0) 带调用栈
resume 后 am start GameSdkLoginActivity 触发设备上报/登录, 让 hdid 被读取/生成.

用法:
  /Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/.venv/bin/python \
      scripts/phone_probe_hdid3.py [serial] [remote]
输出:
  evidence/diag_phone/hdid_probe3.json
"""
from __future__ import annotations

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

import frida

SERIAL = sys.argv[1] if len(sys.argv) > 1 else "5dd8c93f"
REMOTE = sys.argv[2] if len(sys.argv) > 2 else "127.0.0.1:31878"
PACKAGE = "com.duowan.kiwi"
LOGIN_ACT = "com.duowan.kiwi/.loginui.impl.gamesdk.GameSdkLoginActivity"
REPO = Path("/Users/yml/codes/douyu_login_py")
OUT = REPO / "evidence" / "diag_phone" / "hdid_probe3.json"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
ART_CALLSITE = RE / "evidence" / "scripts" / "bypass_msaoaid_maps_art_callsite.js"

JS = r"""
'use strict';
function stdstr(p){
  try{
    if(p.isNull()) return '';
    var first = p.readU8();
    if((first&1)===0){ var l=first>>1; return l? p.add(1).readUtf8String(l):''; }
    else { var d=p.readPointer(); var l=p.add(8).readU64(); return (l&&l<256)? d.readUtf8String(l):''; }
  }catch(e){ return ''; }
}
var hydev=null, udb=null;
function hookHydev(){
  if(hydev) return;
  var md = Process.findModuleByName('libhydeviceid.so');
  if(!md) return;
  hydev = md;
  send({type:'hydev', base:''+md.base, size:md.size});
  var ex = md.enumerateExports();
  var useful = ex.filter(function(e){ return !/^_Z/.test(e.name) && !/^__/.test(e.name) && e.name!==''; });
  send({type:'hydev-exports', count:useful.length,
        exports: useful.map(function(e){ return {off:''+e.address.sub(md.base), name:e.name}; })});
  try{
    Interceptor.attach(md.base.add(0x22c4c8), {
      onEnter:function(){ send({type:'jni-onload-enter', tid:Process.getCurrentThreadId()}); },
      onLeave:function(ret){ send({type:'jni-onload-leave', ret:''+ret}); }
    });
    send({type:'jni-onload-hooked'});
  }catch(e){ send({type:'hydev-err', e:String(e)}); }
}
function hookUdb(){
  if(udb) return;
  var md = Process.findModuleByName('libudbauthunify.so');
  if(!md) return;
  udb = md;
  send({type:'udb', base:''+md.base});
  try{
    Interceptor.attach(md.base.add(0x26a484), {
      onLeave:function(ret){ send({type:'hdid', str:stdstr(ret), tid:Process.getCurrentThreadId()}); }
    });
    send({type:'hdid-hooked'});
  }catch(e){ send({type:'udb-err', e:String(e)}); }
  try{
    Interceptor.attach(md.base.add(0x26a2e0), {
      onEnter:function(a){
        send({type:'setsd', sd:stdstr(a[1]), hd:stdstr(a[2]), tid:Process.getCurrentThreadId()});
        try{
          var bt=Thread.backtrace(this.context, Backtracer.ACCURATE).slice(0,20).map(function(x){return x.toString();});
          send({type:'bt', bt:bt});
        }catch(e){}
      }
    });
    send({type:'setsd-hooked'});
  }catch(e){}
}
setInterval(hookHydev, 40);
setInterval(hookUdb, 40);
"""


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


def main():
    OUT.parent.mkdir(parents=True, exist_ok=True)
    adb("shell", "am", "force-stop", PACKAGE)
    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)
    det = []
    session.on("detached", lambda r, dd: det.append({"reason": r, "detail": str(dd)}) or print(f"[*] detached {r} {dd}", flush=True))
    session.create_script(ART_CALLSITE.read_text()).load()
    result = {"events": [], "detached": det}
    def on_message(m, data):
        if m.get("type") == "error":
            print("[JS-ERR]", str(m)[:200], flush=True); return
        p = m.get("payload") or {}
        result["events"].append(p)
        t = p.get("type")
        if t in ("hdid", "setsd"):
            print(f"[{t}] {p.get('hd') or p.get('str')} tid={p.get('tid')}", flush=True)
        elif t == "bt":
            print("   bt:", " <- ".join(p["bt"][:8]), flush=True)
        elif t == "hydev":
            print(f"[*] libhydeviceid loaded base={p.get('base')}", flush=True)
        elif t == "hydev-exports":
            print(f"[*] hydev 非runtime导出 {p.get('count')}:", flush=True)
            for e in (p.get("exports") or [])[:40]:
                print(f"     0x{e['off']}  {e['name']}")
        elif t in ("udb","hdid-hooked","setsd-hooked","jni-onload-hooked","jni-onload-enter","jni-onload-leave"):
            print(f"[*] {t}", p if t in ("jni-onload-leave",) else "", flush=True)
    sc = session.create_script(JS)
    sc.on("message", on_message)
    sc.load()
    d.resume(pid)
    print("[*] resumed", flush=True)
    time.sleep(5)
    print("[*] 触发登录 Activity...", flush=True)
    adb("shell", "am", "start", "-n", LOGIN_ACT)
    time.sleep(15)
    result["detached"] = det
    OUT.write_text(json.dumps(result, indent=1, ensure_ascii=False), encoding="utf-8")
    print(f"[*] saved {OUT}")


if __name__ == "__main__":
    main()