yml2213
8efd3b8d02
- tools/huya_activity_verify.py: 认证帧+getActUserTaskDetail两帧即验证 (200请求成功+6任务项=已登录; 无cookie对照905=未登录) - 信封格式定论: L2=len(value), L1=L2+14, 身份一致性要求, actId尾字段=11 62 2f - HTTP cdnws端点被volc-dcdn封锁非浏览器TLS, WS wsapi可纯Python直连 - evidence: 金样本(逐字节)+ 浏览器当前会话帧 + 验证结果
313 lines
12 KiB
Python
313 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""虎牙活动页 webActUI TAF 服务端登录态验证 (纯Python, WS 通道)。
|
|
|
|
背景 (2026-08-26 实测):
|
|
- 活动页 zt.huya.com 的 webActUI 服务走 TAF 信封, 认证 = 信封内嵌完整 cookie 串。
|
|
- 两条通道:
|
|
* HTTP POST https://cdnws.api.huya.com/?baseinfo=<b64>×tamp=<ms>
|
|
响应 = base64(TAF信封)。**该端点被 volc-dcdn CDN 封锁非浏览器 TLS
|
|
(requests/curl/curl_cffi 均超时/502), 仅真实浏览器可直连。**
|
|
* WebSocket wss://wsapi.huya.com/?baseinfo=<b64> —— **纯 Python 可直连**,
|
|
本工具走这条通道。
|
|
- 认证/验证序列 (最小两帧, 已实测 200 请求成功):
|
|
* 认证帧 (裸 JCE struct): uid + UA + cookie(STR4), 向连接注册账号身份
|
|
* getActUserTaskDetail (webActUI TAF 信封, 按目标账号重建)
|
|
getActUserTaskDetail 返回 200 请求成功 + 用户任务列表
|
|
=> 服务端确认该 cookie 已登录; 无 cookie 对照 => 905 错误。
|
|
- 关键坑 (2026-08-26 逆向定位):
|
|
* baseinfo/认证帧/task 帧的 uid+cookie 必须全部一致, 否则服务端回
|
|
"wup request biz execption:-1" (iErrorCode 5000)。
|
|
* task 帧尾部 actId 字段 = JCE t1/INT16 标签+值 `11 62 2f` (3字节),
|
|
缺标签字节会解析失败。
|
|
* 活动页会自行补充页面级 cookie (guid/huya_ua/__yasmid/__yamid_* 等),
|
|
纯 Python 产出的 10 个登录 cookie 已足够, 无需页面 cookie。
|
|
|
|
用法:
|
|
python tools/huya_activity_verify.py [cookie文件] [账号块索引]
|
|
默认 cookie 文件: evidence/web_cookies_full.txt, 账号块 0。
|
|
"""
|
|
import base64
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import struct
|
|
import sys
|
|
import time
|
|
import urllib.parse
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
|
try:
|
|
from websocket import create_connection
|
|
except ImportError:
|
|
print("缺少依赖: pip install websocket-client")
|
|
sys.exit(1)
|
|
from tools.huya_wup_encoder import _Writer # noqa: E402
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
UA_ACT = "webh5&0.0.1&websocket&&diypc_52775"
|
|
UA_BROWSER = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# cookie 解析
|
|
# ---------------------------------------------------------------------------
|
|
def parse_cookie_blocks(path):
|
|
"""evidence/web_cookies_full.txt 多账号分块解析 (按空行分块, 行内以;分隔)。"""
|
|
blocks = []
|
|
cur = None
|
|
with open(path) as f:
|
|
for raw in f:
|
|
line = raw.strip()
|
|
if not line:
|
|
cur = None
|
|
continue
|
|
if line.startswith('#'):
|
|
cur = {'account': line.lstrip('#').strip(), 'cookies': {}}
|
|
blocks.append(cur)
|
|
continue
|
|
if cur is None:
|
|
cur = {'account': 'unknown', 'cookies': {}}
|
|
blocks.append(cur)
|
|
for pair in line.split(';'):
|
|
pair = pair.strip()
|
|
if not pair or '=' not in pair:
|
|
continue
|
|
k, v = pair.split('=', 1)
|
|
cur['cookies'][k.strip()] = v.strip()
|
|
return blocks
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TAF 组包 (webActUI 活动接口, 与浏览器字节一致, 金样本对拍通过)
|
|
# ---------------------------------------------------------------------------
|
|
def make_trace_id() -> str:
|
|
return f"{random.getrandbits(64):016x}:{random.getrandbits(64):016x}:0:0"
|
|
|
|
|
|
def build_baseinfo(uid: int, ua: str, cookie_str: str, trace: str) -> bytes:
|
|
"""cdnws/wsapi 的 baseinfo 原始字节 (JCE struct, b64进URL)。"""
|
|
w = _Writer()
|
|
w.int64(0, uid)
|
|
w.string(1, "")
|
|
w.string(2, ua)
|
|
w.string(3, "HUYA&ZH&2052")
|
|
w.string(4, "")
|
|
w.string(5, "")
|
|
w.int8(6, 0)
|
|
w.string(7, "")
|
|
w.string(8, cookie_str) # cookie (len>255 -> STR4)
|
|
w.string(9, trace)
|
|
w.map_begin(10, 0)
|
|
return w.get()
|
|
|
|
|
|
def build_body(servant: str, func: str, uid: int, ua: str, cookie_str: str,
|
|
trace: str, reqid: int = 1, func_tail: bytes = b"",
|
|
with_trace: bool = True) -> bytes:
|
|
"""构造 webActUI TAF 信封 (含4B长度前缀)。已与捕获金样本逐字节一致。"""
|
|
w = _Writer()
|
|
w.struct_begin(0)
|
|
w.struct_begin(0)
|
|
w.int64(0, uid)
|
|
w.string(1, "")
|
|
w.string(2, "")
|
|
w.string(3, ua)
|
|
w.string(4, cookie_str) # cookie (STR4)
|
|
w.int8(5, 0)
|
|
w.string(6, "")
|
|
w.struct_end()
|
|
w.buf += func_tail # func 特有尾部 (value 内, 如 actId)
|
|
w.struct_end()
|
|
value = w.get()
|
|
|
|
body = bytearray()
|
|
body += b"\x10\x03\x2c\x3c\x40" + bytes([reqid]) + b"\x56"
|
|
body += b"\x08" + servant.encode()
|
|
body += b"\x66" + bytes([len(func)]) + func.encode() + b"\x7d"
|
|
body += b"\x00\x01" + struct.pack(">H", len(value) + 14)
|
|
body += b"\x08\x00\x01\x06\x04tReq\x1d\x00\x01" + struct.pack(">H", len(value))
|
|
body += value
|
|
body += b"\x8c\x98\x0c\xa8\x0c"
|
|
if with_trace:
|
|
tr = trace.encode()
|
|
body += b"\x2c\x36" + bytes([len(tr)]) + tr + b"\x4c\x5c\x66\x00"
|
|
return struct.pack(">I", len(body)) + bytes(body)
|
|
|
|
|
|
def wrap_ws(inner: bytes) -> bytes:
|
|
"""4B前缀信封 -> WS 帧 (00 03 1d 00 01 [2B len][4B len][inner])。
|
|
|
|
len = 信封(去4B长度前缀) - 固定头(12 + svc长 + func长), 与捕获帧一致。
|
|
"""
|
|
e = inner[4:]
|
|
svc_len = e[7] # 56 <svc_len> <svc>
|
|
func_len = e[9 + svc_len] # 66 <flen> <func>
|
|
hdr = 12 + svc_len + func_len
|
|
ln = len(e) - hdr
|
|
return b"\x00\x03\x1d\x00\x01" + ln.to_bytes(2, "big") + ln.to_bytes(4, "big") + e
|
|
|
|
|
|
ACT_ID = 0x622F # 25135, 金样本 task 帧尾部 11 62 2f
|
|
ACT_ID_TAIL = b"\x11" + struct.pack(">H", ACT_ID) # JCE: t1/INT16 标签 + 值
|
|
|
|
|
|
def build_auth_frame(uid: int, ua: str, cookie_spaced: str) -> bytes:
|
|
"""认证帧 (裸 JCE struct, 无 TAF 信封; 注册连接的账号身份)。
|
|
|
|
结构 (与实抓逐字节一致): t0 uid + t1 UA + t2 cookie(STR4) + t3 空
|
|
+ t4 1 + t5 版本 + t2..t6 空尾字段。包装头 00 0a 1d 00 01 [2B len]。
|
|
"""
|
|
w = _Writer()
|
|
w.int64(0, uid)
|
|
w.string(1, ua)
|
|
w.string(2, cookie_spaced)
|
|
w.string(3, "")
|
|
w.int8(4, 1)
|
|
w.string(5, "HUYA&ZH&2052")
|
|
w.int8(2, 0)
|
|
w.string(3, "")
|
|
w.int8(4, 0)
|
|
w.int8(5, 0)
|
|
w.string(6, "")
|
|
body = w.get()
|
|
return b"\x00\x0a\x1d\x00\x01" + (len(body) - 7).to_bytes(2, "big") + body
|
|
|
|
|
|
def build_task_frame(uid: int, cookie_tight: str, reqid: int = 5) -> bytes:
|
|
inner = build_body("webActUI", "getActUserTaskDetail", uid, UA_ACT,
|
|
cookie_tight, make_trace_id(), reqid=reqid,
|
|
func_tail=ACT_ID_TAIL)
|
|
return wrap_ws(inner)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 响应解析 (验证只需状态/消息/条目)
|
|
# ---------------------------------------------------------------------------
|
|
def extract_msg(resp: bytes):
|
|
"""从 TAF 响应中提取消息串 (JCE t1 STR1, 常见: 请求成功/活动已下线/…)。"""
|
|
msgs = re.findall(rb"[\xe4-\xe9][\x80-\xbf][\x80-\xbf]", resp)
|
|
txt = "".join(m.decode() for m in msgs)
|
|
if not txt:
|
|
m = re.search(rb"STATUS_RESULT_DESC[\x16]\x1c([\x20-\x7e]+)", resp)
|
|
if m:
|
|
txt = m.group(1).decode(errors="replace")
|
|
return txt or None
|
|
|
|
|
|
def task_ok(resp: bytes) -> tuple:
|
|
"""返回 (is_ok, msg, item_count)。"""
|
|
msg = extract_msg(resp)
|
|
# 状态码: 成功 0a 01 0x00c8=200 / 活动下线 0a 01 0x0259 / 错误 01 0x0389=905 等
|
|
st = None
|
|
m = re.search(rb"\x0a\x01(..)", resp)
|
|
if m:
|
|
st = int.from_bytes(m.group(1), "big")
|
|
else:
|
|
m = re.search(rb"\x01(..)\x81\x13\x88", resp) # tars 错误: 业务码 + iErrorCode5000
|
|
if m:
|
|
st = int.from_bytes(m.group(1), "big")
|
|
# 任务条目 = value 内任务 STRUCT 头 (每任务以 0a 01 62 开头)
|
|
items = 0
|
|
if st == 200:
|
|
items = len(re.findall(rb"\x0a\x01\x62\x2f\x12\x00", resp)) or len(re.findall(rb"\x0a\x01\x62", resp))
|
|
return (st == 200 and "请求成功" in msg, st, msg, items)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 验证主流程
|
|
# ---------------------------------------------------------------------------
|
|
def verify_ws(uid: int, cookie_tight: str, with_cookie: bool = True,
|
|
timeout: float = 30) -> dict:
|
|
"""WS 通道验证登录态: 认证帧(注册连接身份) + getActUserTaskDetail。
|
|
|
|
经验 (2026-08-26 实测): 帧身份必须与连接 baseinfo 一致 ——
|
|
baseinfo/认证帧/task 帧必须同一 uid+cookie, 否则服务端回
|
|
"wup request biz execption:-1"。认证帧 + task 帧两帧即足够。
|
|
"""
|
|
cookie_spaced = "; ".join(p.strip() for p in cookie_tight.split(";") if "=" in p)
|
|
if not with_cookie:
|
|
cookie_spaced, cookie_tight = "", ""
|
|
bi = build_baseinfo(uid, UA_ACT, cookie_spaced, make_trace_id())
|
|
url = "wss://wsapi.huya.com/?" + urllib.parse.urlencode(
|
|
{"baseinfo": base64.b64encode(bi).decode()})
|
|
ws = create_connection(url, timeout=15, header={
|
|
"User-Agent": UA_BROWSER, "Origin": "https://zt.huya.com"})
|
|
out = {}
|
|
try:
|
|
ws.send(build_auth_frame(uid, UA_ACT, cookie_spaced), opcode=2)
|
|
time.sleep(0.3)
|
|
ws.send(build_task_frame(uid, cookie_tight, reqid=5), opcode=2)
|
|
time.sleep(0.5)
|
|
ws.settimeout(timeout)
|
|
deadline = time.time() + timeout
|
|
task_resp = None
|
|
while time.time() < deadline:
|
|
try:
|
|
r = ws.recv()
|
|
except Exception:
|
|
break
|
|
rp = r.encode("latin1") if isinstance(r, str) else r
|
|
if b"getActUserTaskDetail" in rp[:90]:
|
|
task_resp = rp
|
|
break
|
|
out["task_resp_len"] = len(task_resp) if task_resp else 0
|
|
if task_resp:
|
|
ok, st, msg, items = task_ok(task_resp)
|
|
out.update(ok=ok, status=st, msg=msg, items=items)
|
|
else:
|
|
out.update(ok=False, msg="无 getActUserTaskDetail 响应")
|
|
finally:
|
|
ws.close()
|
|
return out
|
|
|
|
|
|
def self_check() -> bool:
|
|
"""金样本回归: 与捕获 baseinfo/body 逐字节比对, 防 builder 漂移。"""
|
|
golden_path = os.path.join(HERE, "..", "evidence", "activity_taf_golden.json")
|
|
if not os.path.exists(golden_path):
|
|
print("[self_check] 无金样本 evidence/activity_taf_golden.json, 跳过")
|
|
return True
|
|
g = json.load(open(golden_path))
|
|
bi = build_baseinfo(g["uid"], g["ua"], g["cookie_spaced"], g["baseinfo_trace"])
|
|
assert bi == bytes.fromhex(g["baseinfo_hex"]), "baseinfo 与金样本不一致!"
|
|
body = build_body("webActUI", "getActUserTaskDetail", g["uid"], g["ua"],
|
|
g["cookie_tight"], g["trace"], reqid=g["reqid"],
|
|
func_tail=bytes.fromhex(g["func_tail_hex"]))
|
|
assert body == bytes.fromhex(g["body_hex"]), "body 与金样本不一致!"
|
|
auth = build_auth_frame(g["uid"], g["ua"], g.get("auth_cookie_spaced", g["cookie_spaced"]))
|
|
assert auth[7:] == bytes.fromhex(g.get("auth_hex", "")), "auth 帧与金样本不一致!"
|
|
print("[self_check] 通过: baseinfo/body/auth 与捕获金样本逐字节一致")
|
|
return True
|
|
|
|
|
|
def main():
|
|
self_check()
|
|
path = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
|
|
HERE, "..", "evidence", "web_cookies_full.txt")
|
|
idx = int(sys.argv[2]) if len(sys.argv) > 2 else 0
|
|
blocks = parse_cookie_blocks(path)
|
|
blk = blocks[idx]
|
|
ck = blk["cookies"]
|
|
uid = int(ck.get("udb_uid") or ck.get("yyuid") or 0)
|
|
cookie_tight = ";".join(f"{k}={v}" for k, v in ck.items())
|
|
print(f"== 账号 [{idx}] {blk['account']} uid={uid} cookies={len(ck)} ==")
|
|
|
|
print(f"\n-- [A] 带 cookie 调 getActUserTaskDetail (复刻浏览器序列) --")
|
|
ra = verify_ws(uid, cookie_tight)
|
|
print(f" 响应 {ra['task_resp_len']}B | status={ra.get('status')} msg={ra.get('msg')} "
|
|
f"任务项={ra.get('items')}")
|
|
print(" =>", "✅ 已登录 (服务端确认: 请求成功 + 用户任务数据)"
|
|
if ra.get("ok") else "❌ 未确认")
|
|
|
|
print(f"\n-- [B] 无 cookie 对照 (应失败/错误) --")
|
|
rb = verify_ws(uid, cookie_tight, with_cookie=False)
|
|
print(f" 响应 {rb['task_resp_len']}B | status={rb.get('status')} msg={rb.get('msg')}")
|
|
|
|
print("\n结论:", "✅ cookie 登录态服务端验证通过"
|
|
if ra.get("ok") else "❌ 未通过 (cookie 可能过期或活动已结束)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |