yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(huya): frida证书生成链逆向 - 全链路hook工具与文档

Browse Source 
- 定位完整调用链: HYUDBMSDKCommon桥接 -> MsgGetH5InfoEx(184549387)
  -> HandlerGetH5InfoEx::onHandler@0x38d548 -> getOtp@0x2689f4
  -> hyudb_otp_encrypt@0x32fa24 -> createWupRequestData@0x38dab0
- 实测参数表: otp(uid串,2,轮换idx,'5008',SESS_HEX32,hyCred114B,4,nonce)
- 密钥字符串确认: 'HuyaUdb1928374650qwertyuiop' (encrypt x3 参数)
- Java桥直呼触发: LoginProxy.getQUrlData/getH5InfoEx/getCred 可编程调用
- 稳定时序固化: spawn挂起->bypass->resume->10s->patch_guard->hooks
  (attach已运行进程必被检测闪退)
- 发现App登录态JSON含现成web cookie.biztoken(344B)
This commit is contained in:
yml2213
2026-08-25 14:21:06 +08:00
parent 9aea9851bd
commit 2076510df9
5 changed files with 816 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/虎牙扫码登录闭环逆向.md
+62
View File
@@ -77,3 +77,65 @@ cookies = {
--har "/Users/yml/Desktop/抓包/hy/8.24/手机扫码登录-手机端2.har"
# 成功记录: evidence/qr_login_success.json（含有效 biztoken，已 gitignore）
```
---
## 七、Frida 逆向补充（2026-08-25 下午，手机实测）
### 7.1 调用链（已全部定位）
```
JS callHandler("HYUDBMSDKCommon"/"HYUDBMSDKQUrlCommon")
└ com.huyaudbunify.HuyaAuth.getH5InfoEx(uid) / getQUrlData(uid,"","")
msgId: MsgGetH5InfoEx=184549387(0x0B00000B), meta id 同族 184549392
└ HandlerGetH5InfoEx::onHandler @0x38d548 (libudbauthunify.so)
├ BusinessCfg::getOtp @0x2689f4
│ ├ loadLoginData() ← 本地存储登录态(hyCred 等)
│ ├ copyBase64StringToVectChar(safedeviceid)
│ ├ getServiceTime + nonce_next → nonce = (time_ms<<16)|seq
│ ├ AESkeyMgr key轮换(b2 参数, 1..14 循环)
│ └ hyudb_otp_encrypt @0x32fa24
└ WupDataPackage<AppCommonData>::createWupRequestData @0x38dab0
```
### 7.2 hook 实测的 otp 输入输出
| 参数 | 值 |
|---|---|
| x0 s1 | uid 十进制字符串 ("1199666914671") |
| w1 | 恒 2 |
| w2 b2 | AESkeyMgr 轮换索引 (1..14 循环) |
| x3 s2 | "5008" |
| x4 s3 | SESS_HEX32 "865a4924a40897ac1fcfe6b4c2cbb0e3" |
| x5 s4 | hyCred 二进制 (114B, base64 后即 udb_cred 152字符) |
| w6 | 恒 4 |
| x7 | nonce |
| out | [2B头][AES密文] (登录态 146B / 匿名 ~118B) |
### 7.3 两层 AES 明文（已抓全）
- 内层 P2 (131B): `\x02\x0c\x00<12B xxtea(arg7)>...` cred_pack 结构
- 外层 P1 (187B): `\x01\x04\x00"5008"\x14\x00<20B随机>\x00<指纹40hex><72 00 0a><10B><cred二进制>`
- 尾部证书 = base64([2B头][AES(P1 补零到192)]) = 194B ✓ 与样本吻合
### 7.4 关键发现：App 登录态 JSON（aes 明文 1106B）
```json
{"biztoken_vec":[{"bizAppid":"5008","biztoken":"AhNEUAts...","errCode":0}],
"cookie":{"biztoken":"AQCRaYXu...(344B web令牌!)","uid":"...","version":"1.0"},
"cred":"CmAozmok10sOm1AwUNs...", ...}
```
→ App 存储里就有现成的 **web cookie.biztoken**！`HuyaAuth.getLoginDataCookies(uid)` 可取。
### 7.5 Java 桥直呼（全自动触发，无需人工）
```js
var seed = Java.use('com.duowan.kiwi.base.login.udb.HuyaLoginProxy');
var F = Java.ClassFactory.get(seed.class.getClassLoader()); // 双loader注意:
// HuyaAuth 系用 com.huyaudbunify.util.HuyaAccountSaveUtils 所在 loader
var inst = F.use('com.hysdkproxy.LoginProxy').getInstance();
inst.getQUrlData(1199666914671, "", ""); // → 1332字符证书wupData
HA.getCred(uid); // → ResGetCred{hyCred(152), yyCred}
```
### 7.6 稳定 hook 时序（不可变）
spawn挂起 → attach → 仅装 bypass_msaoaid_maps_art_callsite → resume → 等10s
→ patch_guard_block_termination → 业务hooks。**attach 已运行进程必闪退。**
工具脚本: scripts/hook_final_capture.py（全量采集）/ hook_cred_dump.py（cred+wupData）
数据: evidence/final_capture.json, final_wupdata.bin, frida_cred_dump.json
scripts/hook_cred_dump.py
+172
View File
@@ -0,0 +1,172 @@
#!/usr/bin/env python3
"""一次性抓取: App 存储的 cred + Common/QUrl 两种 wupData 完整输出。"""
from __future__ import annotations
import argparse
import json
import time
from pathlib import Path
import frida
RE_DIR = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
JS = r"""
'use strict';
function b64ify(ptr, n) {
var bytes = ptr.readByteArray(n);
var arr = new Uint8Array(bytes);
var B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
var chars = "";
for (var i = 0; i < arr.length; i += 3) {
var b0 = arr[i], b1 = i+1 < arr.length ? arr[i+1] : 0, b2 = i+2 < arr.length ? arr[i+2] : 0;
chars += B64[b0>>2] + B64[((b0&3)<<4)|(b1>>4)] +
(i+1 < arr.length ? B64[((b1&15)<<2)|(b2>>6)] : "=") +
(i+2 < arr.length ? B64[b2&63] : "=");
}
return chars;
}
// otp 加密链 hook(二进制安全)
var base = Process.getModuleByName("libudbauthunify.so").base;
Interceptor.attach(base.add(0x32fa24), {
onEnter: function (args) {
function rd(p) {
try {
var b0 = p.readU8();
if ((b0 & 1) === 0) return {t:"s", v:p.add(1).readUtf8String(b0>>1)};
var len = parseInt(p.add(8).readU64().toString());
return {t:"b64", v:b64ify(p.add(16).readPointer(), Math.min(len,4096))};
} catch(e) { return {t:"err", v:String(e)}; }
}
this.s1 = rd(args[0]); this.b2 = args[2].toInt32();
this.s2 = rd(args[3]); this.s3 = rd(args[4]); this.s4 = rd(args[5]);
this.nonce = args[7].toString();
this.outPtr = this.context.sp.readPointer();
},
onLeave: function () {
function rdo(p) {
try {
var b0 = p.readU8();
if ((b0 & 1) === 0) return {t:"b64", v:b64ify(p.add(1), b0>>1)};
var len = parseInt(p.add(8).readU64().toString());
return {t:"b64", v:b64ify(p.add(16).readPointer(), Math.min(len,4096))};
} catch(e){ return {t:"err", v:String(e)}; }
}
send({type:"otp", s1:this.s1, b2:this.b2, s2:this.s2, s3:this.s3,
s4:this.s4, nonce:this.nonce, out:rdo(this.outPtr)});
}
});
function trigger() {
Java.perform(function () {
var attempt = 0;
var timer = null;
function once() {
attempt += 1;
var done = false;
Java.enumerateClassLoaders({
onMatch: function (loader) {
if (done) return;
try {
var f = Java.ClassFactory.get(loader);
var LP = f.use('com.hysdkproxy.LoginProxy');
var inst = LP.getInstance();
var HAS = f.use('com.huyaudbunify.HuyaAccountSaveUtils').getInstance();
var uid = HAS.getUid();
send({type:'uid', uid: String(uid)});
var HA = f.use('com.huyaudbunify.HuyaAuth').getInstance();
var cred = HA.getCred(uid);
if (cred !== null) {
send({type:'cred',
hyCred: cred.getHyCred() ? cred.getHyCred() : null,
yyCred: cred.getYyCred() ? cred.getYyCred() : null});
} else {
send({type:'cred', hyCred:null, yyCred:null});
}
var common = inst.getH5InfoEx();
send({type:'wup_common', len: common?common.length:0, data: common});
var qurl = inst.getQUrlData(uid, "", "");
send({type:'wup_qurl', len: qurl?qurl.length:0, data: qurl});
done = true;
} catch (e) { /* 下一个 loader */ }
},
onComplete: function () {
if (!done) {
if (attempt % 4 === 0) send({type:'retry', n: attempt});
timer = setTimeout(once, 3000);
} else {
send({type:'all_done', attempts: attempt});
}
}
});
}
once();
});
}
trigger();
"""
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--duration", type=float, default=40.0)
ap.add_argument("--stabilize", type=float, default=10.0)
ap.add_argument("--out", default="/tmp/full_dump.json")
ap.add_argument("--package", default="com.duowan.kiwi")
ap.add_argument("--port", default="127.0.0.1:31877")
args = ap.parse_args()
bypass_src = (RE_DIR / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()
patch_guard = (RE_DIR / "evidence/scripts/patch_guard_block_termination.js").read_text()
device = frida.get_device_manager().add_remote_device(args.port)
pid = device.spawn([args.package])
print(f"spawned pid={pid}")
session = device.attach(pid)
session.create_script(bypass_src).load()
device.resume(pid)
time.sleep(args.stabilize)
session.create_script(patch_guard).load()
print("guards loaded")
collected = {}
def on_message(message, _data):
if message.get("type") != "send":
if message.get("type") == "error":
print("ERR:", str(message)[:200])
return
p = message.get("payload") or {}
t = p.get("type")
if t in ("wup_common", "wup_qurl"):
collected[t] = p.get("data")
print(f"{t}: len={p.get('len')}")
elif t == "cred":
collected["cred"] = p
hc, yc = p.get("hyCred"), p.get("yyCred")
print(f"cred: hyCred={str(hc)[:60]}... ({len(hc) if hc else 0}) "
f"yyCred={str(yc)[:60]}... ({len(yc) if yc else 0})")
elif t == "uid":
collected["uid"] = p.get("uid")
print(f"uid={p.get('uid')}")
elif t == "loader_err":
print("loader:", p.get("err"))
else:
print(json.dumps(p, ensure_ascii=False)[:120])
script = session.create_script(JS)
script.on("message", on_message)
script.load()
time.sleep(args.duration - args.stabilize)
out = Path(args.out)
out.write_text(json.dumps(collected, ensure_ascii=False))
print(f"\nsaved: {out} keys={list(collected.keys())}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/hook_crypto_live.py
+173
View File
@@ -0,0 +1,173 @@
#!/usr/bin/env python3
"""最终轮 hook: 加密函数现场抓取(AES key/明文/密文) + Java 触发 getQUrlData。"""
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")
JS = r"""
'use strict';
function b64ify(ptr, n) {
var arr = new Uint8Array(ptr.readByteArray(n));
var B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
var c = "";
for (var i=0;i<arr.length;i+=3){
var b0=arr[i],b1=i+1<arr.length?arr[i+1]:0,b2=i+2<arr.length?arr[i+2]:0;
c+=B64[b0>>2]+B64[((b0&3)<<4)|(b1>>4)]+(i+1<arr.length?B64[((b1&15)<<2)|(b2>>6)]:"=")+(i+2<arr.length?B64[b2&63]:"=");
}
return c;
}
function rd(p){
try{
var b0=p.readU8();
if((b0&1)===0) return {t:"s",v:p.add(1).readUtf8String(b0>>1)};
var len=parseInt(p.add(8).readU64().toString());
if(len>8192) return {t:"err",v:"too long"};
return {t:"b64",v:b64ify(p.add(16).readPointer(),len)};
}catch(e){return {t:"err",v:String(e)};}
}
var armed = false;
var timer = setInterval(function(){
if (armed) { clearInterval(timer); return; }
var base;
try { base = Process.getModuleByName("libudbauthunify.so").base; }
catch(e) { return; }
armed = true;
clearInterval(timer);
send({type:'base', v:String(base)});
// md5_char16(string& src, string& out16)
Interceptor.attach(base.add(0x32e71c), {
onEnter: function(a){
this.inp = rd(a[0]);
this.outp = a[1];
// 原始内存也存一份
try { this.rawin = b64ify(a[0], 40); } catch(e){ this.rawin=''; }
},
onLeave: function(){
send({type:'md5c16', inp:this.inp, out:rd(this.outp), rawin:this.rawin});
}
});
// KeyExpansion: 两个参数都转储
var keCount = 0;
Interceptor.attach(base.add(0x24ebd4), {
onEnter: function(a){
keCount += 1;
if (keCount > 2) return;
try {
var h = function(p,n){ return Array.from(new Uint8Array(p.readByteArray(n))).map(function(x){return ('0'+x.toString(16)).slice(-2);}).join(''); };
send({type:'aeskey_raw', n:keCount, x0:h(a[0],32), x1:h(a[1],48)});
} catch(e){}
}
});
// UdbAESUtil::encrypt: x2 字符串对象的原始内存转储
Interceptor.attach(base.add(0x250038), {
onEnter: function(a){
this.inp = rd(a[1]);
try {
var p = a[2];
var b0 = p.readU8();
var info = {b0:b0};
if ((b0&1)===0) { info.len=b0>>1; info.data=b64ify(p.add(1), Math.min(b0>>1,48)); }
else {
info.len = parseInt(p.add(8).readU64().toString());
info.cap = parseInt(p.readU64().toString());
info.data = b64ify(p.add(16).readPointer(), Math.min(info.len,64));
}
send({type:'aeskey', info:info});
} catch(e){ send({type:'aeskey', info:{err:String(e)}}); }
this.out = a[3];
},
onLeave: function(){ send({type:'aes_out', out:rd(this.out)}); }
});
// hyudb_otp_encrypt
Interceptor.attach(base.add(0x32fa24), {
onEnter: function(a){
this.s1=rd(a[0]); this.b1=a[1].toInt32(); this.b2=a[2].toInt32();
this.s2=rd(a[3]); this.s3=rd(a[4]); this.s4=rd(a[5]);
this.n=a[7].toString();
this.outP=this.context.sp.readPointer();
},
onLeave: function(){
send({type:'otp', s1:this.s1,b1:this.b1,b2:this.b2,s2:this.s2,s3:this.s3,
s4:this.s4,b3:this.b3,nonce:this.n,out:rd(this.outP)});
}
});
send({type:'armed'});
// 触发: 等类可用后轮询
Java.perform(function(){
var n = 0;
function tryOnce(){
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();
var q = inst.getQUrlData(1199666914671, "", "");
send({type:'qurl_done', len: q?q.length:0});
var c = inst.getH5InfoEx();
send({type:'common_done', len: c?c.length:0});
} catch(e) {
n += 1;
if (n % 6 === 0) send({type:'trig_retry', n:n, err:String(e).substring(0,70)});
setTimeout(tryOnce, 5000);
}
}
setTimeout(tryOnce, 3000);
});
}, 400);
"""
def main():
d = frida.get_device_manager().add_remote_device('127.0.0.1:31877')
pid = d.spawn(['com.duowan.kiwi'])
print('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/patch_guard_block_termination.js").read_text()).load()
d.resume(pid)
print('resumed')
events = []
def on_msg(m, _):
if m.get('type') != 'send':
if m.get('type') == 'error':
print('ERR:', str(m)[:180])
return
p = m.get('payload') or {}
events.append(p)
t = p.get('type')
if t == 'otp':
print(f"\nOTP s1={p['s1']['v'][:20]} b1={p['b1']} b2={p['b2']} "
f"s2={p['s2']['v']} s3={str(p['s3'])[:40]}")
s4 = p['s4']
print(f" s4[{s4.get('t')}] {len(s4.get('v',''))}B nonce={p['nonce']}")
sc = s.create_script(JS)
sc.on('message', on_msg)
sc.load()
time.sleep(75)
Path('/tmp/crypto_events.json').write_text(json.dumps(events, ensure_ascii=False))
aes = [e for e in events if e.get('type') == 'aes_in']
print(f"\n事件统计: otp={sum(1 for e in events if e.get('type')=='otp')} "
f"aes={len(aes)} ctp={sum(1 for e in events if e.get('type')=='ctp')}")
for e in aes[:6]:
ki = e['key']
kv = ki['v'] if isinstance(ki, dict) else str(ki)
ii = e['inp']
iv = ii['v'] if isinstance(ii, dict) else str(ii)
print(f"AES key[{kv[:24]}...] in[{len(iv)}B]")
if __name__ == '__main__':
main()
scripts/hook_final_capture.py
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""按已验证成功的时序采集证书生成全链路数据。
时序（不可变）: spawn挂起 -> bypass单独装载 -> resume -> 稳定10s -> patch_guard -> 业务hooks
然后提示用户 手动退出登录 -> 重新登录。
"""
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")
JS = r"""
'use strict';
var base = Process.getModuleByName("libudbauthunify.so").base;
send({type:'armed', base:String(base)});
function b64ify(ptr, n) {
var arr = new Uint8Array(ptr.readByteArray(n));
var B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
var c = "";
for (var i=0;i<arr.length;i+=3){
var b0=arr[i],b1=i+1<arr.length?arr[i+1]:0,b2=i+2<arr.length?arr[i+2]:0;
c+=B64[b0>>2]+B64[((b0&3)<<4)|(b1>>4)]+(i+1<arr.length?B64[((b1&15)<<2)|(b2>>6)]:"=")+(i+2<arr.length?B64[b2&63]:"=");
}
return c;
}
function rdStrObj(p){
try{
var b0=p.readU8();
if((b0&1)===0) return {t:"s",len:b0>>1,v:p.add(1).readUtf8String(b0>>1)};
var len=parseInt(p.add(8).readU64().toString());
if(len>16384) return {t:"err",v:"too long"};
return {t:"b64",v:b64ify(p.add(16).readPointer(),len)};
}catch(e){return {t:"err",v:String(e)};}
}
function hexof(p,n){
try{ return Array.from(new Uint8Array(p.readByteArray(n))).map(function(x){return ('0'+x.toString(16)).slice(-2);}).join(''); }
catch(e){ return 'ERR'; }
}
Interceptor.attach(base.add(0x32e71c), {
onEnter: function(a){ this.inp = rdStrObj(a[0]); this.outp = a[1]; },
onLeave: function(){ send({type:'md5', inp:this.inp, out:rdStrObj(this.outp)}); }
});
Interceptor.attach(base.add(0x32fa24), {
onEnter: function(a){
this.a = {s1:rdStrObj(a[0]), b1:a[1].toInt32(), b2:a[2].toInt32(),
s2:rdStrObj(a[3]), s3:rdStrObj(a[4]), s4:rdStrObj(a[5]),
b3:a[6].toInt32(), nonce:a[7].toString()};
this.outp = this.context.sp.readPointer();
},
onLeave: function(){ send({type:'otp', args:this.a, out:rdStrObj(this.outp)}); }
});
// sret 约定: x0=返回串, x1=this, x2=in, x3=key
Interceptor.attach(base.add(0x250038), {
onEnter: function(a){
this.retp = a[0];
this.inp = rdStrObj(a[2]);
this.key = rdStrObj(a[3]);
this.thishead = hexof(a[1],48);
},
onLeave: function(){
send({type:'aes', inp:this.inp, key:this.key, out:rdStrObj(this.retp),
this_head:this.thishead});
}
});
Interceptor.attach(base.add(0x38dab0), {
onEnter: function(a){ this.ret = a[0]; },
onLeave: function(){ send({type:'createWup', result:rdStrObj(this.ret)}); }
});
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();
var q = inst.getQUrlData(1199666914671, "", "");
send({type:'qurl_done', len: q?q.length:0});
} catch(e){ n+=1; if(n%8===0) send({type:'retry', n:n}); setTimeout(go, 5000); }
}
setTimeout(go, 20000);
});
"""
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)
# 双重防护: callsite bypass + maps 掩盖
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')
events = []
def on_msg(m, _):
if m.get('type') != 'send':
if m.get('type') == 'error':
print('ERR:', str(m)[:160])
return
p = m.get('payload') or {}
t = p.get('type')
if t != 'retry':
events.append(p)
if t == 'armed':
print(f'[armed] base={p["base"]}')
elif t == 'otp':
a = p['args']
print(f"[otp] uid={str(a['s1'].get('v'))[:18]} b={a['b1']},{a['b2']},{a['b3']}")
elif t == 'md5':
print(f"[md5] in={str(p['inp'].get('v',''))[:36]} out={str(p['out'].get('v',''))[:40]}")
elif t == 'aes':
k = p.get('key',{})
print(f"[aes] key[{k.get('t')}]={str(k.get('v',''))[:44]}")
print(f" in={str(p.get('inp',{}).get('v',''))[:90]}")
print(f" out={str(p.get('out',{}).get('v',''))[:60]}")
elif t == 'createWup':
print(f"[createWup] len={len(str(p['result'].get('v','')))}")
elif t == 'qurl_done':
print(f"[qurl_done] {p.get('len')}")
sc = s.create_script(JS)
sc.on('message', on_msg)
sc.load()
print('\n>>> 请现在在手机上: 手动退出登录 -> 重新登录 <<<\n')
deadline = time.time() + 260
while time.time() < deadline:
time.sleep(2)
if any(e.get('type') == 'qurl_done' for e in events):
time.sleep(6)
break
Path('/tmp/final_capture.json').write_text(json.dumps(events, ensure_ascii=False))
types = {}
for e in events:
types[e.get('type')] = types.get(e.get('type'), 0) + 1
print('saved:', types)
if __name__ == '__main__':
import subprocess
for attempt in range(4):
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}')
time.sleep(3)
scripts/hook_h5infoex.py
+240
View File
@@ -0,0 +1,240 @@
#!/usr/bin/env python3
"""全自动 hook：触发并抓取虎牙 App 的 H5InfoEx(证书)生成全链路。
流程: spawn 挂起 → 构造窗口挂 bypass → resume → 稳定后挂业务 hooks
→ 通过 Java 桥直呼 LoginProxy.getH5InfoEx() 触发生成（零人工）
输出: /tmp/h5infoex_dump.jsonl
"""
from __future__ import annotations
import argparse
import json
import time
from pathlib import Path
import frida
RE_DIR = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
JS_AGENT = r"""
'use strict';
function rdStr(p) {
try {
var b0 = p.readU8();
if ((b0 & 1) === 0) {
var n = b0 >> 1;
return {t:"s", v: p.add(1).readUtf8String(n)};
}
var len = parseInt(p.add(8).readU64().toString());
var dp = p.add(16).readPointer();
return {t:"s", v: dp.readUtf8String(len)};
} catch (e) {
// 二进制内容: 走 base64
try {
var b0b = p.readU8();
if ((b0b & 1) === 0) {
var nb = b0b >> 1;
return {t:"b64", v: base64ify(p.add(1), nb)};
}
var len2 = parseInt(p.add(8).readU64().toString());
var dp2 = p.add(16).readPointer();
return {t:"b64", v: base64ify(dp2, Math.min(len2, 4096))};
} catch (e2) {
return {t:"err", v: String(e)};
}
}
}
function base64ify(ptr, n) {
var bytes = ptr.readByteArray(n);
var arr = new Uint8Array(bytes);
var chars = "";
var B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
for (var i = 0; i < arr.length; i += 3) {
var b0 = arr[i], b1 = i+1 < arr.length ? arr[i+1] : 0, b2 = i+2 < arr.length ? arr[i+2] : 0;
chars += B64[b0>>2] + B64[((b0&3)<<4)|(b1>>4)] +
(i+1 < arr.length ? B64[((b1&15)<<2)|(b2>>6)] : "=") +
(i+2 < arr.length ? B64[b2&63] : "=");
}
return chars;
}
var base = Process.getModuleByName("libudbauthunify.so").base;
send({type:"info", msg:"base=" + base});
// hyudb_otp_encrypt(string,u8,u8,string,string,string,u8,ulong,string&)
Interceptor.attach(base.add(0x32fa24), {
onEnter: function (args) {
this.s1 = rdStr(args[0]);
this.b1 = args[1].toInt32();
this.b2 = args[2].toInt32();
this.s2 = rdStr(args[3]);
this.s3 = rdStr(args[4]);
this.s4 = rdStr(args[5]);
this.b3 = args[6].toInt32();
this.nonce = args[7].toString();
this.outPtr = this.context.sp.readPointer();
},
onLeave: function () {
send({type:"otp", s1:this.s1, b1:this.b1, b2:this.b2, s2:this.s2,
s3:this.s3, s4:this.s4, b3:this.b3, nonce:this.nonce,
out: rdStr(this.outPtr)});
}
});
// BusinessCfg::getOtp(u64, string&, int&)
Interceptor.attach(base.add(0x2689f4), {
onEnter: function (args) {
this.uid = args[1].toString();
this.outStr = args[2];
},
onLeave: function () {
send({type:"getOtp", uid:this.uid, out: rdStr(this.outStr)});
}
});
// WupDataPackage<AppCommonData>::createWupRequestData(...)
Interceptor.attach(base.add(0x38dab0), {
onEnter: function (args) {
this.ret = args[0];
this.c3 = args[3].readCString();
this.c4 = args[4].readCString();
},
onLeave: function () {
send({type:"createWup", servant:this.c3, func:this.c4,
result_b64: rdStr(this.ret)});
}
});
send({type:"ready"});
"""
JS_TRIGGER = r"""
'use strict';
function tryTrigger() {
Java.perform(function () {
var found = false;
Java.enumerateClassLoaders({
onMatch: function (loader) {
if (found) return;
try {
var factory = Java.ClassFactory.get(loader);
var LP = factory.use('com.hysdkproxy.LoginProxy');
var inst = LP.getInstance();
var r = inst.getH5InfoEx();
found = true;
send({type:'trigger_ok', len: r ? r.length : 0,
head: r ? r.substring(0, 80) : 'null'});
} catch (e) {
send({type:'trigger_try_err', err: String(e).substring(0,120)});
}
},
onComplete: function () {
if (!found) send({type:'trigger_fail'});
}
});
});
}
tryTrigger();
"""
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--duration", type=float, default=90.0)
ap.add_argument("--stabilize", type=float, default=10.0)
ap.add_argument("--out", default="/tmp/h5infoex_dump.jsonl")
ap.add_argument("--package", default="com.duowan.kiwi")
ap.add_argument("--port", default="127.0.0.1:31877")
args = ap.parse_args()
bypass_src = (RE_DIR / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text()
patch_guard = (RE_DIR / "evidence/scripts/patch_guard_block_termination.js").read_text()
device = frida.get_device_manager().add_remote_device(args.port)
print(f"remote device ok")
# 1) spawn 挂起 → 构造窗口期只挂 bypass
pid = device.spawn([args.package])
print(f"spawned pid={pid}")
session = device.attach(pid)
s_bypass = session.create_script(bypass_src)
s_bypass.load()
device.resume(pid)
print("bypass loaded, resumed")
time.sleep(args.stabilize)
# 2) patch_guard + 业务 hooks
session.create_script(patch_guard).load()
out_path = Path(args.out)
fout = out_path.open("w")
got = {"otp": False, "wup": False}
def on_message(message, _data):
if message.get("type") != "send":
if message.get("type") == "error":
print("SCRIPT ERR:", str(message)[:200])
return
p = message.get("payload") or {}
fout.write(json.dumps(p, ensure_ascii=False) + "\n")
fout.flush()
t = p.get("type")
def unwrap(x):
if isinstance(x, dict) and "t" in x:
v = x["v"]
tag = x["t"]
if isinstance(v, str) and len(v) > 200:
return f"[{tag}]{v[:150]}...({len(v)}B)"
return f"[{tag}]{v}"
return x
if t == "otp":
got["otp"] = True
print("\n=== hyudb_otp_encrypt ===")
for k in ("s1","b1","b2","s2","s3","s4","b3","nonce"):
print(f" {k} = {unwrap(p.get(k))}")
print(f" out = {unwrap(p.get('out'))}")
elif t == "createWup":
got["wup"] = True
r = p["result_b64"]
print(f"\n=== createWupRequestData servant={p['servant']} func={p['func']} ===")
print(f" result({len(r)}B) = {r[:140]}...")
elif t == "getOtp":
o = p.get("out")
if isinstance(o, dict):
o = f"[{o['t']}]{str(o['v'])[:140]}"
print(f"\n=== getOtp uid={p['uid']} out={o}")
elif t in ("ready", "info"):
print(f"[{t}] {p.get('msg','')}")
script = session.create_script(JS_AGENT)
script.on("message", on_message)
script.load()
print("\nhooks armed")
# 3) 轮询式 Java 触发（等 App 完全初始化）
s_trig = session.create_script(JS_TRIGGER)
s_trig.on("message", on_message)
deadline = time.time() + args.duration
triggered = False
while time.time() < deadline and not triggered:
try:
s_t2 = session.create_script(JS_TRIGGER)
s_t2.on("message", on_message)
s_t2.load()
time.sleep(6)
if got["otp"] or got["wup"]:
triggered = True
except Exception as e:
print("trigger retry:", str(e)[:80])
time.sleep(3)
# 等最后的异步 dump 落盘
time.sleep(3)
fout.close()
print(f"\nresult: otp={got['otp']} wup={got['wup']} -> {out_path}")
return 0 if (got["otp"] and got["wup"]) else 1
if __name__ == "__main__":
raise SystemExit(main())