live-hub-py/scripts/hook_xxtea_crypt.py

#!/usr/bin/env python3
"""hook xxtea_crypt_util + hyudbxxt::xxtea_*: 抓 dfpReport 加密的 (in, key, out)。

xxtea_crypt_util @ libudbauthunify.so offset 3336252 (0x32e9bc附近)。
调用者 UdbUserFilterUtils::xxTeaAndBase64 处理 WUP编码体 = 与 dfpReport 同构。
修复 string 参数解析(SSO), onLeave 读输出。命中 len>2000 即持久化。
"""
from __future__ import annotations

import json
import time
from pathlib import Path

import frida

RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
OUT = Path("/Users/yml/codes/douyu_login_py/evidence/dfp_xxtea_crypt.json")

JS = r"""
'use strict';
Process.setExceptionHandler(function(d){ return true; });
send({type:'armed'});

function hexb(p, n){
  try{ return Array.from(new Uint8Array(p.readByteArray(n))).map(x=>('0'+x.toString(16)).slice(-2)).join(''); }
  catch(e){ return null; }
}

// 解析 __ndk1 basic_string (SSO): 字节0低bit: 1=内联(sso), 0=堆指针
function readString(p){
  try{
    var b0 = p.readU8();
    var sso = (b0 & 1) === 1;
    if (sso) {
      var ln = b0 >> 1;
      var data = p.add(1);
      return {len: ln, data: data};
    } else {
      var ln = p.add(8).readPointer().toInt32();
      var data = p.add(16).readPointer();
      return {len: ln, data: data};
    }
  }catch(e){ return null; }
}

function hookOne(name, off, maxLen){
  var mod = Process.findModuleByName('libudbauthunify.so');
  if (!mod) return false;
  var addr = mod.base.add(off);
  try{
    Interceptor.attach(addr, {
      onEnter: function(args){
        this.a = args;
        // 尝试解析 (ret, in, key) / (in, key) / (this, in, key) 多种
        this.in0 = readString(args[1]);  // 常见: (this, in, key)
        this.key0 = readString(args[2]);
        if (!this.in0 && readString(args[0])) { this.in0 = readString(args[0]); this.key0 = readString(args[1]); }
        if (!this.in0) { this.in0 = readString(args[2]); this.key0 = readString(args[3]); }
        this.t0 = Date.now();
      },
      onLeave: function(ret){
        if (!this.in0) return;
        var ln = this.in0.len;
        if (ln > 5000 || ln < 100) return;   // dfpReport 明文 ~3980B, 只抓大输入
        var inhex = hexb(this.in0.data, Math.min(ln, 250));
        if (!inhex) return;
        // 过滤: 找 TAF/JSON 特征
        var low = inhex;
        var isWup = low.indexOf('dfpReport') >= 0 ||
                    low.indexOf('74726571') >= 0 ||        // 'tReq'
                    low.indexOf('68757961') >= 0 ||        // 'huya'
                    low.indexOf('61707049') >= 0 ||        // 'appI'
                    low.indexOf('7b226170') >= 0;          // '{"ap'
        if (!isWup && ln < 2900) return;
        var keyhex = this.key0 ? hexb(this.key0.data, Math.min(this.key0.len, 32)) : null;
        send({type:'xxtea', fn:name, off:off.toString(16),
              inlen:ln, in:inhex, keylen: this.key0 ? this.key0.len : -1, key:keyhex,
              ret:String(ret)});
      }
    });
    return true;
  }catch(e){ return false; }
}

var T = setTimeout(function(){
  var n1 = hookOne('xxtea_crypt_util', 3336252, 99999);
  // hyudbxxt 家族
  var n2 = hookOne('hyudb_xxtea_enc1', 0x25261c, 99999);
  var n3 = hookOne('hyudb_xxtea_dec1', 0x252708, 99999);
  var n4 = hookOne('hyudb_xxtea_enc2', 0x2528e0, 99999);
  send({type:'hooked', n:[n1,n2,n3,n4].filter(Boolean).length});
  if (!n1 && !n2 && !n3 && !n4) setTimeout(arguments.callee, 1000);
}, 800);
"""


def main():
    device = frida.get_device_manager().add_remote_device("127.0.0.1:31877")
    pid = device.spawn(["com.duowan.kiwi"])
    print(f"[*] spawned pid={pid}", flush=True)
    session = device.attach(pid)

    events = []

    def on_message(message, data):
        if message.get("type") == "error":
            print("[JS-ERR]", str(message)[:200], flush=True); return
        if message.get("type") != "send":
            return
        p = message.get("payload") or {}
        t = p.get("type")
        if t == "hooked":
            print(f"[*] hooked {p.get('n')} 个 xxtea", flush=True)
        elif t == "xxtea":
            print(f"[XXTEA] {p.get('fn')}@{p.get('off')} inlen={p.get('inlen')} "
                  f"keylen={p.get('keylen')}", flush=True)
            if p.get('key'):
                print(f"    key: {p.get('key')}", flush=True)
            inh = p.get('in', '')
            print(f"    in[:100]: {inh[:100]}", flush=True)
            # 尝试 ascii
            try:
                bb = bytes.fromhex(inh[:200])
                asc = ''.join(chr(x) if 32<=x<127 else '.' for x in bb)
                print(f"    in ascii: {asc[:120]}", flush=True)
            except Exception: pass
            events.append(p)
            OUT.write_text(json.dumps(events, ensure_ascii=False, indent=1))

    script = session.create_script(JS)
    script.on("message", on_message)
    script.load()
    print("[*] 主JS 已装载(resume前), 加载 bypass + resume", flush=True)
    session.create_script(
        (RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()
    ).load()
    session.create_script(
        (RE / "evidence/scripts/mask_frida_maps_only.js").read_text()
    ).load()
    device.resume(pid)
    time.sleep(11)
    session.create_script(
        (RE / "evidence/scripts/patch_guard_block_termination.js").read_text()
    ).load()
    print("[*] 已 resume, hooking xxtea. 请 退出登录->重新登录, 命中即 pkill", flush=True)
    try:
        while True:
            time.sleep(5)
    except KeyboardInterrupt:
        pass
    print(f"[*] 结束, {len(events)} -> {OUT}", flush=True)


if __name__ == "__main__":
    main()