feat(huya): 纯Python护照客户端骨架 - 信封构建+盲发探测(待真实包对齐)

tools/huya_passport_client.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""纯 Python 护照登录客户端(进行中)。
+
+路线: wsapi.huya.com / servant=huyaudbwebui / func=wupudbrequest_v0
+      内层 MsgRequestLoginPassport(id=0x1001) JSON 载荷
+目标: 响应 JSON 含 cred -> cert_forge -> bind -> 页面 cookie。
+
+状态: 信封构建完成; 内层帧格式待盲发试错(step2)。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import random
+import struct
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import requests
+
+WSAPI = "https://wsapi.huya.com"
+SERVANT = "huyaudbwebui"
+FUNC = "wupudbrequest_v0"
+MSG_ID = 0x1001  # MsgRequestLoginPassport
+
+
+def _wup_packet(servant: str, func: str, body: bytes) -> bytes:
+    """按抓包样本复刻 WUP 包头: [u32be len][10 03][',LV'][servant][func][body]。"""
+    core = (bytes([len(servant)]) + servant.encode()
+            + bytes([len(func)]) + func.encode() + body)
+    return struct.pack(">I", len(core) + 6) + b"\x10\x03, LV".replace(b" ", b"")[:4] + core[3 - 3:] if False else \
+        struct.pack(">I", len(core) + 6) + b"\x10\x03" + b",LV" + core
+
+
+def build_request(name: str, password_sha1: str, user_action: dict | None = None,
+                  biz_appids: list | None = None) -> dict:
+    ua = user_action or {"curl": "登录页", "furl": "我的", "latitude": "-1.0",
+                         "longitude": "-1.0", "ssid": "", "user_action": []}
+    return {
+        "name": name,
+        "password": password_sha1,
+        "userAction": json.dumps(ua, ensure_ascii=False),
+        "isAuthLogin": False,
+        "lgnExtParam": {},
+        "bizAppids": biz_appids or [],
+    }
+
+
+def probe_inner_formats(payload: dict):
+    """step2: 对内层帧做多种候选包装, 盲发读响应定位格式。"""
+    js = json.dumps(payload, separators=(",", ":"))
+    cands = {
+        "raw_json": js.encode(),
+        "u16id_json": struct.pack("<H", MSG_ID) + js.encode(),
+        "u32id_json": struct.pack(">I", MSG_ID) + js.encode(),
+        "json_id_field": json.dumps({"msgId": MSG_ID, "data": payload},
+                                    separators=(",", ":")).encode(),
+    }
+    s = requests.Session()
+    s.trust_env = False
+    for name, body in cands.items():
+        pkt = _wup_packet(SERVANT, FUNC, body)
+        try:
+            r = s.post(WSAPI, data=pkt, headers={
+                "Content-Type": "application/octet-stream",
+                "User-Agent": "okhttp/3.14.9"}, timeout=15)
+            head = r.content[:64]
+            print(f"[{name}] HTTP {r.status_code} {len(r.content)}B {head!r}")
+        except Exception as e:
+            print(f"[{name}] EXC {e}")
+
+
+if __name__ == "__main__":
+    acct = sys.argv[1] if len(sys.argv) > 1 else "hy_300023887"
+    pwd = sys.argv[2] if len(sys.argv) > 2 else ""
+    sha1 = hashlib.sha1(pwd.encode()).hexdigest() if pwd else "0" * 40
+    probe_inner_formats(build_request(acct, sha1))