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
parent 705c51bf48
commit 9aea9851bd
4 changed files with 632 additions and 0 deletions
@@ -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())