live-hub-py/scripts/hook_cred_issue.py

#!/usr/bin/env python3
"""抓 hyCred 签发链: 挂 saveLoginData(落盘) + LoginCred handler(vtable探测)。

用法: 脚本跑起来后, 在手机上 手动退出登录 -> 重新登录 一次。
捕获: 登录态JSON明文(含cred字段) / saveLoginData 入参 / LoginCred handler 触发。
"""
from __future__ import annotations

import json
import subprocess
import time
from pathlib import Path

import frida

RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")

JS = r"""
'use strict';
Process.setExceptionHandler(function(d){ send({type:'segv',info:{t:d.type,a:String(d.address)}}); return true; });
var base = Process.getModuleByName('libudbauthunify.so').base;
send({type:'armed'});
function hx(p,n){ try{ return Array.from(new Uint8Array(p.readByteArray(n))).map(x=>('0'+x.toString(16)).slice(-2)).join(''); }catch(e){ return 'ERR'; } }
function rdStr(p){
  try{
    var b0=p.readU8();
    if((b0&1)===0){ var l=b0>>1; if(l>262144) return {t:'s?',len:l,v:''};
      return {t:'s',len:l,v:l?hx(p.add(1),Math.min(l,4096)):''}; }
    var len=parseInt(p.add(8).readU64().toString());
    if(len>1048576) return {t:'l?',len:len,v:''};
    return {t:'l',len:len,v:hx(p.add(16).readPointer(),Math.min(len,4096))};
  }catch(e){ return {t:'err',len:-1,v:String(e)}; }
}
function ra(ctx){ try{ return DebugSymbol.fromAddress(ctx.returnAddress).toString().slice(0,70);}catch(e){return '?';} }

// 登录态JSON的AES加密(含cred字段) —— 已知走 UdbAESUtil::encrypt
Interceptor.attach(base.add(0x250038), {
  onEnter: function(a){ this.x1=rdStr(a[1]); this.x2=rdStr(a[2]); },
  onLeave: function(){ 
    if(this.x1.len>800||this.x2.len>800)
      send({type:'bigaes', x1:{len:this.x1.len}, x2:this.x2, ra:ra(this)});
  }
});

// saveLoginData(string const&, BusBeansLoginData&, bool) 与 (BusBeansLoginData&, bool)
Interceptor.attach(base.add(0x265fb0), {
  onEnter: function(a){
    send({type:'saveLD_str', str:rdStr(a[1]), bean_head:hx(a[2],128), ra:ra(this)});
  }
});
Interceptor.attach(base.add(0x265a6c), {
  onEnter: function(a){
    send({type:'saveLD_bean', bean_head:hx(a[1],160), b:rdStr(a[1]), flag:a[2].toInt32(), ra:ra(this)});
  }
});

// HandlerRequestLoginCred / HandlerResponseLoginCred vtable 探测:
// vtable+0x10 起为虚函数槽, 逐个attach前3个槽, 打印触发与backtrace首帧
function probeVtable(name, vtAddr){
  for(var slot=0; slot<4; slot++){
    (function(slot){
      try{
        var fp = vtAddr.add(0x10 + slot*8).readPointer();
        if(fp.compare(base) < 0 || fp.compare(base.add(0x480000)) > 0) return;
        Interceptor.attach(fp, {
          onEnter: function(a){
            send({type:'vtable', name:name, slot:slot,
                  a0:a[0].toString(), a1:rdStr(a[1]), a2:rdStr(a[2]),
                  bt:Thread.backtrace(this.context, Backtracer.FUZZY).slice(0,3)
                      .map(function(x){return DebugSymbol.fromAddress(x).toString().slice(0,60);})});
          }
        });
        send({type:'vt_attached', name:name, slot:slot, fp:String(fp.sub(base))});
      }catch(e){}
    })(slot);
  }
}
probeVtable('ReqLoginCred', base.add(0x477f80));
probeVtable('RespLoginCred', base.add(0x478018));
probeVtable('GetCred',      base.add(0x477000)); // 占位: 若无效仅跳过
"""


def main():
    d = frida.get_device_manager().add_remote_device('127.0.0.1:31877')
    pid = d.spawn(['com.duowan.kiwi'])
    s = d.attach(pid)
    s.create_script((RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()).load()
    s.create_script((RE / "evidence/scripts/mask_frida_maps_only.js").read_text()).load()
    d.resume(pid)
    time.sleep(11)
    s.create_script((RE / "evidence/scripts/patch_guard_block_termination.js").read_text()).load()
    print('armed. >>> 请在手机上退出登录并重新登录 <<<')

    sc = s.create_script(JS)
    events = []
    def on_msg(m, _):
        if m.get('type') == 'error':
            print('JS ERR:', str(m)[:200]); return
        if m.get('type') != 'send':
            return
        p = m.get('payload') or {}
        t = p.get('type')
        events.append(p)
        if t == 'vt_attached':
            print(f"[vt] {p['name']} slot{p['slot']} @+{p['fp']}")
        elif t == 'saveLD_str':
            print(f"[saveLD_str] str={p['str']['t']}:{p['str']['len']}:{p['str']['v'][:80]}")
        elif t == 'saveLD_bean':
            print(f"[saveLD_bean] flag={p['flag']} bean={p['bean_head'][:96]}")
            print(f"   asStr={p['b']}")
        elif t == 'bigaes':
            print(f"[bigaes] len={p['x2']['len']} head={p['x2']['v'][:120]}")
        elif t == 'vtable':
            print(f"[VT!] {p['name']}#{p['slot']} a1={p['a1']} bt={p['bt']}")
    sc.on('message', on_msg)
    sc.load()
    deadline = time.time() + 300
    while time.time() < deadline:
        time.sleep(3)
        if any(e.get('type') == 'saveLD_bean' for e in events):
            time.sleep(10)
            break
    Path('/Users/yml/codes/douyu_login_py/evidence/cred_issue_trace.json').write_text(json.dumps(events))
    from collections import Counter
    print('saved:', Counter(e.get('type') for e in events))


if __name__ == '__main__':
    for attempt in range(4):
        try:
            main()
            break
        except frida.InvalidOperationError as e:
            print('detached:', e)
            subprocess.run(['adb', 'shell', 'am', 'force-stop', 'com.duowan.kiwi'])
            time.sleep(3)