live-hub-py/scripts/hook_nonce_native.py

#!/usr/bin/env python3
"""native侧追查nonce生成: 枚举libudbauthunify.so符号 + 窗口内hook sha/hmac/rand。

时序(不可变): force-stop -> spawn挂起 -> 双bypass -> resume -> 10s -> patch_guard
-> 枚举符号 -> hook命中函数 -> Java桥触发getQUrlData -> 收集调用与nonce比对。
"""
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")
UID = 1199666914671

ENUM_JS = r"""
'use strict';
var m = Process.getModuleByName('libudbauthunify.so');
send({type:'info', base:String(m.base), size:m.size});
var pat = /sha1|sha-1|hmac|rand|urandom|sign|digest|nonce|md5|aes_key/i;
var syms = m.enumerateSymbols().filter(function(s){
  return s.name && pat.test(s.name);
}).map(function(s){ return {name:s.name, addr:String(s.address)}; });
var exps = m.enumerateExports().filter(function(e){ return pat.test(e.name); })
  .map(function(e){ return {name:e.name, addr:String(e.address)}; });
send({type:'syms', syms:syms, exps:exps});
"""

HOOK_JS_TPL = r"""
'use strict';
Process.setExceptionHandler(function(d){
  send({type:'segv', info:{type:d.type, addr:String(d.address)}}); return true;
});
var base = Process.getModuleByName('libudbauthunify.so').base;
send({type:'armed', base:String(base)});
var WIN_ON = false;

function rdStr(p){
  try{
    var b0=p.readU8();
    if((b0&1)===0){ var l=b0>>1; if(l>4096) return {t:'s?',len:l,v:''};
      return {t:'s',len:l,v:l?Array.from(new Uint8Array(p.add(1).readByteArray(l))).map(x=>('0'+x.toString(16)).slice(-2)).join(''):''}; }
    var len=parseInt(p.add(8).readU64().toString());
    if(len>65536) return {t:'l?',len:len,v:''};
    var dp=p.add(16).readPointer();
    return {t:'l',len:len,v:Array.from(new Uint8Array(dp.readByteArray(len))).map(x=>('0'+x.toString(16)).slice(-2)).join('')};
  }catch(e){ return {t:'err',len:-1,v:String(e)}; }
}
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 bt(ctx){ return Thread.backtrace(ctx, Backtracer.ACCURATE).map(function(a){
  return DebugSymbol.fromAddress(a).toString().slice(0,110); }).slice(0,6); }

// __HITS__: [ [addrStr, name], ... ]
var hits = __HITS__;
hits.forEach(function(h){
  try {
    Interceptor.attach(ptr(h[0]), {
      onEnter: function(a){
        if (!WIN_ON) return;
        this.a = [a[0],a[1],a[2],a[3]];
        this.bt = bt(this.context);
      },
      onLeave: function(r){
        if (!WIN_ON) return;
        send({type:'hit', name:h[1].slice(0,90),
              args:[rdStr(this.a[0]), rdStr(this.a[1])],
              ret:rdStr(r), bt:this.bt});
      }
    });
    send({type:'hooked', name:h[1].slice(0,80)});
  } catch(e){ send({type:'hookerr', where:h[1], msg:''+e}); }
});

// 窗口控制 + 触发
Java.perform(function(){
  var n=0;
  function go(){
    try{
      var seed=Java.use('com.duowan.kiwi.base.login.udb.HuyaLoginProxy');
      var F=Java.ClassFactory.get(seed.class.getClassLoader());
      var inst=F.use('com.hysdkproxy.LoginProxy').getInstance();
      WIN_ON = true;
      send({type:'stage', msg:'window ON'});
      var q=inst.getQUrlData(__UID__, "", "");
      WIN_ON = false;
      send({type:'qurl_done', len:q?q.length:0, data:q||''});
    }catch(e){ n+=1; window.on=false; if(n%4===0) send({type:'retry',n:n}); setTimeout(go,5000); }
  }
  setTimeout(go, 6000);
});
""".replace('__UID__', str(UID))


def main():
    d = frida.get_device_manager().add_remote_device('127.0.0.1:31877')
    pid = d.spawn(['com.duowan.kiwi'])
    print(f'spawned {pid}')
    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)
    print('resumed')
    time.sleep(11)
    s.create_script((RE / "evidence/scripts/patch_guard_block_termination.js").read_text()).load()
    print('patch_guard on')

    # 1) 枚举符号
    sc0 = s.create_script(ENUM_JS)
    out = {}
    def on0(m, _):
        if m.get('type') == 'send':
            p = m['payload']
            if p.get('type') in ('info', 'syms'):
                out[p['type']] = p
    sc0.on('message', on0)
    sc0.load()
    time.sleep(3)
    info = out['info']
    syms = out['syms']['syms'] + out['syms']['exps']
    print(f"base={info['base']} size={info['size']} 命中符号={len(syms)}")
    for x in syms[:30]:
        print("   ", x['name'][:100])

    # 2) 选 hook 目标: 定向 SHA1/md5/random_device, 上限12个
    def want(nm):
        ln = nm.lower()
        return ('sha_go' in ln or 'shainit' in ln or 'adddatalen' in ln or
                'random_device' in ln or 'md5_char' in ln or
                'huyamd5' in ln and ('digest' in ln or 'tostring' in ln))
    picks = []
    seen = set()
    for x in syms:
        nm = x['name']
        off = int(x['addr'], 16) - int(info['base'], 16)
        if off <= 0 or off > info['size']:
            continue
        if want(nm) and nm not in seen:
            seen.add(nm)
            picks.append([x['addr'], nm])
        if len(picks) >= 12:
            break
    print("hook目标:", [p[1][:60] for p in picks])

    # 3) 装钩 + 触发
    js = HOOK_JS_TPL.replace('__HITS__', json.dumps(picks))
    sc = s.create_script(js)
    events = []
    got_armed = []

    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 == 'armed':
            got_armed.append(1); print('[armed]')
        elif t == 'hooked':
            print('[hooked]', p['name'][:70])
        elif t == 'hookerr':
            print('[hookerr]', p)
        elif t == 'hit':
            a0, a1 = p['args'][0], p['args'][1]
            print(f"\n[HIT] {p['name'][:80]}")
            print(f"   arg0={a0['t']}:{a0['len']} {a0['v'][:64]}")
            print(f"   arg1={a1['t']}:{a1['len']} {a1['v'][:64]}")
            print(f"   ret={p['ret']['t']}:{p['ret']['len']} {p['ret']['v'][:64]}")
            for f in p['bt'][:4]:
                print("    <", f)
        elif t == 'stage':
            print('[stage]', p['msg'])
        elif t == 'qurl_done':
            print('[qurl_done]', p.get('len'), (p.get('data') or '')[:80])

    sc.on('message', on_msg)
    sc.load()
    assert got_armed

    deadline = time.time() + 90
    while time.time() < deadline:
        time.sleep(2)
        if any(e.get('type') == 'qurl_done' for e in events):
            time.sleep(5)
            break
    Path('/tmp/opencode/nonce_native.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(3):
        print(f'===== 尝试 #{attempt+1} =====')
        try:
            main()
            break
        except frida.InvalidOperationError as e:
            print(f'session detached: {e}')
            subprocess.run(['adb', 'shell', 'am', 'force-stop', 'com.duowan.kiwi'])
            time.sleep(3)
        except Exception as e:
            print(f'err: {e}')
            subprocess.run(['adb', 'shell', 'am', 'force-stop', 'com.duowan.kiwi'])
            time.sleep(3)