Files
live-hub-py/scripts/probe_huya_qr_bind.py
T
yml2213 9aea9851bd test(huya): 风控场景/代理/扫码闭环探针与逆向文档
- probe_huya_login_scene: 各 appId/UA 组合的风控策略探测
- probe_huya_proxy: 走代理源对比直连的风控策略
- probe_huya_qr_bind: 扫码登录闭环(getQrId->scanQrPicNotify->
  bindQrLoginUser->tryQrLogin), 实测出有效web会话; 含wupData
  结构地图与证书定位实验结论(见 docs/虎牙扫码登录闭环逆向.md)
2026-08-25 11:51:01 +08:00

363 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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_biztokenApp 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_<uid>
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())