live-hub-py/scripts/patch_t1_experiment.py

#!/usr/bin/env python3
"""改写 libhydeviceid.so .data 中的 t1 指纹, 验证 dfpReport 响应变化。

t1 = 865a4924a40897ac1fcfe6b4c2cbb045 (3处静态内置, datadiv 解密)。
改为随机 32hex 后重新登录, 看 dfpReport 响应 t1 是否变化。
同时 hook SSL_write 抓新请求 body 对比魔数后密文是否变化。
"""
from __future__ import annotations

import json
import random
import string
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_t1_patch.json")

# 新 t1 (随机 32hex)
NEW_T1 = "".join(random.choice("0123456789abcdef") for _ in range(32))
print(f"[*] 新 t1 = {NEW_T1}")

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

var NEW_T1 = '%s';
var T1_OFFS = [0x3bb190, 0x3bb230, 0x3e0e50];

// 等待: so 加载后延时 800ms, 确保 datadiv 就地解密已完成
var tries = 0;
var iv = setInterval(function(){
  tries += 1;
  var m = null;
  try{ m = Process.getModuleByName('libhydeviceid.so'); }catch(e){}
  if (m) {
    clearInterval(iv);
    setTimeout(function(){
      patchAll(m);
    }, 800);
  } else if (tries > 100) { clearInterval(iv); send({type:'no_mod'}); }
}, 50);

function patchAll(m){
  // 先验证当前明文, 若仍是密文说明 datadiv 未跑完, 再等
  try{
    var cur = m.base.add(0x3bb190).readUtf8String(32);
    if (cur.indexOf('865a') !== 0 && cur.indexOf('c1') !== 0) {
      // 还没解密好, 再等
      setTimeout(function(){ patchAll(m); }, 400);
      return;
    }
  }catch(e){}
  T1_OFFS.forEach(function(off){
    try{
      var p = m.base.add(off);
      var cur = p.readUtf8String(32);
      send({type:'backup', off:off.toString(16), cur:cur});
      Memory.protect(p, 64, 'rwx');
      p.writeUtf8String(NEW_T1);
      var after = p.readUtf8String(32);
      send({type:'patched', off:off.toString(16), after:after, ok: after === NEW_T1});
    }catch(e){ send({type:'patch_err', off:off.toString(16), e:String(e)}); }
  });
  send({type:'patch_done'});
}

// hook SSL_write 抓 dfpReport 请求
try{
  var fns = [];
  var r = new ApiResolver('module');
  fns = r.enumerateMatchesSync('exports:*!SSL_write').map(function(x){ return x.address; });
  fns.slice(0, 6).forEach(function(p){
    Interceptor.attach(p, {
      onEnter: function(a){
        var len = a[2].toInt32();
        if (len < 100 || len > 20000) return;
        var head = '';
        try{ head = a[1].readCString(Math.min(len, 2000)); }catch(e){ return; }
        if (head.indexOf('dfpReport') < 0) return;
        var arr = new Uint8Array(a[1].readByteArray(len));
        var hex = '';
        for (var j = 0; j < arr.length; j++) hex += ('0'+arr[j].toString(16)).slice(-2);
        send({type:'wire', len:len, hex:hex});
      }
    });
  });
}catch(e){ send({type:'ssl_err', e:String(e)}); }
""" % NEW_T1


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 == "backup":
            print(f"[备份] 0x{p.get('off')}: {p.get('cur')}", flush=True)
        elif t == "patched":
            print(f"[改写] 0x{p.get('off')}: -> {p.get('after')} ok={p.get('ok')}", flush=True)
        elif t == "patch_done":
            print(f"[*] 改写完成 {p.get('patched')} 处", flush=True)
            events.append({"patched": True, "new_t1": NEW_T1})
            OUT.write_text(json.dumps(events, ensure_ascii=False, indent=1))
        elif t == "wire":
            body = bytes.fromhex(p.get('hex'))
            try: hp = body.index(b'\r\n\r\n') + 4
            except ValueError: hp = 0
            b = body[hp:]
            j = b.find(b'\x57\x18\x82\xcf\x66\x4b\xb3\x94')
            print(f"[wire] len={p.get('len')} 魔数@{j}", flush=True)
            if j >= 0:
                print(f"   密文头: {b[j+10:j+30].hex(' ')}", flush=True)
            rec = {"type": "wire", "len": p.get('len'), "hex": p.get('hex')}
            events.append(rec)
            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(f"[*] 已 resume, t1 已改为 {NEW_T1}。请在手机 退出登录->重新登录", flush=True)
    try:
        while True:
            time.sleep(5)
    except KeyboardInterrupt:
        pass
    print(f"[*] 结束, {len(events)} 条 -> {OUT}", flush=True)


if __name__ == "__main__":
    main()