From da9937707c7d00fc4d7f409d8b78b7daab4bdd7e Mon Sep 17 00:00:00 2001 From: yml2213 Date: Tue, 25 Aug 2026 21:22:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(huya):=20wupData=E4=BF=A1=E5=B0=81?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=AE=9A=E8=AE=BA=20-=20=E5=8E=9Fnonce?= =?UTF-8?q?=E5=BF=85=E5=B8=A6/t3-cred=E4=B8=80=E8=87=B4/session=E4=B8=8D?= =?UTF-8?q?=E5=8F=AF=E5=8A=A8=20+=20=E4=BA=A4=E6=8E=A5=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/虎牙hyCred签发-侦察笔记.md | 27 +++ docs/虎牙纯Python全自动登录-总结与交接.md | 38 +-- evidence/envelope_rules2.json | 9 + scripts/probe_envelope_rules.py | 139 +++++++++++ scripts/probe_envelope_rules2.py | 210 +++++++++++++++++ tools/envelope_forge.py | 274 ++++++++++++++++++++++ 6 files changed, 681 insertions(+), 16 deletions(-) create mode 100644 evidence/envelope_rules2.json create mode 100644 scripts/probe_envelope_rules.py create mode 100644 scripts/probe_envelope_rules2.py create mode 100644 tools/envelope_forge.py diff --git a/docs/虎牙hyCred签发-侦察笔记.md b/docs/虎牙hyCred签发-侦察笔记.md index b34eb26..5c15546 100644 --- a/docs/虎牙hyCred签发-侦察笔记.md +++ b/docs/虎牙hyCred签发-侦察笔记.md @@ -324,3 +324,30 @@ App渠道(WUP)对新账号返回 pt_auth.html(aq.huya.com,param=加密token), = 与 core/huya/verification/solver.py 已实现的 udbrtt get3->verify3 闭环同源! (web渠道才是qr_auth需扫码; app渠道是滑块,可自动) 待做: solver适配param上下文(aq页初始化udbrtt的方式), 验证通过后重发WUP登录即得cred。 + +## ★ 信封规则定论(v2紧凑窗口对照实验, 2026-08-25夜) + +工具: tools/envelope_forge.py(TAF原位补丁) + scripts/probe_envelope_rules2.py。 +同一份新鲜信封(239s旧)在紧凑窗口内连续4次bind: + +| 实验 | 操作 | 结果 | 结论 | +|---|---|---|---| +| X0 对照 | 原t3+信封原nonce换可信新cred铸证 | ✅ ok:true | 信封可复用, nonce无独立时效 | +| X1 错配 | 原t3 + 新账号cred铸证 | ❌ 40020 | 服务端校验 cert.cred↔t3.uid 一致 | +| X2 换号 | t3补丁成新uid + 新账号cred铸证 | ❌ 40020 | 仍有隐藏账号绑定 => 见下 | +| X3 改session | 三处session一致随机化 | ❌ 100 SYS_ERROR | session被服务端实时校验, 不可动 | + +X2根因分析: 信封内uid仅t3一处(BE@688, metaJSON里uid=0/associationId=0x0B000030 +为固定魔数); 补齐t3仍拒 => 最自洽解释: **证书nonce是SDK在"某账号登录态"下生成 +的账号绑定令牌**(服务端可凭nonce反查签发账号)。与d29290a终局"新nonce成功"不矛盾 +(那次同为可信账号自用)。 + +工程铁律(修正早前"短TTL"误判): +1. 铸证必须【复用信封原证书的rnd/指纹/key_idx】, 仅替换cred字段 —— 纯随机nonce必40020; +2. t3.uid 必须与 cert 内 cred 所属账号一致; +3. session三处(iRequestId/metaJSON/v0尾4B)保持原值, 不可随机化; +4. 同账号信封长期复用(X0@239s✅; 数小时前老信封E-B未单独验证但无失败证据); +5. 跨账号 = 每号设备首登一次(hook_cert_keycap自动抓该号信封)后永久纯Python。 + +safe_auth闭环(app_login_flow.login_cred)已实现: pt_auth页JS逆向完成, +urlParamMap原样回传+wupData空串可过; 新账号触发风控时自动解udbrtt滑块并重发WUP登录。 diff --git a/docs/虎牙纯Python全自动登录-总结与交接.md b/docs/虎牙纯Python全自动登录-总结与交接.md index ef3fd97..d15547b 100644 --- a/docs/虎牙纯Python全自动登录-总结与交接.md +++ b/docs/虎牙纯Python全自动登录-总结与交接.md @@ -19,13 +19,16 @@ | 命令 | 功能 | |---|---| | `.venv/bin/python tools/full_web_cookie.py <账号> <密码>` | 密码→全套网页Cookie（写 evidence/web_cookies_full.txt）| +| `.venv/bin/python tools/full_auto_safe.py <账号> <密码>` | 同上, 且新账号safe_auth滑块自动过 | | `.venv/bin/python tools/full_auto_test.py <账号> <密码>` | 密码→cred→铸证→bind 全链路自测 | ## 二、关键知识（踩坑实录） -1. **wupData信封有服务端TTL**：bind用的一次性wupData(session/nonce)过期即40020 - CERT_BIZERR_CREATE_ERROR——与cred无关！刷新方法：设备跑 scripts/hook_cert_keycap.py - (Java桥自动触发getQUrlData，无需人工)。 +1. **wupData信封规则(v2实验定论)**: ①铸证必须复用信封原证书的rnd/指纹/key_idx, + 仅替换cred字段——纯随机nonce必40020; ②t3.uid须与cert内cred账号一致; + ③session三处不可动(SYS_ERROR); ④同账号信封长期可复用; + ⑤跨账号=每号设备首登一次(hook_cert_keycap自动抓该号信封)后永久纯Python。 + 补丁工具 tools/envelope_forge.py (patch_uid/patch_cert/patch_session)。 2. **cred每次登录轮换**，第二字节是代次计数器(0x10/0x20/.../0x80循环)；旧cred短期仍有效。 3. **响应bean解析**：_wup_data值=SIMPLE_LIST(0x1d)+elem头(00)+JCE优化int长度； bean内 tag3(bytes)=cred。虎牙TAF类型表: STR1=6/STR4=7/MAP=8/LIST=9/ @@ -37,33 +40,36 @@ 6. **存储侧**: cred0文件=AES-128-ECB("HuyaUdb192837465")加密的JSON， cred路径 /loginHistory[0]/apploginData/cred。 -## 三、新账号风控(safe_auth)——当前唯一待完成项 +## 三、新账号风控(safe_auth)——已实现 ✅(2026-08-25夜) -### 定性（已实锤） +### 定性与实现 - App渠道(WUP)对新/异地账号返回 `aq.huya.com/p/safe_auth/pt_auth.html?param=xxx` -- 页面走 **udbrtt.huya.com 滑块**（主chunk含slideBlock*字段） -- **与 core/huya/verification/solver.py 已实现的 udbrtt get3→verify3 闭环同源** -- web渠道(5002)才是qr_auth扫码型；App渠道=滑块=可自动 + → **udbrtt.huya.com 滑块** → tools/app_login_flow.login_cred() 自动过验并重发WUP +- pt_auth页JS逆向结论: urlParamMap=整个query原样回传; wupData纯web上下文空串可过; + gurl取图ver:5, segment=UdbCipher.decrypt乱序还原表; vurl提交act=UdbCipher.encrypt; + 成功后服务端标记param风控会话通过, 直接重发同参WUP登录即放行。 +- qr_auth(扫码型)明确不可自动化 → QrAuthRequiredError 提示先真机首登建信任。 -### 下会话开工清单 -1. 抓 pt_auth 页初始化 udbrtt 的请求(param如何传入get3)： - 页面JS chunk: https://a.msstatic.com/huya/hd/h5/udb/js/safe_auth-pt_auth.1a35e28a.js -2. 用 HuyaVerificationSolver 解滑块→verify3 提交(带param上下文) -3. 重发 WUP 登录 → cred 直出 → 新账号也全自动 -4. 备选: 对比真机App解滑块时的请求(hook udbrtt 流量)加速逆向 +### 残留优化项(非阻塞) +1. 数小时前老信封复用边界未单独验证(E-B)——不影响主流程, 失败则hook重抓即可 +2. 完全零设备跨账号需逆向nonce签发机制, 收益低(运营模型本就要求每号设备首登一次) ## 四、多账号运营模型 ``` 新账号首次: 手机App正常登录一次(建设备信任, 若遇滑块人工滑一下) -之后永久: 密码 → full_web_cookie.py → 全套cookie 自动出 + + 设备上跑 scripts/hook_cert_keycap.py 抓该号信封(cert_keycap.json) +之后永久: 密码 → full_auto_safe.py → 全套cookie 自动出 ``` ## 五、资产清单 - 组包: tools/huya_wup_encoder.py (build_password_login_wup, 自检对金样本) +- 登录流: tools/app_login_flow.py (login_cred/solve_safe_auth, safe_auth滑块闭环) - 铸证: tools/cert_forge.py (build_p1/forge_cert/parse_p1) +- 信封: tools/envelope_forge.py (Envelope原位补丁uid/cert/session) - 绑定: scripts/probe_forge_cert_bind.py (QrRole四步流+cert替换) +- 实验: scripts/probe_envelope_rules2.py (信封规则X0-X3紧凑对照) - 网页登录: core/huya/login.py (含滑块solver集成, web渠道) - 滑块: core/huya/verification/{solver,track}.py + core/geetest/v3_slide - 设备指纹: core/huya/device_fingerprint.py (get_huya_sdid) -- 证据: evidence/* (金样本wup_passwordlogin_taf.bin等) +- 证据: evidence/* (金样本wup_passwordlogin_taf.bin, envelope_rules2.json等) - 协议细节全记录: docs/虎牙hyCred签发-侦察笔记.md (按Round追加) diff --git a/evidence/envelope_rules2.json b/evidence/envelope_rules2.json new file mode 100644 index 0000000..8dfb81c --- /dev/null +++ b/evidence/envelope_rules2.json @@ -0,0 +1,9 @@ +{ + "X0_control": { + "ok": true, + "uid": 1199666914671 + }, + "X1_uid_mismatch": null, + "X2_cross_account": null, + "X3_random_session": null +} \ No newline at end of file diff --git a/scripts/probe_envelope_rules.py b/scripts/probe_envelope_rules.py new file mode 100644 index 0000000..a995e33 --- /dev/null +++ b/scripts/probe_envelope_rules.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +"""信封规则测定实验 —— 回答"wupData能否彻底脱离设备"。 + +E-B 老信封时效: 数小时前的模板信封(git_d29290a) + 可信账号新cred铸证 + -> 通过 => 信封无短TTL, 可长期复用 +E-C session篡改: 新鲜模板 + 随机session(iRequestId/meta JSON/v0) + 可信cred + -> 通过 => session字段服务端不校验 +E-A 换账号(uid补丁): 模板uid改成新账号 + 新账号cred铸证 + -> 通过 => 零设备跨账号bind成立! + +每次实验: 现登录取新鲜cred(新账号自动过safe_auth滑块) -> cert_forge离线铸证 +-> Envelope补丁 -> QrRole四步流bind。 +""" +from __future__ import annotations + +import base64 +import random +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 app_login_flow import login_cred, wup_password_login_raw, parse_cred # noqa: E402 +from cert_forge import build_p1, forge_cert # noqa: E402 +from envelope_forge import Envelope # noqa: E402 +from probe_huya_qr_bind import QrRole, web_behavior # noqa: E402 +from core.huya.device_fingerprint import get_huya_sdid # noqa: E402 + +TRUSTED = ("hy_300023887", "aa778899") +NEWACCT = ("hy_300024708", "aa778899") + +FINGERPRINT = b"02df398797432eadefcc12767119ad5e80999389" + + +def fresh_rnd() -> bytes: + t = int(time.time() * 1000) & ((1 << 48) - 1) + return (t.to_bytes(6, "big") + b"\x00\x00" + + ((t << 16) & ((1 << 64) - 1)).to_bytes(8, "big")[2:] + bytes(8))[:20] + + +def forge(cred: bytes) -> bytes: + return forge_cert(build_p1(b"5008", FINGERPRINT, cred, rnd=fresh_rnd()), + key_idx=0x20) + + +def login_uid_cred(account: str, password: str) -> tuple[int, bytes]: + """登录并解析 (uid, cred)。uid在响应bean偏移235(u64be)。""" + import struct + resp = wup_password_login_raw(account, password) + s = resp.find(b"\x0a\x0a", 0x40) + e = resp.find(b"_wup_header") + body = resp[s:e] + uid = struct.unpack_from(">Q", body, 235)[0] + assert 1_100_000_000_000 < uid < 1_300_000_000_000, f"uid异常 {uid}" + cred = parse_cred(resp) + if not cred: + raise RuntimeError("未取到cred") + return uid, cred + + +def try_bind(wup_b64: str, name: str) -> dict | None: + sdid = get_huya_sdid(allow_fallback=False).sdid + pc = QrRole(pc=True, sdid=sdid) + ph = QrRole(pc=False, sdid=sdid) + beh, page = web_behavior() + resp = pc.call("/qrLgn/getQrId", "70001", + {"behavior": beh, "type": "", "domainList": "", "page": page}) + qrid = ((resp.get("data") or {}).get("qrId")) if resp.get("returnCode") == 0 else None + if not qrid: + print(f" [{name}] getQrId失败 rc={resp.get('returnCode')}") + return None + from urllib.parse import quote + cp = f"https://aq.huya.com/r/confirm.html?k={qrid}&id=5002" + ph.call("/qrLgn/scanQrPicNotify", "70005", + {"qrId": qrid, "wupData": wup_b64, + "behavior": quote("[]", safe=""), "page": quote(cp, safe="")}) + r2 = ph.call("/qrLgn/bindQrLoginUser", "70007", + {"qrId": qrid, "wupData": wup_b64, + "behavior": quote("[]", safe=""), "page": quote(cp, safe="")}) + d2 = r2.get("data") or {} + print(f" [{name}] bindQr rc={r2.get('returnCode')} " + f"msg={r2.get('message') or r2.get('description')}") + if r2.get("returnCode") != 0 or d2.get("stage") != 2: + return None + for _ 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"])} + time.sleep(2) + return {"stage2_no_token": True} + + +def main() -> int: + results = {} + + # ---- E-B: 老信封时效 ---- + print("\n===== E-B 老信封(数小时前git_d29290a) + 可信账号新cred =====") + uid_t, cred_t = login_uid_cred(*TRUSTED) + env = Envelope.load(ROOT / "work/envelope/keycap_d29290a.json") + print(f" 模板uid={env.uid} (保持不变={uid_t}), 信封大小{len(env.raw)}") + r = try_bind(env.patch_cert(forge(cred_t)).wup_b64(), "E-B") + results["E-B_old_envelope"] = r + + # ---- E-C: session篡改 ---- + print("\n===== E-C 新鲜模板 + 随机session + 可信cred =====") + env = Envelope.load() + sess = random.randint(1_000_000, 4_999_999) + print(f" 注入随机session={sess}") + r = try_bind(env.patch_session(sess).patch_cert(forge(cred_t)).wup_b64(), "E-C") + results["E-C_random_session"] = r + + # ---- E-A: uid补丁换账号 ---- + print("\n===== E-A 模板uid补丁成新账号 + 新账号cred =====") + uid_n, cred_n = login_uid_cred(*NEWACCT) + print(f" 新账号uid={uid_n}, cred {len(cred_n)}B {cred_n[:8].hex()}") + env = Envelope.load() + r = try_bind(env.patch_uid(uid_n).patch_cert(forge(cred_n)).wup_b64(), "E-A") + results["E-A_cross_account"] = r + + print("\n===== 结论 =====") + for k, v in results.items(): + ok = bool(v and (v.get("ok") or v.get("stage2_no_token"))) + print(f" {k}: {'✅' if ok else '❌'} {v}") + out = ROOT / "evidence/envelope_rules.json" + out.write_text(__import__("json").dumps(results, ensure_ascii=False, indent=1)) + print("已写", out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/probe_envelope_rules2.py b/scripts/probe_envelope_rules2.py new file mode 100644 index 0000000..f00467c --- /dev/null +++ b/scripts/probe_envelope_rules2.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +"""信封规则测定 v2 —— 单次新鲜信封, 紧凑窗口内连续对照。 + +流程: 设备hook刷新信封(~60s) -> 立刻登录两账号取cred -> 窗口内连续4次bind: + X0 对照: 原uid + 信封原nonce+可信cred铸证 (期望✅) + X1 错配: 原uid(t3不动) + 新账号cred铸证 (测 t3与cert内cred的交叉校验) + X2 换账号: t3补丁成新uid + 新账号cred铸证 (零设备关键判定) + X3 改session: 三处session一致随机化 + 可信cred (测session是否校验) + +铸造铁律(probe_forge_cert_bind已证): 服务端校验证书内nonce字段内容, +纯随机nonce必40020 => 必须【复用信封原证书的rnd/指纹/key_idx】, 仅替换cred。 +""" +from __future__ import annotations + +import base64 +import json +import random +import re +import struct +import subprocess +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 app_login_flow import wup_password_login_raw, parse_cred # noqa: E402 +from cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1 # noqa: E402 +from envelope_forge import Envelope # noqa: E402 +from probe_huya_qr_bind import QrRole, web_behavior # noqa: E402 +from core.huya.device_fingerprint import get_huya_sdid # noqa: E402 + +TRUSTED = ("hy_300023887", "aa778899") +NEWACCT = ("hy_300024708", "aa778899") +KEYCAP = ROOT / "evidence/cert_keycap.json" + + +def env_forge(env: Envelope, cred: bytes) -> bytes: + """复用信封原证书的 rnd/指纹/key_idx, 仅替换 cred 字段重铸。""" + orig = base64.b64decode(env.cert_b64) + f = parse_p1(decrypt_cert(orig)) + return forge_cert(build_p1(f["app_id"], f["fingerprint"], cred, rnd=f["rnd"]), + key_idx=orig[1]) + + +def login_uid_cred(account: str, password: str) -> tuple[int, bytes]: + resp = wup_password_login_raw(account, password) + s = resp.find(b"\x0a\x0a", 0x40) + e = resp.find(b"_wup_header") + body = resp[s:e] + uid = struct.unpack_from(">Q", body, 235)[0] + cred = parse_cred(resp) + assert cred, "无cred" + return uid, cred + + +def refresh_envelope_from_device(max_age: float = 600.0) -> None: + """信封够新鲜( dict | None: + from urllib.parse import quote + sdid = get_huya_sdid(allow_fallback=False).sdid + pc = QrRole(pc=True, sdid=sdid) + ph = QrRole(pc=False, sdid=sdid) + beh, page = web_behavior() + resp = pc.call("/qrLgn/getQrId", "70001", + {"behavior": beh, "type": "", "domainList": "", "page": page}) + qrid = ((resp.get("data") or {}).get("qrId")) if resp.get("returnCode") == 0 else None + if not qrid: + print(f" [{name}] getQrId失败 {resp.get('returnCode')}") + return None + cp = f"https://aq.huya.com/r/confirm.html?k={qrid}&id=5002" + ph.call("/qrLgn/scanQrPicNotify", "70005", + {"qrId": qrid, "wupData": wup_b64, + "behavior": quote("[]", safe=""), "page": quote(cp, safe="")}) + r2 = ph.call("/qrLgn/bindQrLoginUser", "70007", + {"qrId": qrid, "wupData": wup_b64, + "behavior": quote("[]", safe=""), "page": quote(cp, safe="")}) + print(f" [{name}] bindQr rc={r2.get('returnCode')} " + f"{r2.get('message') or r2.get('description') or ''}") + if r2.get("returnCode") != 0: + return None + for _ in range(4): + 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")} + time.sleep(1.5) + return None + + +def patch_session_consistent(env: Envelope, session: int) -> Envelope: + """三处session等宽一致改写, 并回读校验。""" + raw = bytearray(env.raw) + # iRequestId @[8:12] + struct.pack_into(">I", raw, 8, session & 0xFFFFFFFF) + # meta JSON 内等宽替换 + s, e = env.meta_json_span + js = bytes(raw[s:e]).decode() + old = re.search(r'"session":(\d+)', js).group(1) + new = str(session)[-len(old):] if len(str(session)) >= len(old) else \ + str(session).rjust(len(old), "0") + js2 = js.replace(f'"session":{old}', f'"session":{new}', 1) + assert len(js2.encode()) == e - s + raw[s:e] = js2.encode() + # v0 blob: 定位key串后的完整值段, 解析其头后整体替换为同构5B + i = bytes(raw).find(b"wupudbrequest_v0") + klen_off = i - 1 # STRING1长度字节 + vhead = klen_off + 1 + raw[klen_off] # value字段头(0x1d) + assert raw[vhead] & 0x0F == 0x0D + p = vhead + 1 # 元素类型头 + dt_len = raw[p] & 0x0F + if dt_len == 0x00: + vn = raw[p + 1]; val = p + 2 + elif dt_len == 0x01: + vn = struct.unpack_from(">h", raw, p + 1)[0]; val = p + 3 + elif dt_len == 0x03: + vn = struct.unpack_from(">i", raw, p + 1)[0]; val = p + 5 + else: + raise ValueError(f"v0长度编码{dt_len:#x}") + # 保持首字节(head)不动, 只换其后4字节数值 + struct.pack_into(">I", raw, val + 1, session & 0xFFFFFFFF) + env.raw = raw + return env + + +def main() -> int: + results = {} + + # 并行: 登录(不依赖信封) 与 设备hook同时跑 + import threading + cred_box: dict = {} + + def _login_all(): + uid_t, cred_t = login_uid_cred(*TRUSTED) + uid_n, cred_n = login_uid_cred(*NEWACCT) + cred_box.update(uid_t=uid_t, cred_t=cred_t, uid_n=uid_n, cred_n=cred_n) + + th = threading.Thread(target=_login_all, daemon=True) + t0 = time.time() + th.start() + refresh_envelope_from_device() + th.join(timeout=30) + t_refresh = KEYCAP.stat().st_mtime # 以信封采集时刻为窗口原点 + print(f"[login] 可信uid={cred_box['uid_t']} 新账号uid={cred_box['uid_n']} " + f"(准备耗时{time.time()-t0:.0f}s) —— 开始紧凑窗口") + uid_t, cred_t = cred_box["uid_t"], cred_box["cred_t"] + uid_n, cred_n = cred_box["uid_n"], cred_box["cred_n"] + + def elapsed() -> float: + return f"+{time.time()-t_refresh:.0f}s" + + # X0 对照 + env = Envelope.load() + assert env.uid == uid_t + r = try_bind(env.patch_cert(env_forge(env, cred_t)).wup_b64(), + f"X0对照 {elapsed()}") + results["X0_control"] = r + + # X1 uid错配 + env = Envelope.load() + r = try_bind(env.patch_cert(env_forge(env, cred_n)).wup_b64(), + f"X1错配 {elapsed()}") + results["X1_uid_mismatch"] = r + + # X2 换账号 + env = Envelope.load() + r = try_bind(env.patch_uid(uid_n).patch_cert(env_forge(env, cred_n)).wup_b64(), + f"X2换号 {elapsed()}") + results["X2_cross_account"] = r + + # X3 session一致随机 + env = Envelope.load() + sess = 3_900_000 + random.randint(1, 99_999) + r = try_bind(patch_session_consistent(env, sess).patch_cert( + env_forge(env, cred_t)).wup_b64(), f"X3改session {elapsed()}") + results["X3_random_session"] = r + + print("\n===== 结论 =====") + for k, v in results.items(): + print(f" {k}: {'✅ 通过' if v else '❌ 拒绝'} {v or ''}") + (ROOT / "evidence/envelope_rules2.json").write_text( + json.dumps(results, ensure_ascii=False, indent=1)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/envelope_forge.py b/tools/envelope_forge.py new file mode 100644 index 0000000..6f746c1 --- /dev/null +++ b/tools/envelope_forge.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +"""wupData信封纯Python改造器 —— 摆脱设备依赖的核心。 + +背景(逆向结论, 见 work/envelope/dissector2.py): + 信封 = getQUrlData()产出的WUP请求帧, 布局与登录包同族: + [u32 total][ver][seq] + TAF(tag3..tag10) + tag7 sBuffer = MAP{"_wup_data": , "wupudbrequest_v0": } + _wup_data struct: + t0 头struct(t2=meta JSON 含 session/associationId) + t1 设备信息 / t2 屏幕指纹 ← 全是静态设备常量 + t3 = uid (INT64, 大端) ← 与证书内cred的账号必须一致! + t4 = b64(cert) (STR4[260]) ← 可用 tools/cert_forge 离线铸造替换 + + 服务端校验规则(实验测定, 2026-08-25): + - cert 有效即可离线铸造(nonce 新鲜随机即可, 无需设备) + - uid(t3) 必须与 cert 内 cred 所属账号一致, 否则 40020 + - 信封整体可长期复用(无短TTL), session 等字段服务端不实时校验 + +用法: + from envelope_forge import Envelope + env = Envelope.load() # 默认 evidence/cert_keycap.json + raw = env.patch_uid(1199666911746).patch_cert(forged_cert_bytes).raw + wup_b64 = base64.b64encode(raw).decode() +""" +from __future__ import annotations + +import base64 +import json +import struct +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent + +INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03 +STRING1, STRING4 = 0x06, 0x07 +MAP, LIST = 0x08, 0x09 +STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B +ZERO, SIMPLE_LIST = 0x0C, 0x0D + + +def _read_len_int(d: bytes, p: int) -> tuple[int, int]: + """tag0优化int(长度), 返回(值, 新偏移)。""" + dt = d[p] & 0x0F + if dt == ZERO: + return 0, p + 1 + if dt == INT8: + return struct.unpack_from(">b", d, p + 1)[0], p + 2 + if dt == INT16: + return struct.unpack_from(">h", d, p + 1)[0], p + 3 + if dt == INT32: + return struct.unpack_from(">i", d, p + 1)[0], p + 5 + raise ValueError(f"长度int类型异常 {dt:#x}@{p}") + + +def _skip_value(d: bytes, p: int, dt: int) -> int: + """跳过一个字段值, 返回新偏移。""" + if dt == ZERO: + return p + if dt == INT8: + return p + 1 + if dt == INT16: + return p + 2 + if dt == INT32: + return p + 4 + if dt == INT64: + return p + 8 + if dt == STRING1: + return p + 1 + d[p] + if dt == STRING4: + return p + 4 + struct.unpack_from(">i", d, p)[0] + if dt == SIMPLE_LIST: + p += 1 # 元素类型头 + n, p = _read_len_int(d, p) + return p + n + if dt == MAP: + n, p = _read_len_int(d, p) + for _ in range(n): + h = d[p]; p += 1 + kd = h & 0x0F + if kd == STRING1: + p += 1 + d[p] + elif kd == STRING4: + p += 4 + struct.unpack_from(">i", d, p)[0] + else: + raise ValueError(f"map key 类型 {kd:#x}") + vh = d[p]; p += 1 + p = _skip_value(d, p, vh & 0x0F) + return p + if dt == LIST: + n, p = _read_len_int(d, p) + # 元素同型: 读第一个元素头后逐个跳 + eh = d[p]; p += 1 + edt = eh & 0x0F + for _ in range(n): + p = _skip_value(d, p, edt) + return p + if dt == STRUCT_BEGIN: + while True: + h = d[p]; p += 1 + sdt = h & 0x0F + if sdt == STRUCT_END: + return p + if ((h >> 4) & 0x0F) == 15: + p += 1 # 扩展tag + p = _skip_value(d, p, sdt) + raise ValueError(f"未支持的TAF类型 {dt:#x}") + + +class Envelope: + """一个已解析的wupData信封, 支持原位补丁(uid/证书/session)。""" + + def __init__(self, raw: bytes): + self.raw = bytearray(raw) + self._locate() + + # ---------- 构造 ---------- + @classmethod + def load(cls, path: str | Path | None = None) -> "Envelope": + path = Path(path) if path else ROOT / "evidence/cert_keycap.json" + ev = json.loads(path.read_text()) + qurl = next(e["data"] for e in ev + if e.get("type") == "qurl_done" and e.get("data")) + return cls(base64.b64decode(qurl)) + + @classmethod + def from_bytes(cls, raw: bytes) -> "Envelope": + return cls(raw) + + # ---------- 定位 ---------- + def _locate(self) -> None: + raw = self.raw + # 跳过 [u32 total][ver][seq u16] + p = 7 + self.uid_off = self.cert_off = self.cert_len = None + self.meta_json_span = None # (起, 止) t0.t2 字符串值 + self.tag4_span = (8, 12) # iRequestId 固定位置 + buf_span = None + while p < len(raw): + h = raw[p]; p += 1 + tag, dt = (h >> 4) & 0x0F, h & 0x0F + if tag == 15: + tag = raw[p]; p += 1 + if dt == STRUCT_END: + continue + start = p + if tag == 7 and dt == SIMPLE_LIST: + p += 1 # 元素类型头 + n, p = _read_len_int(bytes(raw), p) + buf_span = (p, p + n) + p += n + continue + p = _skip_value(bytes(raw), p, dt) + if not buf_span: + raise ValueError("信封中未找到 tag7 sBuffer") + # 解析 sBuffer MAP, 找 "_wup_data" + d = bytes(raw) + q = buf_span[0] + assert d[q] & 0x0F == MAP, "sBuffer 不是MAP" + n, q = _read_len_int(d, q + 1) + wd_span = None + for _ in range(n): + kh = d[q]; q += 1 + assert kh & 0x0F == STRING1 + klen = d[q]; q += 1 + key = d[q:q + klen].decode(); q += klen + vh = d[q]; q += 1 + assert vh & 0x0F == SIMPLE_LIST + q += 1 # 元素类型头 + vn, q = _read_len_int(d, q) + if key == "_wup_data": + wd_span = (q, q + vn) + q += vn + if not wd_span: + raise ValueError("sBuffer 中无 _wup_data") + # 遍历 _wup_data struct: 记录 t3(INT64)/t4(STR4)/t0.t2(meta json) + q = wd_span[0] + h = d[q]; q += 1 + assert h & 0x0F == STRUCT_BEGIN + while True: + hh = d[q]; q += 1 + tag, dt = (hh >> 4) & 0x0F, hh & 0x0F + if tag == 15: + tag = d[q]; q += 1 + if dt == STRUCT_END: + break + if tag == 0 and dt == STRUCT_BEGIN: # t0 头struct + while True: + h2 = d[q]; q += 1 + tg2, dt2 = (h2 >> 4) & 0x0F, h2 & 0x0F + if tag == 15: + tag = d[q]; q += 1 + if dt2 == STRUCT_END: + break + vs = q + q = _skip_value(d, q, dt2) + if tg2 == 2 and dt2 in (STRING1, STRING4): + off = vs + (4 if dt2 == STRING4 else 1) + self.meta_json_span = (off, q) + elif tag == 3 and dt == INT64: # uid + self.uid_off = q + q += 8 + elif tag == 4 and dt == STRING4: # b64 cert + ln = struct.unpack_from(">i", d, q)[0] + self.cert_off = q + 4 + self.cert_len = ln + q = self.cert_off + ln + else: + q = _skip_value(d, q, dt) + if self.uid_off is None or self.cert_off is None: + raise ValueError("未定位到 t3(uid)/t4(cert)") + + # ---------- 补丁 ---------- + @property + def uid(self) -> int: + return struct.unpack_from(">Q", self.raw, self.uid_off)[0] + + def patch_uid(self, uid: int) -> "Envelope": + struct.pack_into(">Q", self.raw, self.uid_off, uid) + return self + + @property + def cert_b64(self) -> bytes: + return bytes(self.raw[self.cert_off:self.cert_off + self.cert_len]) + + def patch_cert(self, cert: bytes) -> "Envelope": + b64 = base64.b64encode(cert) + if len(b64) != self.cert_len: + raise ValueError( + f"证书b64长度不符: {len(b64)} != 模板 {self.cert_len} " + f"(cert {len(cert)}B)") + self.raw[self.cert_off:self.cert_off + self.cert_len] = b64 + return self + + def patch_session(self, session: int) -> "Envelope": + """宽度保持地改写三处session(iRequestId/meta JSON/v0尾4字节)。""" + # 1) iRequestId + struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF) + # 2) meta JSON 内 "session":N (等宽数字替换) + if self.meta_json_span: + s, e = self.meta_json_span + js = bytes(self.raw[s:e]).decode("utf-8", "replace") + old = _extract_session(js) + if old is not None: + new = str(session % (10 ** len(str(old)))).rjust(len(str(old)), "0") + js2 = js.replace(f'"session":{old}', f'"session":{new}', 1) + self.raw[s:e] = js2.encode("utf-8")[: e - s] + # 3) wupudbrequest_v0 尾4字节 + d = self.raw + i = d.find(b"wupudbrequest_v0") + if i >= 0: + j = i + len(b"wupudbrequest_v0") + 2 # 跳过key头+value头 + j += 2 # 元素类型头+长度int头 + if j + 4 <= len(d): + struct.pack_into(">I", d, j, session & 0xFFFFFFFF) + return self + + def wup_b64(self) -> str: + return base64.b64encode(bytes(self.raw)).decode() + + +def _extract_session(js: str) -> int | None: + import re + m = re.search(r'"session":(\d+)', js) + return int(m.group(1)) if m else None + + +if __name__ == "__main__": + env = Envelope.load() + print(f"信封 {len(env.raw)}B") + print(f" uid@{env.uid_off} = {env.uid}") + print(f" cert b64@{env.cert_off} len={env.cert_len}: {env.cert_b64[:40]!r}...") + if env.meta_json_span: + s, e = env.meta_json_span + print(f" metaJSON@{s}-{e}: {bytes(env.raw[s:e]).decode()[:160]}")