live-hub-py/scripts/hook_emu_hdid_spawn.py

#!/usr/bin/env python3
"""模拟器 32hex hdid 抓取 v3: poll 等待模块 + hook createWupDeviceInfo(ret+152=hdid).

createWupDeviceInfo(0x2746A0): 每次 wup 打包必调用 (udb 上报/登录帧都走它),
返回 DeviceInfo 结构 +152 = hdid std::string (文档十九节真机验证过偏移)。
同时 hook setSafeDeviceId(0x26A2E0) / getHdid(0x26A484) 双保险。
"""
import frida, time, re, subprocess, json
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("/tmp/emu_hdid_v3.json")

HDID_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 emit(tag, val){ send({type:'hdid', tag:tag, val:''+(val||''), ts:Date.now()}); }

function install(){
  var md = Process.findModuleByName('libudbauthunify.so');
  if (!md) return false;
  emit('mod-base', ''+md.base);

  // 1) createWupDeviceInfo: 返回结构 +152 = hdid
  try{
    Interceptor.attach(md.base.add(0x2746A0), {
      onLeave:function(ret){
        try{
          var s = readStdString(ret.add(152));
          if (s) emit('WUPDEV_HDID', ''+s);
        }catch(e){ emit('WUPDEV_ERR', ''+e); }
      }
    });
    emit('createWupDev-hooked','');
  }catch(e){ emit('createWupDev-err', ''+e); }

  // 2) setSafeDeviceId: a2=sd, a3=hdid
  try{
    Interceptor.attach(md.base.add(0x26A2E0), {
      onEnter:function(a){
        this.sd = readStdString(a[1]);
        this.hd = readStdString(a[2]);
      },
      onLeave:function(){
        emit('SETSD', (this.hd||'') + ' | sd=' + (this.sd||'').slice(0,24));
      }
    });
    emit('setSd-hooked','');
  }catch(e){ emit('setSd-err', ''+e); }

  // 3) getHdid: this+1088 (x8)
  try{
    Interceptor.attach(md.base.add(0x26A484), {
      onEnter:function(a){
        try{ this.thiz = this.context.x8; }catch(e){ this.thiz = null; }
      },
      onLeave:function(){
        if (!this.thiz) return;
        try{
          var s = readStdString(this.thiz.add(1088));
          emit('GETHDID', ''+s);
        }catch(e){}
      }
    });
    emit('getHdid-hooked','');
  }catch(e){ emit('getHdid-err', ''+e); }

  emit('ready','');
  return true;
}

var installed = false;
function poll(){
  if (!installed) installed = install();
  if (!installed) setTimeout(poll, 200);
}
poll();
"""

MAIN_JS = r"""
'use strict';
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 '';}}
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,800));
      if(h.indexOf('hypasswordLogin')>=0||h.indexOf('huyaudbwebui')>=0){
        send({type:'wup', len:len, hex:hexb(a[1],len), t:Date.now()});
      }
    }});
  });
}catch(e){ send({type:'err', 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)
    load(session, (RE/"evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text())
    load(session, (RE/"evidence/scripts/mask_frida_maps_only.js").read_text())
    hdid_sc = load(session, HDID_JS)
    print("[*] HDID v3 loaded (poll-wait)", flush=True)
    d.resume(pid)
    print("[*] resumed", flush=True)
    load(session, (RE/"evidence/scripts/patch_guard_block_termination.js").read_text(), wait=0.1)

    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 == 'hdid':
            print(f"[HDID {p.get('tag')}] {p.get('val')}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events, indent=1))
        elif t == 'wup':
            b = bytes.fromhex(p.get('hex',''))
            if b'hidden' in b[:50]:
                print(f"[WUP] len={p.get('len')}", flush=True)
            else:
                h = re.search(rb'hdid.{0,120}', b)
                print(f"[WUP] len={p.get('len')} ctx={h.group(0)[:120] if h else b[:60]}", flush=True)
            events.append(p); OUT.write_text(json.dumps(events, indent=1))
        elif t == 'err':
            print("[err]", p.get('e'), flush=True)

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

if __name__ == "__main__":
    main()