- 诊断报告: attach 主进程静默退出/EGL 崩溃, 仅约 4s 窗口可抓帧 - scripts: attach/spawn/hook/emu 系列 Frida 脚本与抓帧/验证工具 - evidence: identity/reqchain/frame/inputbuf/magic_buf/propedge 抓取样本, emu_* 存活对比, diag_* 策略实验, baseline 裸测基准
84 lines
3.4 KiB
Python
84 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""抓帧自动验证: spawn抓dfpReport → 重放 → 输出actionV(hdid).
|
|
用于对比"改IMEI/设备标识后身份是否变化"。
|
|
"""
|
|
from pathlib import Path
|
|
import frida, time, subprocess, json, re, ssl, socket
|
|
REMOTE="127.0.0.1:31878"; PACKAGE="com.duowan.kiwi"
|
|
RE=Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
|
|
OUT=Path("/Users/yml/codes/douyu_login_py/evidence/identity_" + time.strftime("%H%M%S") + ".json")
|
|
MAIN_JS="""
|
|
'use strict';
|
|
send({type:'armed'});
|
|
function hexb(p,n){try{return Array.from(new Uint8Array(p.readByteArray(n))).map(b=>('0'+b.toString(16)).slice(-2)).join('');}catch(e){return '';}}
|
|
function head(p,n){try{return p.readCString(n);}catch(e){return '';}}
|
|
var got=false;
|
|
try{
|
|
var r=new ApiResolver('module');
|
|
r.enumerateMatchesSync('exports:*!SSL_write').forEach(function(m){
|
|
Interceptor.attach(m.address,{onEnter:function(a){
|
|
var len=a[2].toInt32();
|
|
if(len<50||len>50000)return;
|
|
var h=head(a[1],Math.min(len,1500));
|
|
if(h.indexOf('dfpReport')>=0){got=true;send({type:'dfp',len:len,hex:hexb(a[1],len)});}
|
|
}});
|
|
});
|
|
send({type:'hooked'});
|
|
}catch(e){send({type:'err',e:String(e)});}
|
|
"""
|
|
def main():
|
|
d=frida.get_device_manager().add_remote_device(REMOTE)
|
|
result={}
|
|
for attempt in range(1,7):
|
|
subprocess.run(["adb","-s","127.0.0.1:5555","shell","am","force-stop",PACKAGE],capture_output=True)
|
|
time.sleep(1.2)
|
|
try:
|
|
pid=d.spawn([PACKAGE]); s=d.attach(pid)
|
|
except Exception as e:
|
|
print(f"[att{attempt}] spawn/attach err {e}",flush=True); time.sleep(2); continue
|
|
try:
|
|
b=s.create_script((RE/"evidence/scripts/bypass_msaoaid_maps_skip_cleanup.js").read_text()); b.load()
|
|
except: pass
|
|
d.resume(pid)
|
|
got=False
|
|
def on(m,dd):
|
|
nonlocal got
|
|
if m.get('type')!='send':return
|
|
p=m.get('payload') or {}
|
|
if p.get('type')=='dfp':
|
|
got=True
|
|
result['dfp_wire']=p['hex']; result['pid']=pid
|
|
print(f"[att{attempt}] dfpReport len={p['len']}",flush=True)
|
|
sc=s.create_script(MAIN_JS); sc.on('message',on); sc.load()
|
|
t0=time.time()
|
|
while time.time()-t0<18:
|
|
time.sleep(2)
|
|
if got: time.sleep(3); break
|
|
if got and result.get('dfp_wire'):
|
|
# 重放拿 actionV
|
|
wire=bytes.fromhex(result['dfp_wire'])
|
|
ctx=ssl.create_default_context(); ctx.check_hostname=False; ctx.verify_mode=ssl.CERT_NONE
|
|
ss=ctx.wrap_socket(socket.create_connection(("wsapi.huya.com",443),timeout=10),server_hostname="wsapi.huya.com")
|
|
ss.sendall(wire); ss.settimeout(25); buf=b""
|
|
try:
|
|
while True:
|
|
c=ss.recv(8192)
|
|
if not c: break
|
|
buf+=c
|
|
if len(buf)>5000: break
|
|
except socket.timeout: pass
|
|
ss.close()
|
|
av=re.search(rb'actionV\(([0-9a-f]{40})',buf)
|
|
avd=av.group(1).decode() if av else None
|
|
result['actionV']=avd
|
|
print(f"[att{attempt}] actionV = {avd}",flush=True)
|
|
json.dump(result,open(OUT,'w'),indent=2)
|
|
print(f"[*] saved {OUT}",flush=True)
|
|
try: d.kill(pid)
|
|
except: pass
|
|
return
|
|
try: d.kill(pid)
|
|
except: pass
|
|
print("[*] 未抓到 dfpReport",flush=True)
|
|
if __name__=="__main__": main()
|