live-hub-py/scripts/hook_reqchain.py

#!/usr/bin/env python3
"""请求链捕获器: 冷启动 → 完整登录链。

不再只盯 dfpReport。hook SSL_write + SSL_read:
- SSL_write: 读取 body 前 2000B, 按文本标记分类 (dfpReport/hy*Credlogin/hylogout/dckey/PrtReq 等),
  记录完整 hex + len + t + 分类。
- SSL_read: 记录响应 hex + len + t, 按时间与请求配对, 看身份链 (t1/t5/actionV 的建立)。
用途: 搞清楚 dfpReport 在登录链中的角色, 设备身份(t1/t5)到底由哪个接口发放。

约定: 后台启动 + tail 实时看 + 命中即杀 (docs/Hook脚本运行与捕获约定.md)。
spawn 顺序: bypass x2 -> resume+11s -> patch_guard -> 最后主 hook JS。
"""
from __future__ import annotations

import json
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/reqchain_" + time.strftime("%H%M%S") + ".json")

# 请求分类关键字 (wup payload 名 / http 头)
MARKERS = [
    ("dfpReport", "dfpReport"),
    ("hylogout", "hylogout"),
    ("hyCredlogin", "hyCredlogin"),
    ("hyanonymousCredlogin", "hyanonymousCredlogin"),
    ("dckey", "dckey"),
    ("PrtReq", "PrtReq"),
    ("CheckUidIsBind", "CheckUidIsBind"),
    ("GetUidByGameZone", "GetUidByGameZone"),
    ("get3", "get3"),
    ("verify3", "verify3"),
]


def classify(head: bytes) -> str:
    for marker, name in MARKERS:
        if marker.encode() in head:
            return name
    return "other"


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

function hexb(p, n){ try{ return Array.from(new Uint8Array(p.readByteArray(n))).map(function(b){return ('0'+b.toString(16)).slice(-2)}).join(''); }catch(e){ return ''; } }
function head(p, n){ try{ return p.readCString(n); }catch(e){ return ''; } }
function classify(head){
  var ms = [['dfpReport','dfpReport'],['hylogout','hylogut'],['hyanonymousCredlogin','hyanonym'],
            ['hyCredlogin','hyCredlogin'],['dckey','dckey'],['PrtReq','PrtReq'],['get3','get3'],
            ['verify3','verify3']];
  for (var i=0;i<ms.length;i++){ if (head.indexOf(ms[i][0]) >= 0) return ms[i][1]; }
  return 'other';
}

var n = 0;
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 < 40 || len > 50000) return;
        var h = head(a[1], Math.min(len, 2000));
        if (!h) return;
        var name = classify(h);
        // 过滤明显无关 (cert请求等), 保留虎牙业务
        if (name === 'other' && h.indexOf('POST /') < 0 && h.indexOf(' 1@') < 0 && h.indexOf('PUT /') < 0) return;
        n++;
        send({type:'req', n:n, cls:name, len:len, head:h.slice(0,80),
              hex:hexb(a[1], len), t:Date.now()-t0});
      }
    });
  });
  send({type:'sslwrite_hooked'});
}catch(e){ send({type:'sslwrite_err', e:String(e)}); }

try{
  var r2 = new ApiResolver('module');
  r2.enumerateMatchesSync('exports:*!SSL_read').forEach(function(m){
    Interceptor.attach(m.address, {
      onEnter: function(a){ this.buf=a[1]; this.cap=a[2].toInt32(); },
      onLeave: function(ret){
        var n2 = ret.toInt32();
        if (n2 <= 0 || n2 > 3200) return;
        send({type:'resp', len:n2, hex:hexb(this.buf, n2), t:Date.now()-t0});
      }
    });
  });
  send({type:'sslread_hooked'});
}catch(e){ send({type:'sslread_err', e:String(e)}); }
"""


def load_with_retry(session, js, n=5):
    last = None
    for i in range(n):
        try:
            s = session.create_script(js)
            s.load()
            return s
        except Exception as e:
            last = e
            print(f"[retry {i}] {e}", flush=True)
            time.sleep(3)
    raise last


def main():
    # 清旧实例 (多进程坑)
    import subprocess
    subprocess.run(["adb", "shell", "am", "force-stop", PACKAGE], capture_output=True)
    device = frida.get_device_manager().add_remote_device("127.0.0.1:31877")
    pid = device.spawn([PACKAGE])
    print(f"[*] spawned {PACKAGE} pid={pid}", flush=True)
    session = device.attach(pid)

    events = []

    def on_message(message, data):
        if message.get("type") == "error":
            print("[JS-ERR]", str(message)[:300], flush=True)
            return
        if message.get("type") != "send":
            return
        p = message.get("payload") or {}
        t = p.get("type")
        if t == "armed":
            print("[*] JS armed", flush=True)
        elif t == "sslwrite_hooked":
            print("[*] SSL_write hooked", flush=True)
        elif t == "sslread_hooked":
            print("[*] SSL_read hooked", flush=True)
        elif t == "req":
            print(f"[{p.get('t')}ms] [{p.get('n')}] {p.get('cls'):12s} len={p.get('len')}", flush=True)
            events.append({"type": "req", **p})
            OUT.write_text(json.dumps(events))
        elif t == "resp":
            h = p.get('hex', '')
            import re as _re
            av = _re.search(rb'actionV\(([0-9a-f]{40})', bytes.fromhex(h))
            mark = f" actionV={av.group(1).decode()}" if av else ""
            print(f"[{p.get('t')}ms] [resp] len={p.get('len')}{mark}", flush=True)
            events.append({"type": "resp", **p})
            OUT.write_text(json.dumps(events))
        elif t in ("sslwrite_err", "sslread_err"):
            print(f"[*] {t}: {p.get('e')}", flush=True)

    # 严格约定顺序
    print("[*] loading bypass", flush=True)
    load_with_retry(session, (RE / "evidence/scripts/bypass_msaoaid_maps_art_callsite.js").read_text())
    load_with_retry(session, (RE / "evidence/scripts/mask_frida_maps_only.js").read_text())
    device.resume(pid)
    time.sleep(11)
    load_with_retry(session, (RE / "evidence/scripts/patch_guard_block_termination.js").read_text())
    print("[*] loading 主 hook JS (最后)", flush=True)
    script = load_with_retry(session, JS)
    script.on("message", on_message)
    print("[*] 已 resume。捕获取请求链 (冷启动自动触发 + 登录触发)", flush=True)
    try:
        while True:
            time.sleep(5)
    except KeyboardInterrupt:
        pass
    print(f"[*] 结束 -> {OUT}", flush=True)


if __name__ == "__main__":
    main()