live-hub-py/scripts/phone_probe_rn_lazy.py

#!/usr/bin/env python3
"""真机: 全 dlopen 拦截 -> 每个库 JNI_OnLoad 前置 RegisterNatives 钩子,
主动触发 getGUID/NativeBridge 类初始化, 抓 NativeBridge 懒注册归属.

用法: python3 scripts/phone_probe_rn_lazy.py [serial] [remote]
"""
from __future__ import annotations
import json, subprocess, sys, 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"
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 jmod(a){ try{var m=Process.findModuleByAddress(a); return m?{n:m.name,off:'0x'+a.sub(m.base).toString(16)}:{n:'?',off:'?'};}catch(e){return {n:'?',off:'?'};} }
var rnHooked=false;
function hookRN(){
  if(rnHooked) return;
  try{
    var env=Java.vm.tryGetEnv(); if(!env) return;
    var rn=env.handle.readPointer().add(215*Process.pointerSize).readPointer();
    if(rn.isNull()) return;
    Interceptor.attach(rn,{
      onEnter:function(args){
        var methods=args[3], count=args[4].toInt32(), mm=[];
        for(var i=0;i<count;i++){
          var m=methods.add(i*Process.pointerSize*3);
          try{
            var name=m.readPointer().readCString();
            var sig=m.add(Process.pointerSize).readPointer().readCString();
            var fn=m.add(Process.pointerSize*2).readPointer();
            var jm=jmod(fn);
            mm.push({name:name, sig:sig, mod:jm.n, off:jm.off});
          }catch(e){}
        }
        send({type:'rn', methods:mm});
      }
    });
    rnHooked=true;
    send({type:'rn-hooked'});
  }catch(e){}
}
setInterval(hookRN, 30);

function hookJniOnLoad(md){
  try{
    var jni=md.findExportByName('JNI_OnLoad');
    if(!jni) return;
    var name=md.name;
    Interceptor.attach(jni,{
      onEnter:function(){ send({type:'ol', mod:name, base:''+md.base}); hookRN(); },
      onLeave:function(ret){ send({type:'ol-out', mod:name, ret:ret.toString()}); }
    });
    send({type:'jni-hooked', mod:name});
  }catch(e){}
}
var hookedMods={};
setInterval(function(){
  Process.enumerateModules().forEach(function(md){
    if(hookedMods[md.name]) return;
    if(md.name.indexOf('libhydeviceid')>=0||md.name.indexOf('libudb')>=0||md.name.indexOf('msaoaid')>=0||
       md.name.indexOf('device')>=0||md.name.indexOf('sec')>=0||md.name.indexOf('auth')>=0||
       md.name.indexOf('wup')>=0||md.name.indexOf('hycrypto')>=0||md.name.indexOf('jsiwup')>=0){
      hookedMods[md.name]=true;
      hookJniOnLoad(md);
    }
  });
}, 50);

setTimeout(function(){
  Java.perform(function(){
    try{
      // 先强制 NativeBridge 类初始化(触发可能的懒注册)
      var NB=Java.use('com.huya.security.hydeviceid.NativeBridge');
      send({type:'cls-loaded'});
      var NE=Java.use('com.huya.security.hydeviceid.NativeEntry');
      NE.init();
      send({type:'tid', tid:Process.getCurrentThreadId()});
      send({type:'trigger', guid: (function(){ try{ return NE.getGUID(); }catch(e){ return 'ERR:'+String(e).slice(0,60);} })(),
                            mid: (function(){ try{ return NE.getMID(); }catch(e){ return 'ERR';} })() });
      try{ var b100=NB.b(100); send({type:'b100', v:String(b100)}); }catch(e){ send({type:'b100err', e:String(e).slice(0,80)}); }
    }catch(e){ send({type:'err', e:String(e).slice(0,150)}); }
  });
}, 9000);
"""


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


def main():
    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}", flush=True)
    s = d.attach(pid)
    s.create_script(ART_CALLSITE.read_text()).load()
    seen = set()
    regs = []

    def on_m(m, data):
        p = m.get("payload") or {}
        t = p.get("type")
        if t == "rn":
            for mm in p.get("methods", []):
                k = (mm["name"], mm["sig"], mm["mod"], mm["off"])
                if k not in seen:
                    seen.add(k)
                    regs.append(mm)
                    print(f"[RN] {mm['name']} {mm['sig']} @ {mm['mod']} {mm['off']}", flush=True)
        elif t in ("rn-hooked", "cls-loaded", "trigger", "b100", "b100err", "err", "tid"):
            print(f"[*] {t}: {json.dumps(p, ensure_ascii=False)[:220]}", flush=True)
        elif t in ("ol", "ol-out", "jni-hooked"):
            print(f"[*] {t}: {json.dumps(p, ensure_ascii=False)[:160]}", flush=True)
        elif m.get("type") == "error":
            print("[JS-ERR]", m.get("description")[:120], flush=True)

    sc = s.create_script(JS)
    sc.on("message", on_m)
    sc.load()
    d.resume(pid)
    time.sleep(17)
    print(f"[*] total unique regs: {len(regs)}")


if __name__ == "__main__":
    main()