feat(huya): 纯Python hypasswordLogin包服务器接受(HTTP200) - 传输层全通!

tools/huya_longconn.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""虎牙长连接(wup.huya.com:443)纯 Python 客户端骨架。
+
+协议: TAF/mobileuif 长连接; 请求包=标准 TAF RequestPacket JCE 结构
+      {iVer0,cPktType1,iReqId2,sServant3,sFunc4,sBuffer5,iTimeout6,context7,status8}
+      createWupPackage@0x274ea4 实证: ver=3, servant=huyaudbwebui,
+      func=wupudbrequest_v0, sBuffer=JSON 载荷, id=arg(疑似 msgId 0x1001)。
+
+用法: python3 huya_longconn.py probe   # 测连通性+观察服务器主动下发
+"""
+from __future__ import annotations
+
+import json
+import socket
+import ssl
+import struct
+import sys
+import time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+HOST, PORT = "wup.huya.com", 443
+
+
+# ---------- 极简 JCE(TAF) 编码器: 只覆盖本协议需要的类型 ----------
+def jce_zero(tag: int) -> bytes:
+    return bytes([(tag << 4) | 0])          # ZERO_TAG
+
+
+def jce_int(tag: int, v: int) -> bytes:
+    if v == 0:
+        return jce_zero(tag)
+    if -128 <= v <= 127:
+        return bytes([(tag << 4) | 1]) + struct.pack(">b", v)
+    if -32768 <= v <= 32767:
+        return bytes([(tag << 4) | 2]) + struct.pack(">h", v)
+    if -(1 << 31) <= v <= (1 << 31) - 1:
+        return bytes([(tag << 4) | 3]) + struct.pack(">i", v)
+    return bytes([(tag << 4) | 4]) + struct.pack(">q", v)
+
+
+def jce_str(tag: int, s: str | bytes) -> bytes:
+    b = s.encode() if isinstance(s, str) else s
+    n = len(b)
+    head = bytes([(tag << 4) | 6])
+    if n <= 255:
+        return head + bytes([n]) + b
+    return head + b"\xff" + struct.pack(">I", n) + b   # LONG_LEN
+
+
+def jce_bytes(tag: int, b: bytes) -> bytes:
+    """SimpleList(vector<byte>): head byte + size(int) + data。"""
+    return bytes([(tag << 4) | 13, 0]) + struct.pack(">i", len(b)) + b
+
+
+def jce_map_str(tag: int, d: dict[str, str]) -> bytes:
+    out = bytes([(tag << 4) | 8]) + struct.pack(">i", len(d))
+    for k, v in d.items():
+        out += jce_str(0, k) + jce_str(1, v)
+    return out
+
+
+def taf_request_packet(req_id: int, servant: str, func: str, buffer: bytes,
+                       timeout_ms: int = 3000,
+                       context: dict | None = None,
+                       status: dict | None = None) -> bytes:
+    body = (jce_int(0, 3)                      # iVersion=3
+            + jce_int(1, 0)                    # cPacketType
+            + jce_int(2, req_id)               # iRequestId
+            + jce_str(3, servant)
+            + jce_str(4, func)
+            + jce_bytes(5, buffer)
+            + jce_int(6, timeout_ms)
+            + jce_map_str(7, context or {})
+            + jce_map_str(8, status or {}))
+    return struct.pack(">I", len(body) + 4) + body
+
+
+# ---------- 连接层 ----------
+def connect_tls(timeout: float = 10.0) -> ssl.SSLSocket:
+    raw = socket.create_connection((HOST, PORT), timeout=timeout)
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    return ctx.wrap_socket(raw, server_hostname=HOST)
+
+
+def probe(recv_seconds: float = 6.0):
+    print(f"connect {HOST}:{PORT} ...")
+    s = connect_tls()
+    print("TLS ok:", s.version(), s.getpeercert(binary_form=False) is not None)
+    s.settimeout(recv_seconds)
+    try:
+        data = s.recv(4096)
+        print(f"server push on connect: {len(data)}B {data[:64].hex()}")
+    except socket.timeout:
+        print("no push within", recv_seconds, "s")
+    s.close()
+
+
+if __name__ == "__main__":
+    (probe,) or None
+    if len(sys.argv) > 1 and sys.argv[1] == "probe":
+        probe()