yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

虎牙App密码登录: 逆向WUP协议并实现纯Python登录编码器

Browse Source 
- hook WUP序列化层抓取TLS加密前明文，确认密码登录走
  servant=huyaudbwebui / func=hypasswordLogin (msgType=0x1001)
- 逆向JCE RequestPacket帧格式(1015字节TAF二进制)，定稿三层结构:
  WUP header(tag1-10) -> sBuffer Map -> _wup_data业务struct
- 澄清两个疑点字节: 0x66/0x7d 是TAF字段头(非分隔符)
- 实现 tools/huya_wup_encoder.py，逐字节复现金标准
- 实测纯Python动态构造登录成功(返回uid+token)，TLS指纹不阻断
- userAction随机坐标/时间戳同样通过，内置make_user_action辅助函数
- 新增证据金标准 + 协议文档
This commit is contained in:
yml2213
2026-08-25 02:59:20 +08:00
parent 06a0a1e7d3
commit ff99c036f8
14 changed files with 2185 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/huya_app_crypto.py
+238
View File
@@ -0,0 +1,238 @@
"""
虎牙 App 登录 — OTP/dckey 加密算法（已通过 Frida 动态双向验证）
完整加密链（libudbauthunify.so，com.duowan.kiwi 13.4.22 / versionCode 115315）:
┌─────────────────────────────────────────────────────────────┐
│ OTP 加密（hyudb_otp_encrypt） │
│ │
│ 1. XXTEA: xxtea_key = md5(uid字符串).hexdigest()[:16] │
│ xxtea_in = arg7(8字节小端) + 4字节原始长度 │
│ (标准 XXTEA, rounds = 6 + 52/n, DELTA=0x9e3779b9)│
│ │
│ 2. cred_pack 拼装: │
│ [0x02] # 固定 header │
│ [u16_le len][xxtea密文] # string 打包 │
│ [u16_le len][设备指纹] # string 打包 │
│ │
│ 3. AES-128-ECB + ZeroPadding: │
│ aes_key = md5(sessHex32 + AESKEYMGR_KEY[b])[4:12].hex() │
│ cipher = AES_ECB(拼装数据 + 补零) │
└─────────────────────────────────────────────────────────────┘
动态验证样本（两次 hook，两个独立 PID，结果一致）:
uid = "1471238907296"
sessHex32 = "865a4924a40897ac1fcfe6b4c2cbb0e3" (固定)
AESKEYMGR = "MKDKeridjing7avnsasdSDHI" (b=2)
aes_key = "312334e88d8c35cb"
XXTEA 双向、AES 加密方向均已验证通过。
"""
from __future__ import annotations
import hashlib
import struct
from binascii import unhexlify
from Crypto.Cipher import AES
# ---------------------------------------------------------------------------
# AESkeyMgr 硬编码 key 表（按 b 参数索引，1-based）
# b=2..5 已通过 Frida hook AESkeyMgr::getkey(a=1, b) 动态确认；
# 其余为 rodata 相邻字符串推断，待最终确认顺序。
# ---------------------------------------------------------------------------
AESKEYMGR_KEYS: dict[int, str] = {
1: "owNMiaCgcHmqoTr3iRamFuHj",
2: "MKDKeridjing7avnsasdSDHI", # 已验证
3: "nskdI7MDGKSDJsnadjdoonvs", # 已验证
4: "whdFNGFdfJHMJbdfg6jhdsFG", # 已验证
5: "SHBfgytjtoikooruhogji7ER", # 已验证
6: "KNSDNjfohweeromnmkladj3g",
7: "KJNzmZaxUd9kGLoAbtefNzCv",
8: "YnzbiMPNXaGuxxuoNRd9Lzyj",
}
# 固定 32 位 hex 会话值（OTP AES key 源的第一半）
SESS_HEX32 = "865a4924a40897ac1fcfe6b4c2cbb0e3"
DELTA = 0x9E3779B9
def huya_arg7(now_ms: int | None = None) -> int:
"""arg7 = System.currentTimeMillis() << 16（低 16 位恒为 0）。
Frida 实测：arg7 >> 16 ≈ 事件时间戳(ms)，误差为 hook 传播延迟（<600ms）。
"""
import time
if now_ms is None:
now_ms = int(time.time() * 1000)
return now_ms << 16
# ---------------------------------------------------------------------------
# XXTEA（标准实现，已验证）
# ---------------------------------------------------------------------------
def _xxtea_mx(z, y, total, key, p, e):
return (((z >> 5 ^ y << 2) + (y >> 3 ^ z << 4))
^ ((total ^ y) + (key[(p & 3) ^ e] ^ z))) & 0xFFFFFFFF
def _xxtea_rounds(n: int) -> int:
return 6 + 52 // n
def xxtea_encrypt_u32(v: list[int], key: list[int]) -> list[int]:
n = len(v)
if n < 2:
return v
total = 0
z = v[n - 1]
for _ in range(_xxtea_rounds(n)):
total = (total + DELTA) & 0xFFFFFFFF
e = (total >> 2) & 3
for p in range(n):
y = v[(p + 1) % n]
v[p] = (v[p] + _xxtea_mx(z, y, total, key, p, e)) & 0xFFFFFFFF
z = v[p]
return v
def xxtea_decrypt_u32(v: list[int], key: list[int]) -> list[int]:
n = len(v)
if n < 2:
return v
rounds = _xxtea_rounds(n)
total = (rounds * DELTA) & 0xFFFFFFFF
y = v[0]
while total != 0:
e = (total >> 2) & 3
for p in range(n - 1, 0, -1):
z = v[p - 1]
v[p] = (v[p] - _xxtea_mx(z, y, total, key, p, e)) & 0xFFFFFFFF
y = v[p]
z = v[n - 1]
v[0] = (v[0] - _xxtea_mx(z, y, total, key, 0, e)) & 0xFFFFFFFF
y = v[0]
total = (total - DELTA) & 0xFFFFFFFF
return v
def xxtea_key_from_uid(uid: str) -> list[int]:
"""XXTEA key = md5(uid字符串).hexdigest()[:16] 的 16 个 ASCII 字节（4 个小端 u32）。"""
h = hashlib.md5(uid.encode("ascii")).hexdigest()[:16]
return list(struct.unpack("<4I", h.encode("ascii")))
def huya_xxtea_encrypt(data: bytes, uid: str) -> bytes:
"""XXTEA 加密：数据 + 4字节原始长度 padding，再标准 XXTEA。"""
key = xxtea_key_from_uid(uid)
pad = (-len(data)) % 4
buf = data + b"\x00" * pad + struct.pack("<I", len(data))
v = list(struct.unpack("<%dI" % (len(buf) // 4), buf))
enc = xxtea_encrypt_u32(v, key)
return b"".join(struct.pack("<I", x) for x in enc)
# ---------------------------------------------------------------------------
# AES（已验证）
# ---------------------------------------------------------------------------
def huya_aes_key(otp_input: bytes) -> bytes:
"""AES-128 key = md5(otp_input) digest 第 5~12 字节转 hex（16 ASCII 字符）。"""
d = hashlib.md5(otp_input).digest()
return d[4:12].hex().encode("ascii")
def huya_aes_encrypt(plain: bytes, otp_input: bytes) -> bytes:
key = huya_aes_key(otp_input)
pad = (-len(plain)) % 16
return AES.new(key, AES.MODE_ECB).encrypt(plain + b"\x00" * pad)
def huya_aes_decrypt(cipher: bytes, otp_input: bytes) -> bytes:
key = huya_aes_key(otp_input)
return AES.new(key, AES.MODE_ECB).decrypt(cipher).rstrip(b"\x00")
# ---------------------------------------------------------------------------
# cred_pack 打包
# ---------------------------------------------------------------------------
def cred_pack_string(data: bytes) -> bytes:
"""string = [2字节小端长度][数据]（cred_pack::push_varstr）。"""
return struct.pack("<H", len(data)) + data
# ---------------------------------------------------------------------------
# 完整 OTP 加密（对应 hyudb_otp_encrypt）
# ---------------------------------------------------------------------------
def huya_password_sha1(password: str) -> str:
"""密码登录的 password 字段 = 标准 SHA1(明文密码) 的 hex，无盐。
Frida 抓包证据（msgType=0x1001 密码登录请求）:
password = "772ed992b0e161276f44ec63671e60155c506294" (40 位 hex)
"""
return hashlib.sha1(password.encode("utf-8")).hexdigest()
def huya_name(hyname: str) -> str:
"""登录 name 字段 = "hy_" + 虎牙号（纯数字账号）。"""
return f"hy_{hyname}"
def huya_otp_encrypt(
uid: str,
arg7: int,
device_fingerprint: bytes,
b: int = 2,
sess_hex32: str = SESS_HEX32,
) -> bytes:
"""完整 OTP 加密，返回 AES 密文。
uid : 账号 uid 的十进制字符串（XXTEA key 源）
arg7 : 8 字节 ulong（时间戳/计数器）
device_fingerprint : 设备指纹二进制（arg5）
b : AESkeyMgr key 表索引（1..8）
sess_hex32 : 固定 32 位 hex 会话值
"""
# 1. XXTEA 加密 arg7
xxtea_ct = huya_xxtea_encrypt(struct.pack("<Q", arg7), uid)
# 2. cred_pack 拼装
packed = b"\x02" + cred_pack_string(xxtea_ct) + cred_pack_string(device_fingerprint)
# 3. AES-128-ECB
otp_input = (sess_hex32 + AESKEYMGR_KEYS[b]).encode("ascii")
return huya_aes_encrypt(packed, otp_input)
if __name__ == "__main__":
# 自检 1: AES
otp_input = (SESS_HEX32 + AESKEYMGR_KEYS[2]).encode("ascii")
plain = unhexlify(
"020c00cf8b207ddeb2ec4ec2a9344772000a70686484dee8e70dd7649fa14e82"
"86caef42e5ddb9c380aef2485f87170507881b3e11b04c8dc976b8d056dab02"
"45672e3b5595f73d1803c123672fb3525bb3640b986a94d4c00349995e904a1"
"0b27b6d9d0078b1a0c85abdcd1747e588831370d553cf3bd9e3032a9537e30f"
"29dbb4866"
)
expected = unhexlify(
"b2f83ab1970e3d343935b502874a63e3e4a674c60827347a8b81186e11bfb44"
"3a95cd1252536822859daa76685317aab55ca6f50530b98b72122b61f774d6a"
"9bc515930d7a88f387efc6dac90379ee26b346a5a08e42546950b268a8a7f5c"
"ecfe8b93f2b1271c2c40ccddc8f805b2732e2d16e3e79fd526666b58a5fc3d0"
"af366cebfbeaa91089188cb5d7ed48661cbd"
)
assert huya_aes_key(otp_input) == b"312334e88d8c35cb", "AES key 派生失败"
assert huya_aes_encrypt(plain, otp_input) == expected, "AES 加密失败"
assert huya_aes_decrypt(expected, otp_input) == plain, "AES 解密失败"
# 自检 2: XXTEA
uid = "1471238907296"
xxtea_ct = huya_xxtea_encrypt(struct.pack("<Q", 0x01A034AD5ECA0000), uid)
assert xxtea_ct.hex() == "d24e6493d0ee53c395790d67", f"XXTEA 失败: {xxtea_ct.hex()}"
# 自检 3: 密码 SHA1（无盐）
assert huya_password_sha1("password123") == hashlib.sha1(b"password123").hexdigest()
print("自检全部通过")
print(" AES key =", huya_aes_key(otp_input).decode())
print(" XXTEA =", huya_xxtea_encrypt(struct.pack('<Q', 0x01A034AD5ECA0000), uid).hex())
tools/huya_wup_encoder.py
+397
View File
@@ -0,0 +1,397 @@
"""虎牙 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<str, bytes> 共 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_<uid_str>"
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，格式 `<hex16>-<pid>-<毫秒时间戳>`（与服务端生成规则一致）。"""
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<str, bytes>，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)} 字节，与金标准逐字节一致")
tools/hy_har_parse.py
+157
View File
@@ -0,0 +1,157 @@
"""解析虎牙 App HAR：解码 WUP(TAF) 全字段 + dckey/check 分析。
用法:
.venv/bin/python tools/hy_har_parse.py <file.har> [--req] [--resp]
"""
from __future__ import annotations
import base64
import json
import sys
sys.path.insert(0, ".")
from tools.taf_decode import ( # noqa: E402
Taf, STRUCT_END, MAP, SIMPLE_LIST, decode_struct_bytes,
)
STR = (6, 7) # STRING1, STRING4
SIMPLE = 13
def norm_payload(data: bytes) -> bytes:
"""去掉可选 4 字节 WUP 长度前缀。"""
if len(data) < 4:
return data
declared = int.from_bytes(data[0:4], "big")
if declared in (len(data), len(data) - 4):
return data[4:]
return data
def decode_wup_header(data: bytes):
"""返回 (servant, func, sBuffer_bytes)。"""
t = Taf(norm_payload(data))
servant = ""
func = ""
sbuf = b""
while not t.eof():
try:
tag, dt = t.read_head()
except EOFError:
break
if tag > 10:
break
if tag == 5 and dt in STR:
servant = t.read_string_value(dt)
elif tag == 6 and dt in STR:
func = t.read_string_value(dt)
elif tag == 7 and dt == SIMPLE:
t.read_head()
n = t.read_len()
sbuf = t.d[t.p:t.p + n]
t.p += n
else:
t.skip_value(dt)
return servant, func, sbuf
def decode_sbuffer_map(sbuf: bytes):
"""sBuffer 是一个 Map<str, bytes>，返回 {key: bytes}。"""
if not sbuf:
return {}
t = Taf(sbuf)
out = {}
try:
tag, dt = t.read_head()
if not (tag == 0 and dt == MAP):
return {"<raw>": sbuf}
n = t.read_len()
for _ in range(min(n, 20)):
_, kt = t.read_head()
k = t.read_string_value(kt) if kt in STR else f"<k{kt}>"
_, vt = t.read_head()
if vt == SIMPLE:
t.read_head()
ln = t.read_len()
out[k] = t.d[t.p:t.p + ln]; t.p += ln
else:
t.skip_value(vt)
except Exception as e:
out["<err>"] = str(e).encode()
return out
def dump_har(path: str, show_req: bool = True, show_resp: bool = True):
with open(path, encoding="utf-8", errors="ignore") as fp:
h = json.load(fp)
ents = h["log"]["entries"]
targets = ("wup.huya.com", "statwup.huya.com", "wsapi.huya.com",
"udbdf.huya.com", "udblgn.huya.com", "udbreg.huya.com")
for i, e in enumerate(ents):
req = e["request"]
u = req["url"]
if req["method"] != "POST" or not any(x in u for x in targets):
continue
pd = req.get("postData", {})
txt = pd.get("text", "")
mime = pd.get("mimeType", "")
print("=" * 80)
print(f"#{i} {u[:110]}")
print(f" mime={mime} textLen={len(txt)}")
# ---- 请求 ----
if txt:
try:
raw = base64.b64decode(txt)
except Exception:
raw = txt.encode("latin1")
if len(raw) >= 6 and raw[4:6] == b"\x10\x03" and show_req:
servant, func, sbuf = decode_wup_header(raw)
print(f" [REQ WUP] {servant}.{func}")
for k, v in decode_sbuffer_map(sbuf).items():
print(f" {k}: {json.dumps(decode_struct_bytes(v), ensure_ascii=False)[:700]}")
else:
# 加密 binary（如 dckey/check、statwup 应用层加密）
print(f" [REQ BINARY] {len(raw)}B hex={raw[:32].hex()}")
analyze_binary(raw)
# ---- 响应 ----
resp = e.get("response", {})
c = resp.get("content", {})
rtext = c.get("text", "")
if rtext and show_resp:
if rtext.lstrip().startswith("{"):
print(f" [RESP JSON] {rtext[:220]}")
else:
try:
rraw = base64.b64decode(rtext)
except Exception:
rraw = rtext.encode("latin1")
if len(rraw) >= 6 and rraw[4:6] == b"\x10\x03":
servant, func, sbuf = decode_wup_header(rraw)
print(f" [RESP WUP] {servant}.{func}")
for k, v in decode_sbuffer_map(sbuf).items():
print(f" {k}: {json.dumps(decode_struct_bytes(v), ensure_ascii=False)[:700]}")
else:
print(f" [RESP BINARY] {len(rraw)}B hex={rraw[:32].hex()}")
elif rtext:
print(f" [RESP raw] {rtext[:200]!r}")
def analyze_binary(raw: bytes):
"""对加密 body 做基础分析：长度/对齐/头部熵。"""
print(f" len={len(raw)} mod16={len(raw) % 16} mod64={len(raw) % 64}")
# 头部是否像有固定结构（前几字节恒定）
print(f" head32={raw[:32].hex()}")
print(f" tail16={raw[-16:].hex()}")
if __name__ == "__main__":
args = [a for a in sys.argv[1:] if not a.startswith("--")]
path = args[0] if args else "-"
show_req = "--req" not in sys.argv or True
show_resp = "--resp" not in sys.argv or True
dump_har(path)
tools/taf_decode.py
+207
View File
@@ -0,0 +1,207 @@
"""虎牙 TAF(JCE) 纯字节流递归解码器 — 不依赖项目 read_* 高层 API。
独立实现 head/value 解析，直接按 dtype 读取 value（已消费 head 的前提下），
避免与 TafInputStream.read_string 等"内部再读 head"的方法冲突。
"""
from __future__ import annotations
import struct
# 类型标签（与 core/huya/taf_protocol.py TafType 一致）
INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03
FLOAT, DOUBLE = 0x04, 0x05
STRING1, STRING4 = 0x06, 0x07
MAP, LIST = 0x08, 0x09
STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B
ZERO, SIMPLE_LIST = 0x0C, 0x0D
_STR = (STRING1, STRING4)
_INTS = (INT8, INT16, INT32, INT64)
class Taf:
"""带游标的字节读取器"""
def __init__(self, data: bytes):
self.d = data
self.p = 0
def eof(self) -> bool:
return self.p >= len(self.d)
def peek_head(self):
if self.eof():
raise EOFError
b = self.d[self.p]
tag = (b >> 4) & 0x0F
dt = b & 0x0F
if tag == 15:
tag = self.d[self.p + 1]
return tag, dt
def read_head(self):
tag, dt = self.peek_head()
self.p += 1
if (self.d[self.p - 1] >> 4) & 0x0F == 15:
self.p += 1
return tag, dt
def read_len(self) -> int:
"""读 map/list 长度（int32 优化编码）"""
tag, dt = self.read_head()
return self.read_int_value(dt)
def read_int_value(self, dt: int) -> int:
if dt == ZERO:
return 0
if dt == INT8:
v = struct.unpack('b', self.d[self.p:self.p + 1])[0]; self.p += 1; return v
if dt == INT16:
v = struct.unpack('>h', self.d[self.p:self.p + 2])[0]; self.p += 2; return v
if dt == INT32:
v = struct.unpack('>i', self.d[self.p:self.p + 4])[0]; self.p += 4; return v
if dt == INT64:
v = struct.unpack('>q', self.d[self.p:self.p + 8])[0]; self.p += 8; return v
raise ValueError(f"期望整数, 实际 0x{dt:02x} @{self.p}")
def read_string_value(self, dt: int) -> str:
if dt == STRING1:
n = self.d[self.p]; self.p += 1
elif dt == STRING4:
n = struct.unpack('>I', self.d[self.p:self.p + 4])[0]; self.p += 4
else:
raise ValueError(f"期望 string, 实际 0x{dt:02x} @{self.p}")
v = self.d[self.p:self.p + n].decode('utf-8', errors='replace'); self.p += n
return v
def skip_value(self, dt: int):
if dt == ZERO:
return
if dt in _INTS:
self.p += {INT8: 1, INT16: 2, INT32: 4, INT64: 8}[dt]
elif dt in (FLOAT,):
self.p += 4
elif dt in (DOUBLE,):
self.p += 8
elif dt == STRING1:
self.p += 1 + self.d[self.p]
elif dt == STRING4:
self.p += 4 + struct.unpack('>I', self.d[self.p:self.p + 4])[0]
elif dt == SIMPLE_LIST:
self.read_head() # 元素类型
n = self.read_len()
self.p += n
elif dt == MAP:
n = self.read_len()
for _ in range(n):
_, kt = self.read_head(); self.skip_value(kt)
_, vt = self.read_head(); self.skip_value(vt)
elif dt == LIST:
n = self.read_len()
for _ in range(n):
_, it = self.read_head(); self.skip_value(it)
elif dt == STRUCT_BEGIN:
while True:
_, t = self.read_head()
if t == STRUCT_END:
break
self.skip_value(t)
# STRUCT_END 不需要处理
def decode_value(t: Taf, dt: int, depth: int = 0, max_depth: int = 6, max_items: int = 50):
"""给定已消费 head 的 dtype，递归读 value。"""
if dt == ZERO:
return 0
if dt in _INTS:
return t.read_int_value(dt)
if dt == FLOAT:
v = struct.unpack('>f', t.d[t.p:t.p + 4])[0]; t.p += 4; return round(v, 6)
if dt == DOUBLE:
v = struct.unpack('>d', t.d[t.p:t.p + 8])[0]; t.p += 8; return round(v, 6)
if dt in _STR:
return t.read_string_value(dt)
if dt == SIMPLE_LIST:
t.read_head() # 元素类型
n = t.read_len()
raw = t.d[t.p:t.p + n]; t.p += n
# 尝试 utf8 可读
try:
s = raw.decode('utf-8')
if s.isprintable() and n < 500:
return s
except Exception:
pass
return {"_bytes": n, "hex": raw[:48].hex()}
if dt == MAP:
n = t.read_len()
m = {}
for _ in range(min(n, max_items)):
kt_tag, kdt = t.read_head()
k = decode_value(t, kdt, depth + 1) if kdt != STRUCT_BEGIN else _decode_struct(t, depth + 1)
vt_tag, vdt = t.read_head()
v = decode_value(t, vdt, depth + 1) if depth + 1 < max_depth else _skip_str(t, vdt)
m[str(k)] = v
return m
if dt == LIST:
n = t.read_len()
items = []
for _ in range(min(n, max_items)):
it_tag, idt = t.read_head()
items.append(decode_value(t, idt, depth + 1) if depth + 1 < max_depth else _skip_str(t, idt))
return items
if dt == STRUCT_BEGIN:
return _decode_struct(t, depth + 1)
return f"<0x{dt:02x}>"
def _skip_str(t: Taf, dt: int):
t.skip_value(dt)
return f"<0x{dt:02x}>"
def _decode_struct(t: Taf, depth: int = 0):
"""已消费 STRUCT_BEGIN，读到 STRUCT_END。"""
out = {}
cnt = 0
while not t.eof() and cnt < 100:
try:
tag, dt = t.read_head()
except EOFError:
break
if dt == STRUCT_END:
break
key = f"t{tag}"
if depth < 6:
try:
out[key] = decode_value(t, dt, depth)
except Exception as e:
t.skip_value(dt)
out[key] = f"<err {e}>"
else:
t.skip_value(dt)
out[key] = f"<0x{dt:02x}>"
cnt += 1
return out
def decode_struct_bytes(data: bytes):
"""解析一个 TAF struct 字节串（可含或不含 STRUCT_BEGIN 头）。"""
t = Taf(data)
out = {}
cnt = 0
while not t.eof() and cnt < 100:
try:
tag, dt = t.read_head()
except EOFError:
break
if dt == STRUCT_END:
break
key = f"t{tag}"
try:
out[key] = decode_value(t, dt, 0)
except Exception as e:
t.skip_value(dt)
out[key] = f"<err {e}>"
cnt += 1
return out
tools/taf_dump.py
+148
View File
@@ -0,0 +1,148 @@
"""通用 Taf/JCE 递归解析 dump 工具(调试用)。"""
from __future__ import annotations
import base64
import json
import sys
sys.path.insert(0, ".")
from core.huya.taf_protocol import TafInputStream, TafType # noqa: E402
STR = (TafType.STRING1, TafType.STRING4)
INTS = (TafType.INT8, TafType.INT16, TafType.INT32, TafType.INT64)
def parse_struct(i: TafInputStream, depth: int = 0):
"""解析到 STRUCT_END 或数据结束, 返回 dict。"""
out = {}
pad = " " * depth
while True:
try:
tag, dtype = i.peek_head()
except Exception:
break
if dtype == TafType.STRUCT_END:
i.read_head()
break
key = f"tag{tag}"
if dtype == TafType.STRUCT_BEGIN:
i.read_head()
print(f"{pad}{key}: {{")
out[key] = parse_struct(i, depth + 1)
print(f"{pad}}}")
elif dtype in STR:
val = i.read_string(tag)
show = val if len(val) <= 300 else val[:300] + f"...({len(val)})"
print(f"{pad}{key}: STR {show!r}")
out[key] = val
elif dtype in INTS:
val = {TafType.INT8: i.read_int8, TafType.INT16: i.read_int16,
TafType.INT32: i.read_int32, TafType.INT64: i.read_int64}[dtype](tag)
print(f"{pad}{key}: INT {val}")
out[key] = val
elif dtype == TafType.ZERO:
i.read_head()
print(f"{pad}{key}: 0")
out[key] = 0
elif dtype == TafType.MAP:
n = i.read_int32(0)
print(f"{pad}{key}: MAP[{n}]")
m = {}
for _ in range(n):
kt, kdt = i.peek_head()
if kdt in STR:
kv = i.read_string(kt)
elif kdt in INTS:
kv = str({TafType.INT8: i.read_int8, TafType.INT16: i.read_int16,
TafType.INT32: i.read_int32, TafType.INT64: i.read_int64}[kdt](kt))
else:
break
vt, vdt = i.peek_head()
if vdt == TafType.STRUCT_BEGIN:
i.read_head()
print(f"{pad} [{kv}]: {{")
m[str(kv)] = parse_struct(i, depth + 2)
print(f"{pad} }}")
elif vdt == TafType.SIMPLE_LIST:
sv = i.read_bytes(vt)
m[str(kv)] = sv.hex()
print(f"{pad} [{kv}] BYTES[{len(sv)}] {sv[:40].hex()}")
elif vdt in STR:
sv = i.read_string(vt); m[str(kv)] = sv
print(f"{pad} [{kv}] STR {sv[:120]!r}")
else:
i.read_head(); i.skip_field(vdt)
out[key] = m
elif dtype == TafType.LIST:
n = i.read_int32(0)
print(f"{pad}{key}: LIST[{n}]")
items = []
for _ in range(n):
it, idt = i.read_head()
if idt in STR:
items.append(i.read_string(it))
elif idt in INTS:
items.append({TafType.INT8: i.read_int8, TafType.INT16: i.read_int16,
TafType.INT32: i.read_int32, TafType.INT64: i.read_int64}[idt](it))
else:
i.skip_field(idt)
out[key] = items
elif dtype == TafType.SIMPLE_LIST:
i.read_head() # 元素类型 INT8
n = i.read_int32(0)
raw = (getattr(i, 'stream', None) or getattr(i, 'buf')).read(n)
print(f"{pad}{key}: BYTES[{n}] {raw[:60].hex()}")
out[key] = raw.hex()
else:
print(f"{pad}{key}: ?type={hex(dtype)}")
try:
i.skip_field(dtype)
except Exception:
break
return out
def dump_packet(data: bytes, label: str):
print("=" * 70)
print(label, f"({len(data)} bytes)")
print("=" * 70)
from core.huya.wup_protocol import WupResponse
r = WupResponse()
try:
r.decode(data)
print(f"iVersion={r.iVersion} servant={r.sServantName!r} func={r.sFuncName!r}")
for k, v in r.newdata.items():
print(f"--- newdata[{k!r}] len={len(v)}")
if not k or len(v) < 8:
print(" ", v.hex())
continue
try:
parse_struct(TafInputStream(v))
except Exception as e:
print(" parse err:", e, "| hex:", v.hex()[:200])
except Exception as e:
# 可能没有长度前缀, 直接当 body
print("decode with prefix failed:", e, "-> try raw")
parse_struct(TafInputStream(data))
def main():
import argparse
ap = argparse.ArgumentParser()
ap.add_argument("path")
ap.add_argument("--b64", action="store_true", help="输入是 base64")
args = ap.parse_args()
if args.path == "-":
raw = sys.stdin.buffer.read()
else:
raw = open(args.path, "rb").read()
if args.b64:
data = base64.b64decode(raw)
else:
data = raw
dump_packet(data, args.path)
if __name__ == "__main__":
main()