yml2213
eccc00e090
- docs/HUYA_ID_NOMENCLATURE.md: 规范总表 + 三条强制规则 + 代码变量对应 - 4 篇主文档头部去混声明 + §11.x 被证伪假设行标注更正 - 代码: GOLDEN_HDID→GOLDEN_HDID32, device_profile.HDID→HDID32, wup_encoder/app_login 注释前缀化 - 后续所有表述必须以 GUID32(doLaunch) / HDID32(登录t1.t0) / DEVID40(t5) 前缀书写
257 lines
6.9 KiB
Python
257 lines
6.9 KiB
Python
"""虎牙 App 密码登录 WUP(TAF) 请求帧编码器。
|
|
|
|
纯 Python 构造符合 TAF/WUP 协议的请求二进制帧。
|
|
"""
|
|
|
|
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)
|
|
self.int32(0, len(b))
|
|
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。"""
|
|
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。"""
|
|
if now_ms is None:
|
|
now_ms = int(_time.time() * 1000)
|
|
t1 = now_ms
|
|
t2 = now_ms + random.randint(200, 900)
|
|
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:
|
|
"""编码 _wup_data struct。"""
|
|
meta_json = _build_meta_json(session, trace_id)
|
|
name = _make_name(uid_str)
|
|
|
|
w.struct_begin(0)
|
|
|
|
# -- t0: 请求头 struct --
|
|
w.struct_begin(0)
|
|
w.int8(0, 0)
|
|
w.string(1, "1.0")
|
|
w.string(2, meta_json)
|
|
w.string(3, "5008")
|
|
w.int8(4, 3)
|
|
w.string(5, safedeviceid)
|
|
w.string(6, "")
|
|
w.string(7, "")
|
|
w.string(8, user_action_json)
|
|
w.string(9, "")
|
|
w.struct_end()
|
|
|
|
# -- t1: 设备信息 struct --
|
|
di = device_info
|
|
w.struct_begin(1)
|
|
w.string(0, hdid) # t1.t0 = HDID32 (登录帧32hex设备证书, libhydeviceid.so 硬锚, 换→APP_SIGN_NOT_MATCH)
|
|
w.string(1, di.get("app_version", "13.4.22"))
|
|
w.string(2, di.get("sdk_version", "1.0.80138"))
|
|
w.string(3, "")
|
|
w.string(4, di.get("ip", "127.0.0.1"))
|
|
w.string(5, di.get("vendor", "xiaomi"))
|
|
w.string(6, "")
|
|
w.struct_end()
|
|
|
|
# -- t2: 屏幕/设备特征 struct --
|
|
w.struct_begin(2)
|
|
w.int8(0, 1)
|
|
w.string(1, di.get("model", "M2102J2SC"))
|
|
w.string(2, di.get("fingerprint", ""))
|
|
w.string(3, di.get("os", "android"))
|
|
w.string(4, di.get("screen", "M2102J2SC,30,11"))
|
|
w.string(6, str(di.get("width", "1080")))
|
|
w.string(7, str(di.get("height", "2120")))
|
|
w.string(8, di.get("device_id", ""))
|
|
w.struct_end()
|
|
|
|
# -- 登录字段 --
|
|
w.string(3, name)
|
|
w.string(4, sha1_password)
|
|
w.list_begin(5, 1)
|
|
w.string(0, "5008")
|
|
w.int8(6, 1)
|
|
w.map_begin(7, 0)
|
|
w.bytes(8, b"")
|
|
|
|
w.struct_end()
|
|
|
|
|
|
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 请求体。"""
|
|
wd = _Writer()
|
|
_build_wup_data(wd, uid_str, sha1_password, safedeviceid, hdid,
|
|
session, trace_id, user_action_json, device_info)
|
|
wup_data = wd.get()
|
|
|
|
req = _Writer()
|
|
req.int32(0, session)
|
|
wupdbreq_v0 = req.get()
|
|
|
|
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()
|
|
|
|
w = _Writer()
|
|
w.int16(1, 3)
|
|
w.int8(2, 0)
|
|
w.int8(3, 0)
|
|
w.int32(4, session)
|
|
w.string(5, "huyaudbwebui")
|
|
w.string(6, "hypasswordLogin")
|
|
w.bytes(7, s_buffer)
|
|
w.int32(8, 0)
|
|
w.map_begin(9, 0)
|
|
w.map_begin(10, 0)
|
|
wup_body = w.get()
|
|
|
|
return struct.pack(">I", 4 + len(wup_body)) + wup_body
|