"""虎牙 App 密码登录 WUP(TAF) 请求帧编码器。 逆向自真实抓包金标准 `evidence/wup_passwordlogin_taf.bin`（1015 字节， POST https://udblgn.huya.com/open/hy/passwordLogin 的 base64 解码后 body）。 帧结构（偏移从 0 开始，含 4 字节长度前缀）: [0:4] 长度 = 1015 (big-endian uint32) WUP header（TAF 字段，tag 1..10）: tag1 iVersion = 3 (INT8) tag2 cPacketType = 0 (ZERO) tag3 iMessageType = 0 (ZERO) tag4 iRequestId = session (INT32) tag5 sServantName = "huyaudbwebui" (STRING1) tag6 sFuncName = "hypasswordLogin" (STRING1) tag7 sBuffer = bytes (SIMPLE_LIST, 961 字节) tag8 iTimeout = 0 (ZERO) tag9 context = {} (空 MAP) tag10 status = {} (空 MAP) sBuffer = Map 共 2 项: "_wup_data" -> 915 字节 TAF struct（业务数据，见下） "wupudbrequest_v0" -> 5 字节 = INT32(session) ← 裸 INT32 字段，非 struct _wup_data struct（915 字节，STRUCT_BEGIN 包裹，字段 t0..t8）: t0 (struct, 请求"头"): t0 = 0 (ZERO) t1 = "1.0" t2 = 元数据 JSON (compact, 固定键序) t3 = bizAppid "5008" t4 = 3 (INT8) t5 = safedeviceid t6 = "" t7 = "" t8 = user_action_json t9 = "" t1 (struct, 设备信息): t0 = hdid t1 = app_version "13.4.22" t2 = sdk_version "1.0.80138" t3 = "" t4 = ip "127.0.0.1" t5 = vendor "xiaomi" t6 = "" t2 (struct, 屏幕/设备特征，注意 t5 缺失): t0 = 1 (INT8) t1 = model "M2102J2SC" t2 = fingerprint "02df3987..." t3 = os "android" t4 = screen "M2102J2SC,30,11" t6 = width "1080" t7 = height "2120" t8 = device_id "7c5387e0..." t3 = name "hy_" t4 = sha1_password t5 = LIST[1] = ["5008"] t6 = 1 (INT8) t7 = {} (空 MAP) t8 = 空 SIMPLE_LIST (bytes) 两个疑点字节的最终结论: - 偏移 27 的 0x66 ('f'): 并非字面 'f'，而是 TAF head 字节 (tag6<<4 | STRING1=6) = sFuncName 字段的 STRING1 头。ASCII 'f' 纯属巧合。 - 偏移 44 的 0x7d ('}'): 并非字面 '}'，而是 TAF head 字节 (tag7<<4 | SIMPLE_LIST=13) = sBuffer 字段的 SIMPLE_LIST(byte[]) 头。 偏移 45..51 的 7 字节 = SIMPLE_LIST 元素类型头(00) + 长度(01 03 c1 = 961) + sBuffer 内层 Map 头(08) + Map 长度(00 02 = 2)。 """ from __future__ import annotations import json import random import struct import time as _time from typing import Any, Dict # --------------------------------------------------------------------------- # TAF 类型标签 # --------------------------------------------------------------------------- INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03 STRING1, STRING4 = 0x06, 0x07 MAP, LIST = 0x08, 0x09 STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B ZERO, SIMPLE_LIST = 0x0C, 0x0D class _Writer: """极简 TAF 输出流（仅够构造本请求帧）。""" def __init__(self) -> None: self.buf = bytearray() def get(self) -> bytes: return bytes(self.buf) def head(self, tag: int, dtype: int) -> None: if tag < 15: self.buf.append((tag << 4) | dtype) else: self.buf.append(0xF0 | dtype) self.buf.append(tag) def int8(self, tag: int, v: int) -> None: if v == 0: self.head(tag, ZERO) else: self.head(tag, INT8) self.buf += struct.pack('b', v) def int16(self, tag: int, v: int) -> None: if -128 <= v <= 127: self.int8(tag, v) else: self.head(tag, INT16) self.buf += struct.pack('>h', v) def int32(self, tag: int, v: int) -> None: if -32768 <= v <= 32767: self.int16(tag, v) else: self.head(tag, INT32) self.buf += struct.pack('>i', v) def int64(self, tag: int, v: int) -> None: if -2147483648 <= v <= 2147483647: self.int32(tag, v) else: self.head(tag, INT64) self.buf += struct.pack('>q', v) def string(self, tag: int, s: str) -> None: b = s.encode('utf-8') if len(b) > 255: self.head(tag, STRING4) self.buf += struct.pack('>I', len(b)) else: self.head(tag, STRING1) self.buf += struct.pack('B', len(b)) self.buf += b def bytes(self, tag: int, b: bytes) -> None: """SIMPLE_LIST (byte[])。""" self.head(tag, SIMPLE_LIST) self.head(0, INT8) # 元素类型固定 INT8 self.int32(0, len(b)) # 长度（int32 优化编码） self.buf += b def struct_begin(self, tag: int) -> None: self.head(tag, STRUCT_BEGIN) def struct_end(self) -> None: self.head(0, STRUCT_END) def list_begin(self, tag: int, n: int) -> None: self.head(tag, LIST) self.int32(0, n) def map_begin(self, tag: int, n: int) -> None: self.head(tag, MAP) self.int32(0, n) def _build_meta_json(session: int, trace_id: str) -> str: """构造 _wup_data.t0.t2 元数据 JSON（固定键序，compact）。""" meta: Dict[str, Any] = { "associationId": 8193, "funcName": "hypasswordLogin", "group": 1, "id": 4097, "session": session, "step": 1, "stillLogin": False, "traceId": trace_id, "type": 3, "uid": 0, "userContext": "", } return json.dumps(meta, ensure_ascii=False, separators=(',', ':')) def _make_name(uid_str: str) -> str: """登录名 = "hy_" + 虎牙号（若已带前缀则原样返回）。""" if uid_str.startswith("hy_"): return uid_str return "hy_" + uid_str def make_user_action(now_ms: int | None = None) -> str: """生成一条随机的 userAction 风控行为 JSON。 实测（2025-08 抓包 + 重放）: 服务端对该字段不做严格校验， 随机坐标/时间戳与旧轨迹一样能通过登录（返回 uid+token）。 参数: now_ms: 基准毫秒时间戳；缺省取当前时间。 """ if now_ms is None: now_ms = int(_time.time() * 1000) t1 = now_ms t2 = now_ms + random.randint(200, 900) # 两次点击间隔 200~900ms return json.dumps( { "curl": "登录页", "furl": "我的", "latitude": "-1.0", "longitude": "-1.0", "ssid": "", "user_action": [ {"id": "24", "time": str(t1), "x": str(random.randint(150, 900)), "y": str(random.randint(800, 1600))}, {"id": "11", "time": str(t2), "x": str(random.randint(150, 900)), "y": str(random.randint(800, 1600))}, ], }, ensure_ascii=False, separators=(",", ":"), ) def make_trace_id(pid: int = 0) -> str: """生成 traceId，格式 `--<毫秒时间戳>`（与服务端生成规则一致）。""" return f"{random.getrandbits(64):016x}-{pid}-{int(_time.time() * 1000)}" def _build_wup_data(w: _Writer, uid_str: str, sha1_password: str, safedeviceid: str, hdid: str, session: int, trace_id: str, user_action_json: str, device_info: Dict[str, str]) -> None: """编码 915 字节的 _wup_data struct。""" meta_json = _build_meta_json(session, trace_id) name = _make_name(uid_str) # ---- _wup_data struct（即"外层" struct）---- w.struct_begin(0) # _wup_data struct (0x0a) # -- t0: 请求"头" struct -- w.struct_begin(0) # t0 (0x0a) w.int8(0, 0) # t0.t0 = 0 -> ZERO (0x0c) w.string(1, "1.0") # t0.t1 = "1.0" w.string(2, meta_json) # t0.t2 = 元数据 JSON w.string(3, "5008") # t0.t3 = bizAppid w.int8(4, 3) # t0.t4 = 3 -> INT8 w.string(5, safedeviceid) # t0.t5 w.string(6, "") # t0.t6 w.string(7, "") # t0.t7 w.string(8, user_action_json) # t0.t8 w.string(9, "") # t0.t9 w.struct_end() # 0x0b # -- t1: 设备信息 struct -- di = device_info w.struct_begin(1) # t1 (0x1a) w.string(0, hdid) # t1.t0 = hdid w.string(1, di["app_version"]) # t1.t1 = "13.4.22" w.string(2, di["sdk_version"]) # t1.t2 = "1.0.80138" w.string(3, "") # t1.t3 w.string(4, di["ip"]) # t1.t4 = "127.0.0.1" w.string(5, di["vendor"]) # t1.t5 = "xiaomi" w.string(6, "") # t1.t6 w.struct_end() # 0x0b # -- t2: 屏幕/设备特征 struct（注意 t5 缺省）-- w.struct_begin(2) # t2 (0x2a) w.int8(0, 1) # t2.t0 = 1 -> INT8 w.string(1, di["model"]) # t2.t1 = "M2102J2SC" w.string(2, di["fingerprint"]) # t2.t2 = 40 位指纹 w.string(3, di["os"]) # t2.t3 = "android" w.string(4, di["screen"]) # t2.t4 = "M2102J2SC,30,11" # (t5 缺省) w.string(6, di["width"]) # t2.t6 = "1080" w.string(7, di["height"]) # t2.t7 = "2120" w.string(8, di["device_id"]) # t2.t8 = 40 位设备ID w.struct_end() # 0x0b # -- 登录字段 -- w.string(3, name) # t3 = "hy_300023887" w.string(4, sha1_password) # t4 = SHA1 hex w.list_begin(5, 1) # t5 = LIST[1] w.string(0, "5008") # item = "5008" w.int8(6, 1) # t6 = 1 -> INT8 w.map_begin(7, 0) # t7 = 空 MAP w.bytes(8, b"") # t8 = 空 bytes (8d 00 0c) w.struct_end() # 0x0b (_wup_data struct 结束) def build_password_login_wup( uid_str: str, sha1_password: str, safedeviceid: str, hdid: str, session: int, trace_id: str, user_action_json: str, device_info: Dict[str, str], ) -> bytes: """构造密码登录的 WUP TAF 请求体（1015 字节那种，未 base64）。 参数: uid_str : 虎牙号（如 "300023887"），登录名 = "hy_" + uid_str。 若已带 "hy_" 前缀则原样使用。 sha1_password : SHA1(明文密码) 的 40 位 hex 字符串。 safedeviceid : 设备安全 ID（长 base64 串）。 hdid : 硬件设备 ID（32 位 hex）。 session : 会话/请求 id（同时写入 iRequestId、元数据 session、 "wupudbrequest_v0" 值）。 trace_id : 元数据 traceId。 user_action_json : 用户行为 JSON 字符串（curl/furl/user_action 等）。 device_info : dict，需包含键: app_version, sdk_version, vendor, model, os, fingerprint, screen, width, height, device_id, ip。 返回: 1015 字节 TAF 二进制（HTTP body 为 base64 编码后发送）。 """ # 1. _wup_data struct wd = _Writer() _build_wup_data(wd, uid_str, sha1_password, safedeviceid, hdid, session, trace_id, user_action_json, device_info) wup_data = wd.get() # 2. "wupudbrequest_v0" 值 = 裸 INT32(session)（非 struct） req = _Writer() req.int32(0, session) wupdbreq_v0 = req.get() # 3. sBuffer = Map，2 项 sb = _Writer() sb.map_begin(0, 2) sb.string(0, "_wup_data") sb.bytes(1, wup_data) sb.string(0, "wupudbrequest_v0") sb.bytes(1, wupdbreq_v0) s_buffer = sb.get() # 4. WUP header w = _Writer() w.int16(1, 3) # iVersion = 3 -> INT8 w.int8(2, 0) # cPacketType = 0 -> ZERO w.int8(3, 0) # iMessageType = 0 -> ZERO w.int32(4, session) # iRequestId -> INT32 w.string(5, "huyaudbwebui") # sServantName w.string(6, "hypasswordLogin") # sFuncName w.bytes(7, s_buffer) # sBuffer w.int32(8, 0) # iTimeout = 0 -> ZERO w.map_begin(9, 0) # context = 空 MAP w.map_begin(10, 0) # status = 空 MAP wup_body = w.get() # 5. 长度前缀 return struct.pack('>I', 4 + len(wup_body)) + wup_body # --------------------------------------------------------------------------- # 自检 # --------------------------------------------------------------------------- if __name__ == "__main__": import os here = os.path.dirname(os.path.abspath(__file__)) golden_path = os.path.join(here, "..", "evidence", "wup_passwordlogin_taf.bin") golden = open(golden_path, "rb").read() device_info = { "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", } out = build_password_login_wup( uid_str="300023887", sha1_password="772ed992b0e161276f44ec63671e60155c506294", safedeviceid=("PQwemAN9NHkZKoMqVTFUZBIypqMTaQEOrmXr37xQVhQZqrL/gUKEQ11xvE0ju48V8O/" "t9UBGSp27m4+6bP4IiAEnpaR5Rj1kHEfN2SPLPqYZW9vroxUSoAvjJn6ezTP9jWGxxlRDCbt" "Py4Rd6MencYT/pNImVIWK+YbNKZt1O05bHUFhqHf3"), hdid="ed0db8334cadd236c00cadf7e11ab5a5", session=3251699, trace_id="0b8f098ff64a5bdc-23473-94783833787595172451", user_action_json=('{"curl":"登录页","furl":"我的","latitude":"-1.0","longitude":"-1.0",' '"ssid":"","user_action":[{"id":"24","time":"1787595171950","x":"277","y":"1057"},' '{"id":"11","time":"1787595172444","x":"296","y":"952"}]}'), device_info=device_info, ) assert out == golden, "编码结果与金标准不一致！" print("编码器验证通过") print(f" 输出长度 = {len(out)} 字节，与金标准逐字节一致")