From 9aea9851bd098fd17010860dd0a6fe27e2a352f3 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Tue, 25 Aug 2026 11:51:01 +0800 Subject: [PATCH] =?UTF-8?q?test(huya):=20=E9=A3=8E=E6=8E=A7=E5=9C=BA?= =?UTF-8?q?=E6=99=AF/=E4=BB=A3=E7=90=86/=E6=89=AB=E7=A0=81=E9=97=AD?= =?UTF-8?q?=E7=8E=AF=E6=8E=A2=E9=92=88=E4=B8=8E=E9=80=86=E5=90=91=E6=96=87?= =?UTF-8?q?=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - probe_huya_login_scene: 各 appId/UA 组合的风控策略探测 - probe_huya_proxy: 走代理源对比直连的风控策略 - probe_huya_qr_bind: 扫码登录闭环(getQrId->scanQrPicNotify-> bindQrLoginUser->tryQrLogin), 实测出有效web会话; 含wupData 结构地图与证书定位实验结论(见 docs/虎牙扫码登录闭环逆向.md) --- docs/虎牙扫码登录闭环逆向.md | 79 +++++++ scripts/probe_huya_login_scene.py | 105 +++++++++ scripts/probe_huya_proxy.py | 86 +++++++ scripts/probe_huya_qr_bind.py | 362 ++++++++++++++++++++++++++++++ 4 files changed, 632 insertions(+) create mode 100644 docs/虎牙扫码登录闭环逆向.md create mode 100644 scripts/probe_huya_login_scene.py create mode 100644 scripts/probe_huya_proxy.py create mode 100644 scripts/probe_huya_qr_bind.py diff --git a/docs/虎牙扫码登录闭环逆向.md b/docs/虎牙扫码登录闭环逆向.md new file mode 100644 index 0000000..ecccc12 --- /dev/null +++ b/docs/虎牙扫码登录闭环逆向.md @@ -0,0 +1,79 @@ +# 虎牙扫码登录闭环逆向（2026-08-25） + +数据来源：用户抓包 `~/Desktop/抓包/hy/8.24`（手机扫码登录-1 / -手机端2 / -pc端2.har） +探针脚本：`scripts/probe_huya_qr_bind.py` + +## 一、协议链（qrLgn JSON 协议，纯 HTTPS，无原生 WUP 套接字） + +``` +PC角色(appId=5002, sign=1ce3bf68..., version=2.6, context WB-xxx) + ① POST /qrLgn/getQrId uri=70001 -> data.qrId + ⑥ POST /qrLgn/tryQrLogin uri=70003 轮询 stage: 0->1->2 + stage=2 时返回 uid + biztoken(344B) => 构造 cookie 即登录成功 + +手机角色(appId=5131, sign=0ba67962ab9e12387648efeae2750777, version=2.5, context H5-xxx) + ② GET https://aq.huya.com/r/confirm.html?k=&id=5002 (App WebView 内) + ③ POST /qrLgn/scanQrPicNotify uri=70005 data={qrId,wupData,...} -> stage=1 + 登录方信息 + ④ POST /qrLgn/bindQrLoginUser uri=70007 data={qrId,wupData,...} -> stage=2 +``` + +注：`-1.har` 里还有旧版协议 `udbrtt.huya.com/auth/qr/notify|verify`（confirm 页 s/confirm.html）， +新流程走 qrLgn 直连，二者 wupData 结构相同。 + +## 二、web 会话构造（已验证 rc=0） + +```python +cookies = { + 'udb_uid': str(uid), 'yyuid': str(uid), + 'udb_passport': 'hy_xxx', 'username': 'hy_xxx', + 'udb_version': '1.0', 'udb_biztoken': , + 'udb_origin': '1', 'udb_status': '1', +} +# 验证: udbreg /web/v2/modifyPswByCheck 无cookie/垃圾token -> rc=2005 USER_NO_LOGIN +# 扫码会话最小cookie -> rc=0 ★（不需要 udb_cred/udb_deviceid/udb_other） +``` + +## 三、wupData 结构地图（998B，base64 后随请求提交） + +| offset | 内容 | 服务端校验 | +|---|---|---| +| 15 | servant `huyaudbwebui` func `default` | - | +| 70-273 | meta JSON `{associationId:184549392,id:184549392,session,traceId,type:2,uid:0,...}` | ❌ session/traceId 乱改仍通过 | +| 274 | `"5008"` | - | +| 282-463 | blob1 181B（与 safedeviceid 同前缀 `PQwemAN9...`） | ❌ 整段替换金标准 safedeviceid 仍通过 | +| 473-506 | hdid `ed0db8334cadd236c00cadf7e11ab5a5`（=金标准） | 未单独测 | +| 508-686 | 设备信息 13.4.22 / 1.0.801386 / 127.0.0.1 / xiaomi / M2102J2SC+指纹hash / android / 1080x2120 / device_id hash | 未单独测 | +| **701-962** | **尾部 blob 261B —— 登录证书** | ✅ 破坏→`rc=40020 CERT_BIZERR_CREATE_ERROR`；清空→`rc=100 SYS_ERROR`；缺失→`rc=101 PARAM_ERROR` | + +## 四、关键实验结论 + +1. **身份完全由尾部证书 blob 决定**： + - 无 cookie / 垃圾 cookie / 真·APP WUP token 当 `udb_biztoken` → 全部无视， + 绑定结果始终 = 抓包手机上登录的账号（hy_300023887 uid=1199666914671）。 + - 先用 WUP 把 hy_299961311 登到同一设备身份再扫码 → 绑定仍是 hy_300023887 + （服务端不做"设备最新登录"映射）。 +2. **整个 web 登录不需要密码**：getQrId → bind(wupData) → tryQrLogin 三步即出会话。 +3. **证书可复用、可并发**：同一 wupData 已复用 10+ 次（跨 ~1h），5 路并发 2s 全成功。 +4. APP WUP 密码登录（tools/huya_wup_encoder.py）对任意账号可用， + 其响应中的 token(346B) 与 web biztoken(344B) 格式同族（解码后均以 `01 00` 开头）， + 但**不能**直接互换使用。 + +## 五、剩余缺口：每账号一张证书 + +尾部 blob 是与"App 内已登录会话"绑定的证书，生成逻辑在 App 原生 SDK +（confirm.html 会先调 `udbdf.huya.com/dckey/check`，请求体本身加密）。 + +可行路径： +- **A. 每账号一次种子采集**：手机登录某号 → 触发一次任意 udb H5 confirm 流程 → + mitmproxy 抓 `bindQrLoginUser` 的 wupData 存库 → 该账号永久全自动。 + （批量成本 = 每号一次人工，之后零人工） +- **B. 逆向证书生成算法**：dckey/check 协议 + 原生加密，周期长不确定。 +- 待观察：证书有效期（当前样本 >1h 仍有效）。 + +## 六、复现命令 + +```bash +.venv/bin/python scripts/probe_huya_qr_bind.py <账号> <密码> \ + --har "/Users/yml/Desktop/抓包/hy/8.24/手机扫码登录-手机端2.har" +# 成功记录: evidence/qr_login_success.json（含有效 biztoken，已 gitignore） +``` diff --git a/scripts/probe_huya_login_scene.py b/scripts/probe_huya_login_scene.py new file mode 100644 index 0000000..e30cc7a --- /dev/null +++ b/scripts/probe_huya_login_scene.py @@ -0,0 +1,105 @@ +"""探测虎牙密码登录在不同 appId/UA 组合下的风控策略。 + +只走到「触发风控」这一步，打印风控策略（strategy 值 + data URL 路径）， +不真正提交滑块/短信/扫码验证，也不写库。 + +用法: + .venv/bin/python scripts/probe_huya_login_scene.py <账号> <明文密码> [--app-id 5002|5008] [--ua desktop|android] + +会依次测试多组组合，对比风控返回的是: + qr_auth -> 扫码验证（无法自动化） + dx_auth -> 短信验证（无法自动化） + pt_auth / get3 / slider -> 滑块/点选（可自动化） +""" +from __future__ import annotations + +import argparse +import sys + +from core.huya.login import ( + APP_ID, + APP_ID_MOBILE, + APP_UA_MOBILE, + DEFAULT_UA, + HuyaPasswordLogin, +) + + +def classify_strategy(payload: dict) -> str: + """从登录响应里识别风控策略类型。""" + data = payload.get("data") or {} + strategies = data.get("strategys") or [] + if not strategies: + return "unknown" + urls = [str(s.get("data") or "") for s in strategies] + joined = " ".join(urls).lower() + strategy_values = [str(s.get("strategy")) for s in strategies] + if "qr_auth" in joined or "64" in strategy_values: + return "qr_auth(扫码)" + if "dx_auth" in joined: + return "dx_auth(短信)" + if "pt_auth" in joined or "get3" in joined or "slider" in joined: + return "pt_auth(滑块)" + return f"其他(strategy={','.join(strategy_values)})" + + +def probe(username: str, password: str, app_id: str, ua: str | None = None) -> dict: + login = HuyaPasswordLogin(username=username, password=password, app_id=app_id, ua=ua) + ua_label = login.ua[:40] + print(f"\n=== app_id={app_id} | app_sign={login.app_sign[:8]} | ua={ua_label}... ===") + login.prepare_device() + login.init_udb_middle() + payload = login._password_login_once() + + return_code = int(payload.get("returnCode") or 0) + print(f"returnCode={return_code}") + if return_code == 0: + print(" → 登录成功（无需验证）") + return payload + if return_code in (10030, 10039): + kind = classify_strategy(payload) + print(f" → 风控策略: {kind}") + data = payload.get("data") or {} + for s in data.get("strategys") or []: + print(f" strategy={s.get('strategy')} dataType={s.get('dataType')}") + url = str(s.get("data") or "") + print(f" data_url={url.split('?')[0]}") + return payload + print(f" message={payload.get('message')} desc={payload.get('description')}") + return payload + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("username") + parser.add_argument("password") + parser.add_argument("--app-id", default=None, help="只测单个 appId") + parser.add_argument("--ua", default=None, choices=["desktop", "android"], help="只测单种 UA") + args = parser.parse_args() + + combos = [] + app_ids = [args.app_id] if args.app_id else [APP_ID, APP_ID_MOBILE] + for app_id in app_ids: + if args.ua == "desktop": + combos.append((app_id, DEFAULT_UA)) + elif args.ua == "android": + combos.append((app_id, APP_UA_MOBILE)) + else: + # 默认每组都测；5002 也测 Android UA（验证设备画像自洽假设） + if app_id == APP_ID_MOBILE: + combos.append((app_id, None)) # None -> 自动选 Android UA + else: + combos.append((app_id, None)) # None -> 桌面 UA + combos.append((app_id, APP_UA_MOBILE)) # 强制 Android UA + + for app_id, ua in combos: + try: + probe(args.username, args.password, app_id, ua) + except Exception as exc: # noqa: BLE001 + print(f" !! 探测异常: {type(exc).__name__}: {exc}") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_huya_proxy.py b/scripts/probe_huya_proxy.py new file mode 100644 index 0000000..6b22a58 --- /dev/null +++ b/scripts/probe_huya_proxy.py @@ -0,0 +1,86 @@ +"""用代理池实测虎牙密码登录：验证换 IP 能否把风控从扫码降回滑块。 + +用法: + .venv/bin/python scripts/probe_huya_proxy.py <账号> <明文密码> --proxy-api <代理API URL> + +代理 API 返回格式支持: + - JSON: {"code":0, "data":[{"IP":"x","Port":y}, ...]} 或 {"code":0, "data":[{"ip":"x","port":y}]} + - 纯文本: 多行 "ip:port" 或 "ip,port" + +流程: + 1. 直连（当前出口 IP）走一次登录，记录风控策略 + 2. 从代理 API 取代理，换 IP 走登录，记录每个 IP 的风控策略 + 3. 汇总对比 + +不会写库，不会提交验证码（只走到触发风控）。 +""" +from __future__ import annotations + +import argparse +import sys + +from core.douyu.proxy_fetcher import ProxyFetcher +from core.huya.login import HuyaPasswordLogin + + +def classify(payload: dict) -> str: + data = payload.get("data") or {} + strategies = data.get("strategys") or [] + joined = " ".join(str(s.get("data") or "") for s in strategies).lower() + strategy_values = [str(s.get("strategy")) for s in strategies] + if "qr_auth" in joined or "64" in strategy_values: + return "扫码(qr_auth)" + if "dx_auth" in joined: + return "短信(dx_auth)" + if "pt_auth" in joined or "get3" in joined: + return "滑块(pt_auth)" + return f"其他(strategy={','.join(strategy_values) or '无'})" + + +def one_pass(username: str, password: str, proxies=None, label: str = "") -> dict: + login = HuyaPasswordLogin(username=username, password=password, proxies=proxies) + login.prepare_device() + login.init_udb_middle() + payload = login._password_login_once() + rc = int(payload.get("returnCode") or 0) + if rc == 0: + kind = "成功(无需验证)" + elif rc in (10030, 10039): + kind = classify(payload) + else: + kind = f"code={rc} message={payload.get('message')}" + print(f" [{label or '直连'}] returnCode={rc} -> {kind}") + return payload + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("username") + parser.add_argument("password") + parser.add_argument("--proxy-api", required=True, help="代理提取 API URL") + parser.add_argument("--max-proxies", type=int, default=3, help="最多测几个代理 IP") + args = parser.parse_args() + + print(f"=== 账号 {args.username} ===") + print("1) 直连基线:") + one_pass(args.username, args.password, label="直连") + + print(f"\n2) 代理池测试 (最多 {args.max_proxies} 个 IP):") + fetcher = ProxyFetcher(api_url=args.proxy_api) + for i in range(args.max_proxies): + proxy = fetcher.fetch_new_proxy() + if not proxy: + print(" 代理耗尽/获取失败，停止") + break + proxies = {"http": proxy, "https": proxy} + print(f" --- 代理 #{i + 1}: {proxy} ---") + try: + one_pass(args.username, args.password, proxies=proxies, label=f"IP#{i + 1}") + except Exception as exc: # noqa: BLE001 + print(f" !! 异常: {type(exc).__name__}: {exc}") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/probe_huya_qr_bind.py b/scripts/probe_huya_qr_bind.py new file mode 100644 index 0000000..8187143 --- /dev/null +++ b/scripts/probe_huya_qr_bind.py @@ -0,0 +1,362 @@ +#!/usr/bin/env python3 +"""虎牙扫码登录闭环探针：APP 协议登录态 + 自演手机端确认。 + +流程（依据 2026-08-25 用户抓包 /Users/yml/Desktop/抓包/hy/8.24）： + 1. APP WUP 密码登录（tools/huya_wup_encoder.py 动态构造） + POST https://udblgn.huya.com/open/hy/passwordLogin -> uid + token + 2. token 作为 Cookie udb_biztoken（App WebView 的登录态载体） + 3. PC 角色：POST /qrLgn/getQrId (uri 70001, appId 5002) -> qrId + 4. 手机角色：POST /qrLgn/scanQrPicNotify (uri 70005, appId 5131) -> stage 1 + 5. 手机角色：POST /qrLgn/bindQrLoginUser (uri 70007, appId 5131) -> stage 2 + 6. PC 角色：POST /qrLgn/tryQrLogin 轮询 (uri 70003) -> stage 2 + uid + biztoken + +用法: + .venv/bin/python scripts/probe_huya_qr_bind.py <账号> <密码> [--har 抓包文件] + --har 提供 8.24 抓包时，会提取其中的 wupData 用于重放对比实验。 +""" +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import random +import re +import sys +import time +import uuid +from pathlib import Path +from urllib.parse import quote, urlencode + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools")) + +import requests + +from huya_wup_encoder import build_password_login_wup, make_trace_id # noqa: E402 +from core.huya.device_fingerprint import get_huya_sdid # noqa: E402 + +BASE = "https://udblgn.huya.com" +APP_SIGN_WEB = "1ce3bf682483d03f146f58232ec10635" # appId 5002 (web) +APP_SIGN_H5 = "0ba67962ab9e12387648efeae2750777" # appId 5131 (App WebView H5) + +UA_PC = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36" +) +UA_PHONE = ( + "Mozilla/5.0 (Linux; Android 11; M2102J2SC Build/RKQ1.200826.002; wv) " + "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/149.0.0.0 " + "Mobile Safari/537.36" +) + +GOLDEN_DEVICE = { + "app_version": "13.4.22", + "sdk_version": "1.0.80138", + "vendor": "xiaomi", + "model": "M2102J2SC", + "os": "android", + "ip": "127.0.0.1", + "fingerprint": "02df398797432eadefcc12767119ad5e80999389", + "screen": "M2102J2SC,30,11", + "width": "1080", + "height": "2120", + "device_id": "7c5387e0539c023c31c4ff0e807e7256117385ee", +} +GOLDEN_SAFEDEVICEID = ( + "PQwemAN9NHkZKoMqVTFUZBIypqMTaQEOrmXr37xQVhQZqrL/gUKEQ11xvE0ju48V8O/" + "t9UBGSp27m4+6bP4IiAEnpaR5Rj1kHEfN2SPLPqYZW9vroxUSoAvjJn6ezTP9jWGxxlRDCbt" + "Py4Rd6MencYT/pNImVIWK+YbNKZt1O05bHUFhqHf3" +) +GOLDEN_HDID = "ed0db8334cadd236c00cadf7e11ab5a5" + + +def log(msg: str) -> None: + print(msg, flush=True) + + +# --------------------------------------------------------------------------- +# 第一步：APP WUP 密码登录 -> token +# --------------------------------------------------------------------------- + +def app_password_login(account: str, password: str) -> dict: + """纯 Python 复刻 App 密码登录，返回 {uid,name,mask,token}。""" + uid_str = account[2:] if account.startswith("hy_") else account + sha1_pwd = hashlib.sha1(password.encode()).hexdigest() + session = random.randint(3_100_000, 3_400_000) + trace_id = make_trace_id() + user_action = json.dumps( + { + "curl": "登录页", "furl": "我的", "latitude": "-1.0", "longitude": "-1.0", + "ssid": "", + "user_action": [ + {"id": "24", "time": str(int(time.time() * 1000) - 500), + "x": str(random.randint(50, 300)), "y": str(random.randint(800, 1100))}, + {"id": "11", "time": str(int(time.time() * 1000)), + "x": str(random.randint(50, 300)), "y": str(random.randint(800, 1100))}, + ], + }, + ensure_ascii=False, + ) + body = build_password_login_wup( + uid_str=uid_str, + sha1_password=sha1_pwd, + safedeviceid=GOLDEN_SAFEDEVICEID, + hdid=GOLDEN_HDID, + session=session, + trace_id=trace_id, + user_action_json=user_action, + device_info=GOLDEN_DEVICE, + ) + s = requests.Session() + s.trust_env = False + r = s.post( + f"{BASE}/open/hy/passwordLogin", + data=body, + headers={ + "Content-Type": "application/x-wup", + "User-Agent": UA_PHONE, + "Host": "udblgn.huya.com", + }, + timeout=20, + ) + log(f"[app-login] HTTP {r.status_code}, body {len(r.content)}B") + if r.status_code != 200 or len(r.content) < 64: + raise RuntimeError(f"WUP 登录失败: {r.status_code} {r.content[:120]!r}") + # 响应为 TAF 二进制，token 是其中最长的 base64url 风格字符串 + m = max(re.finditer(rb"[A-Za-z0-9_\-]{300,}", r.content), + key=lambda x: len(x.group()), default=None) + if m is None: + raise RuntimeError(f"响应中未找到 token: {r.content[:160]!r}") + token = m.group().decode() + # uid: 紧跟 name 字符串 hy_xxx 附近；直接找 hy_ + mn = re.search(rb"hy_(\d{6,})", r.content) + mask_m = re.search(rb"00\*+\d{4}", r.content) + return { + "uid_str": uid_str, + "name": mn.group(0).decode() if mn else account, + "uid": int(mn.group(1)) if mn else None, + "mask": mask_m.group(0).decode() if mask_m else "?", + "token": token, + } + + +# --------------------------------------------------------------------------- +# token -> udb_biztoken 候选变体 +# --------------------------------------------------------------------------- + +def biztoken_variants(token: str) -> dict[str, str]: + v = {"as-is": token} + v["strip2"] = token[2:] + try: + pad = lambda s: s + "=" * (-len(s) % 4) + raw = base64.urlsafe_b64decode(pad(token.replace("-", "_").replace("+", "/") )) + pass + except Exception: + pass + try: + std = token.replace("-", "+").replace("_", "/") + raw = base64.b64decode(std + "=" * (-len(std) % 4)) + v["b64-strip2"] = base64.urlsafe_b64encode(raw[2:]).decode().rstrip("=") + except Exception: + pass + return v + + +# --------------------------------------------------------------------------- +# qrLgn JSON 协议两个角色 +# --------------------------------------------------------------------------- + +class QrRole: + """udblgn.huya.com/qrLgn/* 的 JSON 协议封装。pc=True 走 web(5002)，否则 H5(5131)。""" + + def __init__(self, pc: bool, sdid: str): + self.pc = pc + self.sdid = sdid + self.s = requests.Session() + self.s.trust_env = False + ctx_hex = uuid.uuid4().hex + tail = ("CBC9F93DBEB000011DB11EC02A401BAD-" if pc + else uuid.uuid4().hex.upper()) + tail = tail if tail.endswith("-") else tail + "-" + prefix = "WB" if pc else "H5" + self.context = f"{prefix}-{ctx_hex}-{tail}" + self.page_id = random.randint(40_000_000, 41_000_000) + self.req_counter = random.randint(40_000_000, 41_000_000) + self.s.headers.update({ + "User-Agent": UA_PC if pc else UA_PHONE, + "Origin": BASE, + "content-type": "application/json;charset=UTF-8", + "Accept": "*/*", + }) + + def _headers(self, uri: str) -> dict: + mid = "2.6" if self.pc else "2.5" + return { + "context": self.context, + "uri": uri, + "reqid": str(self.req_counter), + "lcid": "2052", + "Referer": f"{BASE}/web/middle/{mid}/{self.page_id}/https/" + f"{self.context.split('-')[1]}", + } + + def call(self, path: str, uri: str, data: dict, cookies: dict | None = None) -> dict: + envelope = { + "uri": uri, + "version": "2.6" if self.pc else "2.5", + "context": self.context, + "appId": "5002" if self.pc else "5131", + "appSign": APP_SIGN_WEB if self.pc else APP_SIGN_H5, + "authId": "", + "sdid": self.sdid, + "lcid": "2052", + "byPass": "3", + "requestId": str(self.req_counter), + "data": data, + } + self.req_counter += random.randint(120, 400) + r = self.s.post(f"{BASE}{path}", json=envelope, + headers=self._headers(uri), cookies=cookies, timeout=20) + try: + return r.json() + except Exception: + return {"returnCode": -1, "httpStatus": r.status_code, + "description": r.text[:200]} + + +def web_behavior() -> tuple[str, str]: + b = urlencode({"furl": "https://www.huya.com/g", + "curl": "https://www.wuya.com/g".replace("wuya", "huya"), + "user_action": []}, quote_via=quote) + p = quote("https://www.huya.com/g", safe="") + return b, p + + +def extract_replay_wupdata(har_path: str | None) -> str | None: + if not har_path: + return None + with open(har_path) as fp: + d = json.load(fp) + for e in d.get("log", {}).get("entries", []): + req = e.get("request", {}) + u = req.get("url", "") + if "bindQrLoginUser" in u: + body = json.loads(req.get("postData", {}).get("text", "{}")) + wd = body.get("data", {}).get("wupData", "") + if wd: + return wd + return None + + +# --------------------------------------------------------------------------- +# 主流程 +# --------------------------------------------------------------------------- + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("account") + ap.add_argument("password") + ap.add_argument("--har", default=None, help="8.24 抓包 har，用于提取重放 wupData") + ap.add_argument("--biz", default="auto", + choices=["auto", "as-is", "strip2", "b64-strip2"], + help="token->udb_biztoken 转换方式") + args = ap.parse_args() + + log("== 步骤1: APP WUP 密码登录 ==") + info = app_password_login(args.account, args.password) + log(f" uid={info['uid']} name={info['name']} 掩码={info['mask']}") + log(f" token 长度={len(info['token'])} 前24={info['token'][:24]}") + + variants = biztoken_variants(info["token"]) + if args.biz != "auto": + variants = {args.biz: variants[args.biz]} + + replay_wup = extract_replay_wupdata(args.har) + log(f"== 重放 wupData {'可用(' + str(len(replay_wup)) + 'B)' if replay_wup else '未提供'} ==\n") + + pc_sdid_r = get_huya_sdid(allow_fallback=False) + ph_sdid_r = get_huya_sdid(allow_fallback=False) + pc = QrRole(pc=True, sdid=pc_sdid_r.sdid) + ph = QrRole(pc=False, sdid=ph_sdid_r.sdid) + beh, page = web_behavior() + confirm_page = quote( + f"https://aq.huya.com/r/confirm.html?k={{QRID}}&id=5002", safe="") + + for bname, biz in variants.items(): + log(f"\n======== 变体 [{bname}] biztoken 前20={biz[:20]} ========") + cookies = {"udb_biztoken": biz} + + # --- PC: getQrId --- + 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"[pc getQrId] rc={rc} qrId={qrid} desc={resp.get('description')}") + if not qrid: + continue + + # --- 手机角色两种 wupData 各试一次 --- + wup_candidates = {"empty": ""} if not replay_wup else \ + {"empty": "", "replay": replay_wup} + bound = False + for wname, wup in wup_candidates.items(): + cpage = confirm_page.replace("{QRID}", qrid) + r1 = ph.call("/qrLgn/scanQrPicNotify", "70005", { + "qrId": qrid, "wupData": wup, + "behavior": quote(f'[{{"codSDK.{int(time.time()*1000)}}}":"28800.002"}}]' + .replace("}}]", "}]"), safe=""), + "page": cpage}, cookies=cookies) + st1 = (r1.get("data") or {}).get("stage") + log(f"[ph scanQr({wname})] rc={r1.get('returnCode')} stage={st1} " + f"msg={r1.get('message')} desc={r1.get('description')} " + f"data={json.dumps(r1.get('data'), ensure_ascii=False)[:160]}") + if r1.get("returnCode") != 0: + continue + + r2 = ph.call("/qrLgn/bindQrLoginUser", "70007", { + "qrId": qrid, "wupData": wup, + "behavior": quote("[]", safe=""), + "page": cpage}, cookies=cookies) + st2 = (r2.get("data") or {}).get("stage") + log(f"[ph bindQr({wname})] rc={r2.get('returnCode')} stage={st2} " + f"msg={r2.get('message')} desc={r2.get('description')}") + if r2.get("returnCode") != 0 or st2 != 2: + continue + + # --- PC 轮询 tryQrLogin --- + for i in range(12): + rt = pc.call("/qrLgn/tryQrLogin", "70003", { + "qrId": qrid, "remember": "1", "domainList": "", + "behavior": beh, "page": page}) + dt = rt.get("data") or {} + stage = dt.get("stage") + if i % 3 == 0 or stage not in (0, 1): + log(f"[pc try#{i}] rc={rt.get('returnCode')} stage={stage} " + f"uid={dt.get('uid')} biz={'有' if dt.get('biztoken') else '无'}") + if stage == 2 and dt.get("biztoken"): + log("\n★★★ 扫码登录闭环成功！ ★★★") + log(f" uid={dt.get('uid')}") + log(f" biztoken({len(dt['biztoken'])})={dt['biztoken'][:40]}...") + out = Path("evidence/qr_login_success.json") + out.write_text(json.dumps( + {"account": args.account, "qrId": qrid, + "uid": dt.get("uid"), "biztoken": dt["biztoken"], + "biz_variant": bname, "wup_variant": wname, + "ts": time.time()}, ensure_ascii=False, indent=1)) + log(f" 已保存 {out}") + return 0 + if rt.get("returnCode") != 0: + break + time.sleep(2.5) + bound = True + break + if bound: + break + + log("\n==== 全部变体未打通 ====") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())