Files
live-hub-py/scripts/probe_bigout.py
T

129 lines
4.0 KiB
Python

#!/usr/bin/env python3
"""定位 dfpReport 加密体生成函数: 批量 hook libhydeviceid.so / libtscsdk.so
导出函数, 记录返回长度 >500B 的调用。登录触发后, 找输出含魔数
571882cf664bb39401ee 或长度 ≈3908B 的函数。
"""
from __future__ import annotations
import json
import sys
import time
from pathlib import Path
import frida
PACKAGE = "com.duowan.kiwi"
RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
OUT = Path("/Users/yml/codes/douyu_login_py/evidence/dfp_bigout_capture.json")
JS = r"""
'use strict';
Process.setExceptionHandler(function(d){
send({type:'segv', info:{type:d.type, addr:String(d.address)}});
return true;
});
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 ra(ctx){
try{ return DebugSymbol.fromAddress(ctx.returnAddress).toString().slice(0,90);}catch(e){return '?';}
}
var TARGETS = ['libhydeviceid.so', 'libtscsdk.so', 'libturingga.so'];
var MIN_LEN = 3000;
var MAX_LEN = 6000;
var hooked = 0;
TARGETS.forEach(function(modName){
try{
var m = Process.getModuleByName(modName);
}catch(e){ send({type:'skip', mod:modName}); return; }
var exps = m.enumerateExports();
exps.forEach(function(exp){
var n = exp.name;
if (n.indexOf('datadiv') === 0) return; // 字符串解码器, 跳过
if (n.indexOf('std::') === 0) return; // std 符号, 跳过
if (/^(_Z|Java_|JNI_)/.test(n)) return; // C++ 符号/Java 方法名太长, 跳过
try{
Interceptor.attach(exp.address, {
onEnter: function(args){
this.t0 = Date.now();
this.args = [];
for (var i = 0; i < Math.min(4, args.length); i++) {
try{
var v = args[i].toInt32();
if (v > 0 && v < 10000000) this.args.push(v);
}catch(e){}
}
},
onLeave: function(ret){
var dt = Date.now() - this.t0;
var v = ret.toInt32();
if ((v >= MIN_LEN && v <= MAX_LEN) || dt > 30) {
send({type:'big', mod:modName, fn:exp.name, ret:v, dt:dt,
args:this.args, r:ra(this.ctx)});
}
}
});
hooked += 1;
}catch(e){}
});
});
send({type:'hooked_count', n:hooked});
"""
def main():
device = frida.get_device_manager().add_remote_device("127.0.0.1:31877")
pid = device.spawn([PACKAGE])
print(f"[*] spawned {PACKAGE} pid={pid}")
session = device.attach(pid)
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()
events = []
def on_message(message, data):
if message.get("type") == "error":
print("[JS-ERR]", str(message)[:160]); return
if message.get("type") != "send":
return
p = message.get("payload") or {}
t = p.get("type")
if t == "big":
print(f"[big] {p.get('mod')} {p.get('fn')} ret={p.get('ret')} dt={p.get('dt')}ms r={p.get('r','')}")
events.append(p)
elif t == "hooked_count":
print(f"[*] hooked {p.get('n')} exports")
elif t == "skip":
print(f"[skip] {p.get('mod')}")
script = session.create_script(JS)
script.on("message", on_message)
script.load()
print("[*] hooks 装好, 请在手机上 退出登录 -> 重新登录 触发 dfpReport 链 (120s)")
deadline = time.time() + 120
try:
while time.time() < deadline:
time.sleep(5)
except KeyboardInterrupt:
pass
OUT.write_text(json.dumps(events, ensure_ascii=False, indent=1))
print(f"[*] 捕获 {len(events)} 条 -> {OUT}")
if __name__ == "__main__":
main()