dfpReport: 确定加密为逐字节XOR(动态keystream)+同步pair捕获+IDA导出线索, 记录破解进度

This commit is contained in:
yml2213
2026-08-26 14:17:58 +08:00
parent acb0a737eb
commit 4eec01568f
38 changed files with 4942 additions and 3 deletions
+179
View File
@@ -0,0 +1,179 @@
#!/usr/bin/env python3
"""attach 到已运行的虎牙App进程, hook dfpReport 加密链 + datadiv 解码器。
不 spawn,直接 attach 到用户正在操作的实例, 避免挂错进程。
用法: 脚本跑起来后, 在手机上 退出登录 -> 重新登录 一次。
"""
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_attach_capture.json")
JS = r"""
'use strict';
Process.setExceptionHandler(function(d){ 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 rdStr(p){
try{
var b0=p.readU8();
if((b0&1)===0){ var l=b0>>1; if(l>524288) return {t:'s?',len:l,v:''};
return {t:'s',len:l,v:l?hx(p.add(1),Math.min(l,16384)):''}; }
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,16384))};
}catch(e){ return {t:'err',len:-1,v:String(e)}; }
}
function ra(ctx){
try{ return DebugSymbol.fromAddress(ctx.returnAddress).toString().slice(0,90);}catch(e){return '?';}
}
var KEYWORDS = ['dfp', '5718', '82cf', 'b394', 'collect', 'report'];
// 1) datadiv 解码器 (libhydeviceid + libudbauthunify)
var ddHooked = 0;
['libhydeviceid.so', 'libudbauthunify.so'].forEach(function(modName){
var m = null;
try{ m = Process.getModuleByName(modName); }catch(e){ return; }
m.enumerateExports().forEach(function(exp){
if (exp.name.indexOf('datadiv') < 0) return;
try{
Interceptor.attach(exp.address, {
onLeave: function(ret){
var s = '';
try{ s = ret.readUtf8String(); }catch(e){}
if (!s || s.length < 3 || s.length > 2000) return;
var low = s.toLowerCase();
var hit = false;
KEYWORDS.forEach(function(k){ if (low.indexOf(k) >= 0) hit = true; });
if (hit) send({type:'dd', mod:modName, fn:exp.name, str:s.slice(0,300)});
}
});
ddHooked += 1;
}catch(e){}
});
});
// 2) xxtea (hyudb_crypt_util + hyudbxxt)
var base = Process.getModuleByName('libudbauthunify.so').base;
try{
Interceptor.attach(base.add(0x32e83c), {
onEnter: function(a){
send({type:'xxtea', r:ra(this), in:rdStr(a[1]), key:rdStr(a[2])});
}
});
}catch(e){}
// 3) UdbAESUtil::encrypt (in len 大 -> dfp)
try{
Interceptor.attach(base.add(0x250038), {
onEnter: function(a){
var inp = rdStr(a[0]);
if (inp.len > 500) {
send({type:'aes_in', r:ra(this), in:inp});
}
},
onLeave: function(ret){
var out = rdStr(this.context.x0);
if (out.len > 500) send({type:'aes_out', out:out});
}
});
}catch(e){}
// 4) enpack_header / enpack_body (打包)
try{
Interceptor.attach(base.add(0x32ff44), {
onEnter: function(a){ this.h=a[0].toInt32()&0xff; this.r=ra(this); },
onLeave: function(){ send({type:'enpack_hdr', h:this.h, r:this.r, out:rdStr(this.context.x2)}); }
});
}catch(e){}
try{
Interceptor.attach(base.add(0x3300c8), {
onEnter: function(a){
this.h=a[0].toInt32()&0xff; this.r=ra(this);
send({type:'enpack_body_in', h:this.h, r:this.r,
s1:rdStr(a[1]), s2:rdStr(a[2]), s3:rdStr(a[3]), s4:rdStr(a[4])});
},
onLeave: function(){ send({type:'enpack_body_out', h:this.h, out:rdStr(this.context.x5)}); }
});
}catch(e){}
send({type:'dd_hooked', n:ddHooked});
"""
def main():
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--pid", type=int, default=7296)
args = parser.parse_args()
device = frida.get_device_manager().add_remote_device("127.0.0.1:31877")
session = device.attach(args.pid)
print(f"[*] attached pid={args.pid}")
# attach 模式不需要 msaoaid bypass (进程已过检测点), 但保险起见加载 patch_guard
try:
session.create_script(
(RE / "evidence/scripts/patch_guard_block_termination.js").read_text()
).load()
except Exception as e:
print("[warn] patch_guard:", str(e)[:80])
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 == "dd":
print(f"[dd] {p.get('mod')} {p.get('fn')}: {p.get('str')!r}")
elif t == "xxtea":
inp = p.get("in", {}); key = p.get("key", {})
print(f"[xxtea] in.len={inp.get('len')} key={key.get('v','')[:40]} r={p.get('r','')[:50]}")
elif t == "aes_in":
inp = p.get("in", {})
print(f"[aes_in] len={inp.get('len')} head={inp.get('v','')[:64]} r={p.get('r','')[:50]}")
elif t == "aes_out":
out = p.get("out", {})
print(f"[aes_out] len={out.get('len')} head={out.get('v','')[:64]}")
elif t == "enpack_hdr":
print(f"[enpack_hdr] h={p.get('h')} out.len={p.get('out',{}).get('len','?')} r={p.get('r','')[:40]}")
elif t == "enpack_body_in":
print(f"[enpack_body_in] h={p.get('h')} s1={p.get('s1',{}).get('v','')[:48]} r={p.get('r','')[:40]}")
elif t == "enpack_body_out":
print(f"[enpack_body_out] h={p.get('h')} out.len={p.get('out',{}).get('len','?')}")
elif t == "armed":
print("[*] armed")
elif t == "dd_hooked":
print(f"[*] datadiv hooked: {p.get('n')}")
events.append(p)
script = session.create_script(JS)
script.on("message", on_message)
script.load()
print("[*] hooks 装好, 请在手机上 退出登录 -> 重新登录 (180s)")
deadline = time.time() + 180
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()