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

feat(huya): 分身 CK 支持 cred0 解密补全，WSS 帧改为 JCE WebSocketCommand

Browse Source 
- scripts/huya_clone_ck.py: root 读取小X分身/真身 WebView Cookie，默认
  用 AES-128-ECB(UDB 主密钥前 16B) 解 cred0 注入 udb_cred，新增
  --anon-device 补匿名设备字段、--no-cred0 退回旧行为
- core/huya/wss_client.py: 帧层改用浏览器 HUYA.WebSocketCommand(JCE)
  结构，vData 直接放 Wup 原包，按 WUP iRequestId 匹配响应，初始化顺序
  对齐 RegisterGroup→wsLaunch→VerifyCookie→getConfig→UpdateUserExps
- tests: 帧级断言改为新结构，假 WS 按 requestId 兑现
- docs: 新增分身 CK 提取、HAR 提取 CK 与积分中心说明

实测：四个分身 cred0 解出 udb_cred（39/43 字段），WSS
PlayMallUI.queryAllocation 返回 code=200、18 个商品；tests -k huya 394 passed
This commit is contained in:
yml2213
2026-10-26 11:23:29 +08:00
parent 6fc453017c
commit 03bc0818dd
9 changed files with 1188 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/shop_structs.py
+16 -4
View File
@@ -1178,11 +1178,13 @@ def parse_points_center_gifts(payload: dict | None) -> list[dict]:
→ tag4 分类名
→ tag5 图标
→ tag9/tag10 金额（分，不是积分）
→ tag13 库存
→ tag14 剩余库存（tag17[0].tag3 兜底）
→ tag13 累计兑换/销量
→ tag75.tag0.tag2 积分
→ tag34.tag26 积分（重复字段，兜底）
注意：``tag13`` 在这里是库存，不是 ``ActPrizeItem.price`` 的那个 tag13。
注意：2026-10-25 抓包里 ``tag14`` 才是剩余库存，和 SKU ``tag17[0].tag3``
一致；``tag13`` 是累计兑换/销量。旧抓包没有 ``tag14`` 时才回退 ``tag13``。
"""
if not isinstance(payload, dict):
return []
@@ -1234,14 +1236,24 @@ def parse_points_center_gifts(payload: dict | None) -> list[dict]:
else None
)
stock = _as_int(basic.get("tag13"))
# 2026-10-25 抓包：tag13 是累计兑换/销量，tag14 才是剩余库存，
# 且与 SKU 列表 tag17[0].tag3 一致；旧抓包没有 tag14 时回退 tag13。
claimed = _as_int(basic.get("tag13"))
stock = _as_int(basic.get("tag14"))
if stock is None:
skus = basic.get("tag17")
if isinstance(skus, list) and skus and isinstance(skus[0], dict):
stock = _as_int(skus[0].get("tag3"))
if stock is None:
stock = claimed
gifts.append(
{
"product_id": spu,
"name": name[:256],
"price": points,
"remain_text": f"剩{stock}" if stock is not None else "",
"remain_text": str(stock) if stock is not None else "",
"stock": stock,
"claimed": claimed,
"score": points,
"category_name": (_as_text(basic.get("tag4")) or category_name)[
:64
core/huya/wss_client.py
+153 -112
View File
@@ -18,8 +18,6 @@
import asyncio
import random
import re
import struct
from collections import deque
from collections.abc import Callable
from typing import Any, cast
@@ -41,8 +39,6 @@ ACTIVITY_WS_HOST = "d35bf373-ws.va.huya.com"
# wsLaunch tReq: lUid=0, sGuid="", sUA="webh5&0.0.1&websocket&&h5_/index.html"
SHOP_BASEINFO = "DBYAJiV3ZWJoNSYwLjAuMSZ3ZWJzb2NrZXQmJmg1Xy9pbmRleC5odG1sNgxIVVlBJlpIJjIwNTJGAFYAbHYAhgCWAKgM"
TAIL_BYTES = bytes.fromhex("2c36004c5c6600")
BIG_TAIL_SUFFIX = bytes.fromhex("4c5c6600")
WSS_BIZ_UA = "web&1.0.0&huya"
WSS_COOKIE_UA = "webh5&0.0.1&websocket&&diypc_52775"
WSS_SHOP_SCENE = 4
@@ -114,33 +110,81 @@ POINTS_ORDER_METHODS = frozenset({"createOrderV5", "payOrderSubmitV5"})
class WssCommand:
RPC_REQUEST = 0x03
RPC_RESPONSE = 0x04
HEARTBEAT = 0x05
HEARTBEAT_ACK = 0x06
AUTH = 0x0A
SERVER_PUSH1 = 0x0B
HEARTBEAT_SEND = 0x10
HEARTBEAT_RECV = 0x11
CONFIRM_SEND = 0x17
SERVER_PUSH2 = 0x18
REGISTER = 0x21
AUTH_RESPONSE = 0x0B
HEARTBEAT_SEND = 0x10 # EWSCmdC2S_RegisterGroupReq
HEARTBEAT_RECV = 0x11 # EWSCmdS2C_RegisterGroupRsp
CONFIRM_SEND = 0x17 # EWSCmdC2S_UpdateUserExpsReq
CONFIRM_RECV = 0x18
PING = 0x14 # EWSCmdC2S_HeartBeatReq
PING_ACK = 0x15 # EWSCmdS2C_HeartBeatRsp
REGISTER = 0x21 # EWSCmdC2S_UpdateUserInfoReq
class WssMessage:
def __init__(self, command: int, sequence: int, body: bytes):
"""浏览器 ``HUYA.WebSocketCommand``（TAF/JCE 结构，2026-10-25 对齐 huya-go）。
结构字段：
0 iCmdType(int32) 1 vData(bytes) 2 lRequestId(int64)
3 traceId(string) 4 iEncryptType(int32) 5 lTime(int64) 6 sMD5(string)
``sequence`` 保留旧命名，实际承载 ``lRequestId``；``body`` 即 ``vData``。
"""
def __init__(
self,
command: int = 0,
sequence: int = 0,
body: bytes = b"",
*,
trace_id: str = "",
encrypt_type: int = 0,
timestamp: int = 0,
md5: str = "",
):
self.command = command
self.sequence = sequence
self.body = body
self.trace_id = trace_id
self.encrypt_type = encrypt_type
self.timestamp = timestamp
self.md5 = md5
def encode(self) -> bytes:
header = struct.pack(">BBI", 0x00, self.command, self.sequence)
return header + self.body
os = TafOutputStream()
os.write_int32(0, int(self.command))
os.write_bytes(1, self.body or b"")
os.write_int64(2, int(self.sequence))
os.write_string(3, self.trace_id)
os.write_int32(4, int(self.encrypt_type))
os.write_int64(5, int(self.timestamp))
os.write_string(6, self.md5)
return os.get_bytes()
@classmethod
def decode(cls, data: bytes) -> "WssMessage":
if len(data) < 6:
if len(data) < 2:
raise ValueError(f"消息太短: {len(data)} bytes")
_version, command = struct.unpack(">BB", data[0:2])
sequence = struct.unpack(">I", data[2:6])[0]
body = data[6:]
return cls(command=command, sequence=sequence, body=body)
ins = TafInputStream(data)
# 必须按结构字段顺序读取：tag0 命令 → tag1 数据 → tag2 requestId → ...
command = ins.read_int32(0)
body = ins.read_bytes(1)
request_id = ins.read_int64(2)
trace_id = ins.read_string(3)
encrypt_type = ins.read_int32(4)
timestamp = ins.read_int64(5)
md5 = ins.read_string(6)
return cls(
command=command,
sequence=request_id,
body=body,
trace_id=trace_id,
encrypt_type=encrypt_type,
timestamp=timestamp,
md5=md5,
)
class HuyaWssClient:
@@ -153,7 +197,9 @@ class HuyaWssClient:
self.ws: Any = None
self._biz_seq = SEQ_BUSINESS # 业务 RPC 递增用
self._wup_req_id = 8
self.rpc_queue = deque()
# 按 WUP iRequestId 匹配响应（浏览器/HAR 实测：服务端回包 requestID 与请求一致）
self._pending: dict[int, asyncio.Future] = {}
self.logger = logger or print
self._recv_task = None
self._heartbeat_task = None
@@ -183,49 +229,24 @@ class HuyaWssClient:
return req_id
@staticmethod
def _make_big_tail() -> bytes:
"""生成 WSS 大包尾部 trace（格式与浏览器 HAR 一致）"""
def _new_trace_id() -> str:
"""浏览器帧 traceId: ``<hex>:<hex>:0:<enableTrace>``。"""
trace = f"{random.getrandbits(64):016x}"
return f",6%{trace}:{trace}:0:3".encode("ascii") + BIG_TAIL_SUFFIX
return f"{trace}:{trace}:0:0"
@staticmethod
def _encode_rpc_body(wup_data: bytes) -> bytes:
"""按浏览器规则封装 WSS RPC body
"""业务 RPC 的 vData 就是 WUP 原包（长度前缀 + WUP 结构）。
小包: [4B WUP长度][WUP body] + TAIL_BYTES
大包: [1B len低字节][4B WUP body长度][裸WUP body] + trace尾部
2026-10-25 对齐 huya-go：浏览器把 ``Wup.encode()`` 的结果原样放进
``WebSocketCommand.vData``，不再拼 trace/尾部字节。
"""
if len(wup_data) < 4:
return wup_data + TAIL_BYTES
declared_len = int.from_bytes(wup_data[:4], "big")
if declared_len == len(wup_data):
wup_body = wup_data[4:]
else:
wup_body = wup_data
if len(wup_body) <= 0xFF:
return wup_data + TAIL_BYTES
# 大包头里的长度使用原始 WUP 包长度（含 4B 长度前缀），但正文只放裸 WUP body；
# 浏览器随后用 trace 尾部的前 4 字节补足该长度窗口。
packet_len = len(wup_data)
return (
bytes([packet_len & 0xFF])
+ packet_len.to_bytes(4, "big")
+ wup_body
+ HuyaWssClient._make_big_tail()
)
return wup_data
@staticmethod
def _encode_activity_launch_body(wup_data: bytes) -> bytes:
"""活动 wsLaunch 使用浏览器的固定大包头，即使包体小于 255B。"""
if len(wup_data) < 4:
return wup_data + TAIL_BYTES
packet_len = len(wup_data)
return (
bytes([packet_len & 0xFF])
+ packet_len.to_bytes(4, "big")
+ wup_data[4:]
+ TAIL_BYTES
)
"""wsLaunch 同样直接使用 WUP 原包。"""
return wup_data
@staticmethod
def _normalize_biz_cookie(cookie: str) -> str:
@@ -402,42 +423,65 @@ class HuyaWssClient:
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[WSS] 接收循环异常: {type(e).__name__}: {e}")
@staticmethod
def _wup_request_id(body: bytes) -> int | None:
"""从 WSS 回包 vData 里解出 WUP iRequestId（失败返回 None）。"""
try:
wup = WupResponse()
wup.decode(HuyaWssClient._extract_wup(body))
return int(wup.iRequestId)
except Exception: # noqa: BLE001 外部响应可能不完整，按未匹配处理
return None
async def _handle_message(self, msg: WssMessage):
if msg.command == WssCommand.RPC_RESPONSE:
if self.rpc_queue:
future = self.rpc_queue.popleft()
if not future.done():
future.set_result(msg.body)
request_id = self._wup_request_id(msg.body)
future = (
self._pending.pop(request_id, None) if request_id is not None else None
)
if future is None and len(self._pending) == 1:
# 兜底：服务端偶尔不回原始 requestId，只剩一个在途请求时直接认领
fallback_id, fallback = next(iter(self._pending.items()))
self._pending.pop(fallback_id, None)
future = fallback
if future is not None and not future.done():
future.set_result(msg.body)
else:
self.logger(f"[WSS] 未匹配的 RPC 响应 seq={msg.sequence}")
elif msg.command == WssCommand.HEARTBEAT_RECV:
pass # 心跳已在上层 _recv_loop 打过日志
elif msg.command in (WssCommand.SERVER_PUSH1, WssCommand.SERVER_PUSH2):
pass # 推送已在上层打过日志
else:
pass # 未知消息已在上层打过日志
self.logger(f"[WSS] 未匹配的 RPC 响应 requestId={request_id}")
elif msg.command in (
WssCommand.PING_ACK,
WssCommand.HEARTBEAT_ACK,
WssCommand.HEARTBEAT_RECV,
WssCommand.AUTH_RESPONSE,
WssCommand.CONFIRM_RECV,
):
pass # 控制类响应无需处理，日志已在 _recv_loop 打出
async def _heartbeat_loop(self):
try:
while True:
await asyncio.sleep(8)
await self.send_heartbeat()
await self.send_ping()
except asyncio.CancelledError:
pass
except Exception as e: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
self.logger(f"[WSS] 心跳循环异常: {e}")
async def send_ping(self):
"""EWSCmdC2S_HeartBeatReq(20)，vData 为空（对齐 huya-go）。"""
msg = WssMessage(command=WssCommand.PING, sequence=0, body=b"")
await self.ws.send(msg.encode())
async def send_heartbeat(self):
# body = LIST(tag0,count=1) + 元素{tag0="live:0", tag1=""} + TAIL
"""EWSCmdC2S_RegisterGroupReq(16)：vData = list{"live:0"} + token。"""
os = TafOutputStream()
os.write_head(0, TafType.LIST) # 09
os.write_int32(0, 1) # 00 01 (count=1)
os.write_string(0, "live:0") # 06 06 live:0
os.buf.write(b"\x16\x00") # 16 00 (tag1 "")
body = os.get_bytes() + TAIL_BYTES
msg = WssMessage(
command=WssCommand.HEARTBEAT_SEND, sequence=SEQ_HEARTBEAT, body=body
)
os.write_string(1, "") # 16 00
body = os.get_bytes()
msg = WssMessage(command=WssCommand.HEARTBEAT_SEND, sequence=0, body=body)
self.logger(
format_wss_log(body, WssCommand.HEARTBEAT_SEND, SEQ_HEARTBEAT, "发")
)
@@ -530,18 +574,15 @@ class HuyaWssClient:
wup.iTimeout = 0
wup.newdata["tReq"] = os.get_bytes()
body = self._encode_activity_launch_body(wup.encode())
msg = WssMessage(WssCommand.RPC_REQUEST, SEQ_ACTIVITY_WSLAUNCH, body)
msg = WssMessage(WssCommand.RPC_REQUEST, 0, body)
future = asyncio.Future()
self.rpc_queue.append(future)
self.logger(
format_wss_log(body, WssCommand.RPC_REQUEST, SEQ_ACTIVITY_WSLAUNCH, "发")
)
self._pending[-1] = future
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, -1, "发"))
await self.ws.send(msg.encode())
try:
response = await asyncio.wait_for(future, timeout)
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self._pending.pop(-1, None)
return None
self.logger(
format_wss_log(
@@ -601,18 +642,16 @@ class HuyaWssClient:
wup_data = wup.encode()
body = self._encode_rpc_body(wup_data)
seq = SEQ_ACTIVITY_WSLAUNCH
msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body)
msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=0, body=body)
future = asyncio.Future()
self.rpc_queue.append(future)
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发"))
self._pending[-1] = future
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, -1, "发"))
await self.ws.send(msg.encode())
try:
body = await asyncio.wait_for(future, timeout)
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self._pending.pop(-1, None)
self.logger("[✗] wsLaunch 超时")
return None
@@ -716,8 +755,6 @@ class HuyaWssClient:
``ua`` 可覆盖：积分中心页面为 ``h5_/points.html``（2026-10-13 抓包）。
"""
os = TafOutputStream()
if self._activity_mode or self._points_mode:
os.write_head(3, TafType.ZERO)
os.write_int64(0, int(uid or 0))
os.write_string(1, ua or WSS_COOKIE_UA)
os.write_string(2, HuyaWssClient._normalize_biz_cookie(cookie))
@@ -728,13 +765,12 @@ class HuyaWssClient:
or self._launch_guid
)
os.write_string(3, auth_guid)
os.write_int8(4, 1)
os.write_int32(4, 1) # AutoRegisterUID
os.write_string(5, "HUYA&ZH&2052")
os.write_string(6, "")
auth_body = os.get_bytes() + TAIL_BYTES
auth_body = os.get_bytes()
msg = WssMessage(
command=WssCommand.AUTH,
sequence=sequence,
sequence=0,
body=auth_body,
)
await self.ws.send(msg.encode())
@@ -796,8 +832,8 @@ class HuyaWssClient:
os.write_struct_end()
os.write_map(7, {})
os.write_int32(8, 0)
body = os.get_bytes() + TAIL_BYTES
msg = WssMessage(command=WssCommand.REGISTER, sequence=SEQ_REGISTER, body=body)
body = os.get_bytes()
msg = WssMessage(command=WssCommand.REGISTER, sequence=0, body=body)
self.logger(format_wss_log(body, WssCommand.REGISTER, SEQ_REGISTER, "发"))
await self.ws.send(msg.encode())
@@ -809,10 +845,8 @@ class HuyaWssClient:
"""
os = TafOutputStream()
os.write_map(0, version_map or {})
body = os.get_bytes() + TAIL_BYTES
msg = WssMessage(
command=WssCommand.CONFIRM_SEND, sequence=SEQ_CONFIRM, body=body
)
body = os.get_bytes()
msg = WssMessage(command=WssCommand.CONFIRM_SEND, sequence=0, body=body)
self.logger(format_wss_log(body, WssCommand.CONFIRM_SEND, SEQ_CONFIRM, "发"))
await self.ws.send(msg.encode())
@@ -880,28 +914,32 @@ class HuyaWssClient:
rsp_class=None,
timeout: float = 10.0,
):
seq = self._next_biz_seq(method in POINTS_ORDER_METHODS)
wup = WupRequest()
wup.setServant(service)
wup.setFunc(method)
wup.setRequestId(self._next_wup_req_id())
req_id = self._next_wup_req_id()
wup.setRequestId(req_id)
wup.iTimeout = 0
wup.writeStruct("tReq", req_struct)
wup_data = wup.encode()
body = self._encode_rpc_body(wup_data)
msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body)
# 浏览器帧的 lRequestId 固定 0，区分请求靠 traceId + WUP iRequestId
msg = WssMessage(
command=WssCommand.RPC_REQUEST,
sequence=0,
body=body,
trace_id=self._new_trace_id(),
)
future = asyncio.Future()
self.rpc_queue.append(future)
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发"))
self._pending[req_id] = future
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, req_id, "发"))
await self.ws.send(msg.encode())
try:
body = await asyncio.wait_for(future, timeout)
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self._pending.pop(req_id, None)
self.logger(f"[✗] {service}.{method} 超时")
return None
@@ -1302,28 +1340,31 @@ class HuyaWssClient:
os.write_struct_end()
seq = self._next_biz_seq(True)
wup = WupRequest()
wup.setServant("shopMiddleUI")
wup.setFunc("payOrderSubmitV5")
wup.setRequestId(self._next_wup_req_id())
req_id = self._next_wup_req_id()
wup.setRequestId(req_id)
wup.iTimeout = 0
wup.newdata["tReq"] = os.get_bytes()
wup_data = wup.encode()
body = self._encode_rpc_body(wup_data)
msg = WssMessage(command=WssCommand.RPC_REQUEST, sequence=seq, body=body)
msg = WssMessage(
command=WssCommand.RPC_REQUEST,
sequence=0,
body=body,
trace_id=self._new_trace_id(),
)
future = asyncio.Future()
self.rpc_queue.append(future)
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, seq, "发"))
self._pending[req_id] = future
self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, req_id, "发"))
await self.ws.send(msg.encode())
try:
body = await asyncio.wait_for(future, 15.0)
except TimeoutError:
if future in self.rpc_queue:
self.rpc_queue.remove(future)
self._pending.pop(req_id, None)
self.logger("[✗] payOrderSubmitV5 超时")
return None
docs/虎牙/HAR提取CK与积分中心.md
+127
View File
@@ -0,0 +1,127 @@
# 精英宝典 HAR：完整 CK 与积分中心商品
> 抓包：`/Users/yml/Desktop/抓包/hy/10.25/精英宝典日常皮肤-1.har`
> 时间：2026-10-25 · 页面：`m-shop.huya.com/points.html?pagecode=heping_baodian_credit_center_page&sourceId=yellowcarlist`
## 一、结论
- 这是 **Web H5 抓包**（WebInspector，117 条请求），不是 App 抓包。
- 请求里的 `PlayMallUI.queryAllocation` 带 **完整浏览器 CK**：
- 原始 `tReq.tag0.tag4` 有 79 个键（含重复），去重后 **43 个字段 / 2338B**；
- 包含 `udb_cred`、`udb_biztoken`、`udb_deviceid`、`udb_guiddata`、`udb_anobiztoken`；
- 9 个 WSS 必填字段**全齐**，可以直接走 WSS。
- 账号：`hy_302081117` / UID `1199668650719`。
- 当前积分中心返回 **18 个「精英专享」商品**；现有 `parse_points_center_gifts()` 解析结果与抓包一致，核心解析代码不需要改。
- 赛季：配置 `scoreCenterConfigId=15`，活动 ID `3203`，`精英宝典福利季丨Q4`，有效期 **2026-10-09 00:00 → 2026-12-31 23:59:59（UTC+8）**。
## 二、HAR 里哪些请求有用
真正有业务数据的只有 7 条 `cdnws.api.huya.com` 的 WUP 请求：
| 函数 | 作用 |
|---|---|
| `launch.wsLaunch` | 建立 WSS 会话，拿服务端地址 |
| `launch.pingSignal` | 心跳 |
| `huyauserui.getUserProfile` | 用户资料 |
| `wupui.getUserFollowWeChatOfficialStatus` | 微信关注状态 |
| `PlayMallUI.queryAllocation` | **积分中心商品/配置（核心）** |
| `launch.queryHttpDns` | HTTP DNS |
其余是静态资源、埋点 `ylog/metric/statwup`、商城图片等。
## 三、完整 CK 在哪
`queryAllocation` 请求解码后：
```text
tReq.tag0.tag0 = 1199668650719 # UID
tReq.tag0.tag4 = <完整 document.cookie>
tReq.tag2 = heping_baodian_credit_center_page
tReq.tag1.tag6 = yellowcarlist
```
注意：HAR 里的原始 Cookie 有大量重复键（79 个），必须先按 key 去重、保留最后一次出现的值，得到 43 个字段。直接把原始 4090B 字符串塞进系统会带旧值。
脚本 `scripts/huya_har_ck.py` 会自动做这一步。
```bash
# 摘要 + 18 个商品
.venv/bin/python scripts/huya_har_ck.py \
"/Users/yml/Desktop/抓包/hy/10.25/精英宝典日常皮肤-1.har" --goods
# 落盘为 `账号----ck`（权限 0600）
.venv/bin/python scripts/huya_har_ck.py \
"/Users/yml/Desktop/抓包/hy/10.25/精英宝典日常皮肤-1.har" \
--out /tmp/huya_har_ck.txt
# JSON（默认不含完整 CK；加 --print 才包含）
.venv/bin/python scripts/huya_har_ck.py xxx.har --json
```
本次实测已生成 `/tmp/huya_har_ck.txt`：
```text
hy_302081117----<完整 CK>
```
## 四、18 个精英专享商品（2026-10-25 抓包快照）
| # | SPU | 名称 | 积分 | 库存 |
|---|---|---|---|---|
| 01 | `hy-8328509` | 套装-雪山精英 | 600 | 5054 |
| 02 | `hy-8347511` | 套装-青涩年华 | 488 | 6920 |
| 03 | `hy-8327363` | 套装-暗影哥特 | 888 | 3293 |
| 04 | `hy-8349003` | 套装-浪漫波比 | 1280 | 2207 |
| 05 | `hy-8315965` | 套装-浪漫天命 | 360 | 3298 |
| 06 | `hy-8348009` | 套装-牛仔情诗 | 550 | 454 |
| 07 | `hy-8335113` | 黑色高级特训官上衣 | 360 | 800 |
| 08 | `hy-8315982` | 幸运币礼包（小） | 20 | 9235 |
| 09 | `hy-8344526` | 荣耀勋章礼包（小） | 20 | 1951 |
| 10 | `hy-8338110` | 套装-甜星爱神 | 1280 | 246 |
| 11 | `hy-8335118` | 荣耀勋章礼包（中） | 300 | 448 |
| 12 | `hy-8329225` | 荣耀勋章礼包（大） | 900 | 549 |
| 13 | `hy-8329234` | 幸运币礼包（中） | 300 | 2301 |
| 14 | `hy-8328508` | 幸运币礼包（大） | 900 | 6647 |
| 15 | `hy-8348012` | 套装-甜心信使 | 1280 | 197 |
| 16 | `hy-8331702` | 浪漫波比头盔 | 1680 | 28 |
| 17 | `hy-8346046` | 浪漫波比背包 | 1680 | 442 |
| 18 | `hy-8324889` | 蜘蛛突变者大礼包 | 5500 | 78 |
> 积分取 `tag34.tag26`（600/488/888...）。同一商品里的 `tag9/tag10` 是 6000/4880/8880，别误当成积分；现有解析器用的就是 `tag34.tag26`，实测与页面一致。
>
> 库存取 `tag14`（与 SKU `tag17[0].tag3` 一致）。`tag13` 在本次抓包里是累计兑换/销量（例如雪山精英 `tag13=43194`、`tag14=5054`），不能当剩余库存；旧抓包无 `tag14` 时解析器回退 `tag13`。
## 五、赛季与日常任务
`queryAllocation` 响应 `tag2[1].tag2_2`：
- `tag0`：配置 `15`、名称 `精英宝典福利季`、副标题 `精英宝典福利季丨Q4`、活动 `3203`；
- `tag10/tag11`：`1791475200000` / `1798732799000`；
- `tag4`：`{8000, 0, 0, 1, 0}`（页面侧积分/状态快照）。
日常任务（`tag1`）：
1. 开通宝典并订阅「和平精英福利提醒」→ **5 积分**；
2. 开通宝典并每天观看和平精英直播 10 分钟 → **2 积分**。
任务和商品库存都会变，实际执行前以实时 `queryAllocation` 为准。
## 六、验证
用 HAR 提取的完整 CK 实时调用现有 HTTP 通道：
```text
play_mall_query_allocation(uid=1199668650719, deep=True)
→ code=200
→ parse_points_center_gifts → 18 个商品
→ 第一个：套装-雪山精英 600 积分
```
本次抓包的 CK 已通过该验证；由于 `udb_cred` 存在，也可以走 WSS。
## 七、注意事项
- `/tmp/huya_har_ck.txt` 是登录凭据，文件权限已 `0600`；导入系统后请删除。
- HAR 本身也含完整 Cookie，不要提交 Git、不要外发。
- 商品库存是抓包瞬间的值，正式下单前必须重新查询。
- `scripts/*` 被 `.gitignore` 忽略；脚本是本地工具，若要入库需 `git add -f scripts/huya_har_ck.py`。
docs/虎牙/分身CK提取.md
+190
View File
@@ -0,0 +1,190 @@
# 小X分身虎牙 CK 提取（Android / root）
> 2026-10-25 实测设备：OnePlus PHK110，Android 16，SukiSU Ultra v4.2.1。
> 结论：**root 后读每个分身的 WebView Cookies SQLite 即可拿到可用 CK**；不需要改 App，也不需要 Frida。
## 一、设备上的分身
小X分身并不是把多个账号塞进同一个进程，而是给每个分身分配独立的包/UID：
| 实例 | 进程（示例） | UID | 数据目录 |
|---|---|---|---|
| 主分身 1 | `com.bly.dkplat:p0` 内的 `com.duowan.kiwi` | 10363 | `/data/user/0/com.bly.dkplat/chaos/data/user/1/com.duowan.kiwi` |
| 主分身 2 | `com.bly.dkplat:p3` 内的 `com.duowan.kiwi` | 10363 | `/data/user/0/com.bly.dkplat/chaos/data/user/2/com.duowan.kiwi` |
| 独立分身 A | `dkplugin.duq.lls:p0` 内的 `com.duowan.kiwi` | 10365 | `/data/user/0/dkplugin.duq.lls/virtual/data/user/0/com.duowan.kiwi` |
| 独立分身 B | `dkplugin.zoc.bcn:p0` 内的 `com.duowan.kiwi` | 10366 | `/data/user/0/dkplugin.zoc.bcn/virtual/data/user/0/com.duowan.kiwi` |
| 真身（可选） | `com.duowan.kiwi` | 10364 | `/data/user/0/com.duowan.kiwi` |
每个分身真正登录态落在：
```text
<数据目录>/app_webview/Default/Cookies
```
这是 WebView 的 SQLite Cookie 库，`value` 是明文（当前设备上 `encrypted_value` 全为 0）。
## 二、为什么必须 root
无 root 时以下路径都走不通（已逐项实测）：
- `run-as`：分身和真身都 `not debuggable`
- `am dumpheap`：`Process not debuggable`
- `content query`：`CFileProvider` / `DKFileProvider` 均未导出
- `/sdcard/Android/data/...`：只有 cache / 日志，没有 Cookie
- WebView 调试：`/proc/net/unix` 没有 `webview_devtools_remote`
- `adb backup`：包没有 `ALLOW_BACKUP`，且 Android 16 对 targetSdk 33+ 的限制
root 后直接 `cat` 上面的 `Cookies` 即可，不需要停 App。
## 三、一键提取
脚本：`scripts/huya_clone_ck.py`
```bash
# 只提 4 个分身，终端看脱敏摘要
.venv/bin/python scripts/huya_clone_ck.py
# 连真身一起提，落盘为 `账号----ck`（文件权限 0600）
.venv/bin/python scripts/huya_clone_ck.py --include-real --out /tmp/huya_cks.txt
# 需要直接看完整 CK 时（会打印到终端，注意不要留 shell 历史）
.venv/bin/python scripts/huya_clone_ck.py --include-real --print
# 用活动积分接口验证（sid 换成当前活动；活动结束返回 603 也算登录态被接受）
.venv/bin/python scripts/huya_clone_ck.py --include-real --verify-sid 2203
```
脚本行为：
1. `adb shell su -c id` 必须返回 `uid=0`，否则直接报错并提示去 SukiSU 授权；
2. 自动发现 `dkplugin.*` 的 virtual 数据目录，以及 `com.bly.dkplat/chaos/data/user/*/com.duowan.kiwi`；
3. `cat` 每个分身的 `Cookies` 到临时文件，用 SQLite 读取 `host_key like '%huya.com'` 的非空 Cookie；
4. 默认调用现有 `fill_web_cookie_fields()` 补齐网页设备态，并把 `udb_appid` 修正为 `5002`；
5. 按 `(udb_uid, udb_passport)` 去重，避免真身/分身重复账号；
6. 可用 `--out` 输出 `账号----ck`，直接贴进协作台账的「贴 / 换 CK」。
> 脚本默认**不打印完整 CK**，只打印实例、UID、账号、字段数、WSS 是否可用和验证结果；必须显式 `--print` 或 `--out` 才拿明文。
## 四、实测结果（2026-10-25）
| 实例 | UID | 账号 | Cookie 字段 | 接口验证 |
|---|---|---|---|---|
| `clone:dkplat/1` | 1199668927843 | `hy_302407019` | 38 | 603 活动已结束 |
| `clone:dkplat/2` | 1199668931394 | `hy_302409741` | 38 | 603 活动已结束 |
| `clone:duq.lls` | 1199668932116 | `hy_302407828` | 38 | 603 活动已结束 |
| `clone:zoc.bcn` | 1199668932128 | `hy_302407842` | 38 | 603 活动已结束 |
| `real:huya` | 1199668930633 | `hy_302408724` | 38 | 603 活动已结束 |
`query_user_score(sid=2203)` 返回 `603 活动已结束` 说明请求已经带着有效登录态进入了业务层；同一接口用伪造 Cookie 调用时没有 `tRsp` 数据，可直接区分。
## 五、已知限制：WSS 字段不全
虎牙 App 的 `udb_cred` 不在 WebView Cookie 库里，而在：
```text
<数据目录>/files/huyaudb/cred0
```
`cred0` 是加密落盘（`UDB_SHAREAPP_DATA` / `credAnonymous` 同样加密）。**2026-10-25 已实现离线解密，见第八节**；下面三条是解密之前的历史结论：
- **HTTP cdnws 通道可用**：实测 `getUserScore` 能正常进入业务层；
- ~~**WSS 通道不可用**~~：当时缺 `udb_cred` / `udb_deviceid` / `udb_guiddata` / `udb_anobiztoken`，会回退 HTTP；现已由 cred0 + 匿名设备态协议补齐；
- ~~需要 Frida hook 或重新签发 cred~~：不需要，直接 AES-128-ECB 解密即可。
- 另一种拿到完整 WSS CK 的方式是从已登录的 H5 抓包提取，见 `docs/虎牙/HAR提取CK与积分中心.md`；注意 `udb_cred` 是账号级凭据，不能从别的账号挪用。
## 六、注意事项
- CK 就是登录凭据，`--out` 生成的文件权限已设为 `0600`；不要提交到 Git、不要发群。
- 不要对分身执行 `pm clear` / 卸载 / 清数据，否则 Cookie 和登录态一起没了。
- 虎牙 App 更新后 Cookie 库结构一般不变；如果 `Cookies` 变成 `encrypted_value`，脚本只会在备注里提示，不会输出错误 CK。
- root 授权被回收后，`su: inaccessible`，重新去 SukiSU 给 Shell 授权即可。
## 七、手机 CK 转网页 CK 实测（2026-10-25）
用一个小X分身账号做了完整链路验证，测试过程中不输出 CK 值：
| 步骤 | 结果 |
|---|---|
| WebView 原始 CK | 17 字段；无 `udb_cred` / `guid` / `udb_deviceid` / `udb_guiddata` / `udb_anobiztoken` |
| `fill_web_cookie_fields()` 补齐后 | 38 字段；仍缺 `udb_cred` / `udb_deviceid` / `udb_guiddata` / `udb_anobiztoken` |
| `checkLogin`（原始 CK） | HTTP 200，`isLogined=true`，UID 匹配 |
| `POST /web/cookie/verify` | HTTP 200，`returnCode=0`，`data=null`，**无 Set-Cookie** |
| 打开积分中心 H5 | HTTP 200，**无 Set-Cookie**，Cookie 数量不变 |
| `PlayMallUI.queryAllocation`（原始 CK） | `code=200`，`msg=成功`，18 个商品 |
| `PlayMallUI.queryAllocation`（38 字段 CK） | `code=200`，`msg=成功`，18 个商品 |
| 网页请求后的 CK 再查询 | `code=200`，`msg=成功`，18 个商品 |
结论：
1. **手机 WebView 的 CK 本身已经能作为虎牙网页 HTTP CK 使用**，可以直接调用积分中心等 H5/HTTP 活动接口。
2. 虎牙**没有**提供一个“提交手机 CK、由服务端下发完整网页 CK”的兑换接口。`web/cookie/verify` 成功时不回 `Set-Cookie`，积分中心页面也不会补发登录 Cookie。
3. 本地 `fill_web_cookie_fields()` 只补齐 `guid`、QIMEI、页面统计等网页运行字段，不会生成或取得 `udb_cred`。
4. 因此手机 CK 可以稳定支持 **HTTP 活动查询**；依赖完整 9 字段的 WSS、部分领取/兑换类动作仍不可用，除非另外解密 `cred0` 或从已登录 H5 抓包取得 `udb_cred`。
5. 本次只验证了只读活动查询，没有执行积分领取、兑换或支付。
## 八、cred0 离线解密：手机 CK 直接升级为完整 WSS CK（2026-10-25）
### 8.1 结论
`files/huyaudb/cred0` 是 **AES-128-ECB + 零填充（zero padding）** 加密的登录历史 JSON，
密钥是虎牙 UDB 主密钥的前 16 字节：
```text
HuyaUdb1928374650qwertyuiop[:16] → "HuyaUdb192837465"
```
解密后结构（`loginHistory[0].apploginData`）：
| 字段 | 含义 |
|---|---|
| `cred` | 152 字符 base64url，**就是 `udb_cred`**（114 字节 hyCred） |
| `uid` / `passport` | 账号 UID 与 `hy_xxxx` 账号 |
| `cookie.biztoken` | 网页 `udb_biztoken`（344 字符） |
| `cookie.status` / `cookie.version` | `udb_status` / `udb_version` |
| `biztoken_vec[0].biztoken` | App 通道 216 字符 biztoken（与网页不同） |
同一套密钥也能解 `credAnonymous`（匿名态）和 `UDB_SHAREAPP_DATA`。
### 8.2 脚本支持
```bash
# 默认：WebView Cookie + cred0 凭据 + 网页设备态补齐（39 字段）
.venv/bin/python scripts/huya_clone_ck.py --serial <serial> --json
# 完整 WSS CK：再加匿名设备态四字段（43 字段）
.venv/bin/python scripts/huya_clone_ck.py --serial <serial> --anon-device --out /tmp/cks.txt
```
新增参数：
- `--no-cred0`：只取 WebView Cookie，不解密 cred0；
- `--anon-device`：额外走 `core/huya/anon_device.py` 的 middle + anonymousLogin，
补齐 `udb_deviceid` / `udb_guiddata` / `udb_anobiztoken` / `udb_anouid`（需要外网）。
JSON 摘要新增 `wss_ready`（登录态字段齐全）与 `missing_device`（待匿名设备态补齐的字段）。
### 8.3 实测（2026-10-25，四个分身）
| 实例 | 账号 | 无 `--anon-device` | 加 `--anon-device` | WSS `PlayMallUI.queryAllocation` |
|---|---|---|---|---|
| `clone:dkplat/1` | `hy_302407019` | 39 字段，有 `udb_cred` | 43 字段 | `code=200`，18 个商品 |
| `clone:dkplat/2` | `hy_302409741` | 39 | 43 | 未单独跑（同链路） |
| `clone:duq.lls` | `hy_302407828` | 39 | 43 | `code=200`，18 个商品 |
| `clone:zoc.bcn` | `hy_302407842` | 39 | 43 | 未单独跑（同链路） |
对照：同一个 CK 在 Go 参考实现（`~/codes/huya-go`）里调用
`webActUI.getActPrizeList(sid=2277)` 也是 `status=200 / 18 项`，两边商品一致。
### 8.4 顺带修掉的 WSS 帧层 Bug
原先 `core/huya/wss_client.py` 用「6 字节自定义头 + 尾部字节」拼帧，服务端对
`launch.wsLaunch` 一直不回包（日志 `wsLaunch 无响应`）。按 `huya-go` 的实现改为浏览器真正的
`HUYA.WebSocketCommand`（TAF/JCE 结构：`iCmdType` / `vData` / `lRequestId` / `traceId` /
`iEncryptType` / `lTime` / `sMD5`）后：
- `vData` 直接放 `Wup.encode()` 原包，不再拼 `TAIL_BYTES`；
- 响应按 WUP `iRequestId` 匹配（不再 FIFO）；
- 初始化顺序对齐：`RegisterGroup(16)` → `wsLaunch(-1)` → `VerifyCookie(10)` →
`getConfig` → `UpdateUserExps(23)` → `RegisterGroup(16)` → `HeartBeat(20)`。
修改后 `tests/ -k huya` 全绿（394 passed），并用真号 CK 实测 WSS 返回 18 个商品。
scripts/huya_clone_ck.py
+622
View File
@@ -0,0 +1,622 @@
"""从小X分身/虎牙 App 的 WebView Cookie 库提取虎牙 CK。
前提：手机已 root（SukiSU/KernelSU/Magisk），且 `adb shell su -c id` 返回 uid=0。
脚本只读设备数据，不修改 App 数据；默认只输出脱敏摘要，避免 CK 直接落到终端。
典型用法：
.venv/bin/python scripts/huya_clone_ck.py
.venv/bin/python scripts/huya_clone_ck.py --include-real --out /tmp/huya_cks.txt
.venv/bin/python scripts/huya_clone_ck.py --verify-sid 2203
输出文件格式为 `账号----ck`，可直接贴入现有虎牙协作台账的「贴 / 换 CK」。
"""
from __future__ import annotations
import argparse
import json
import os
import shlex
import sqlite3
import subprocess
import sys
import tempfile
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from core.huya.cookie_utils import cookie_pairs, cookie_value, normalize_cookie_pairs
from core.huya.web_cookie_fields import fill_web_cookie_fields
HUYA_PACKAGE = "com.duowan.kiwi"
DKPLAT_PACKAGE = "com.bly.dkplat"
COOKIE_REL = "app_webview/Default/Cookies"
CRED0_REL = "files/huyaudb/cred0"
# 虎牙 App 本地登录态文件 cred0 的 AES-128-ECB 密钥（UDB 主密钥前 16 字节）
UDB_MASTER_KEY = b"HuyaUdb1928374650qwertyuiop"[:16]
# WSS 登录态字段：App 登录 + cred0 就能给全
LOGIN_REQUIRED = ("udb_cred", "udb_biztoken", "udb_uid", "yyuid", "sdid")
# 匿名设备态字段：由匿名设备态协议签发（--anon-device / 后端自动补）
DEVICE_REQUIRED = ("udb_deviceid", "udb_guiddata", "udb_anobiztoken")
WSS_REQUIRED = LOGIN_REQUIRED + DEVICE_REQUIRED
class CliError(RuntimeError):
"""可预期的命令行错误。"""
@dataclass(frozen=True)
class CloneSource:
label: str
kind: str # clone / real
package: str
data_dir: str
cookies_path: str = ""
@dataclass
class Extracted:
source: CloneSource
uid: int
passport: str
cookie: str
raw_cookie: str
has_cred: bool
missing_wss: tuple[str, ...]
notes: list[str] = field(default_factory=list)
verify: dict[str, object] | None = None
@property
def account(self) -> str:
return self.passport or f"uid_{self.uid}"
@property
def wss_ready(self) -> bool:
"""登录态字段齐全即可走 WSS（匿名设备字段可另补）。"""
present = {key for key, value in cookie_pairs(self.cookie) if value}
return all(key in present for key in LOGIN_REQUIRED)
class AdbDevice:
def __init__(self, adb: str, serial: str | None) -> None:
self.adb = adb
self.serial = serial
def _invoke(
self, mode: str, command: str, *, root: bool
) -> subprocess.CompletedProcess[bytes]:
args = [self.adb]
if self.serial:
args.extend(["-s", self.serial])
args.append(mode)
args.append(f"su -c {shlex.quote(command)}" if root else command)
try:
return subprocess.run(args, capture_output=True, check=False)
except FileNotFoundError as exc:
raise CliError(f"找不到 adb 可执行文件: {self.adb}") from exc
def shell_text(self, command: str, *, root: bool = False) -> str:
proc = self._invoke("shell", command, root=root)
if proc.returncode != 0:
stderr = proc.stderr.decode("utf-8", "replace").strip()
raise CliError(f"adb shell 失败（{command}）：{stderr or proc.returncode}")
return proc.stdout.decode("utf-8", "replace")
def exec_out(self, command: str, *, root: bool = False) -> bytes:
proc = self._invoke("exec-out", command, root=root)
if proc.returncode != 0 or not proc.stdout:
stderr = proc.stderr.decode("utf-8", "replace").strip()
raise CliError(f"adb exec-out 失败（{command}）：{stderr or '无输出'}")
return proc.stdout
def ensure_root(self) -> None:
proc = self._invoke("shell", "id", root=True)
output = (proc.stdout + proc.stderr).decode("utf-8", "replace")
if proc.returncode != 0 or "uid=0" not in output:
raise CliError(
"adb shell 没有 root。请先在 SukiSU/KernelSU 里给 Shell 授权，"
"确认 `adb shell su -c id` 返回 uid=0。"
)
def pick_serial(adb: str, requested: str | None) -> str | None:
if requested:
return requested
try:
proc = subprocess.run([adb, "devices"], capture_output=True, check=False)
except FileNotFoundError as exc:
raise CliError(f"找不到 adb 可执行文件: {adb}") from exc
serials = [
line.split()[0]
for line in proc.stdout.decode("utf-8", "replace").splitlines()
if "\tdevice" in line and line.split()
]
if not serials:
raise CliError("没有已连接的 adb 设备。")
if len(serials) > 1:
raise CliError(
"连接了多个 adb 设备，请用 --serial 指定目标：" + ", ".join(serials)
)
return serials[0]
def discover_sources(device: AdbDevice, *, include_real: bool) -> list[CloneSource]:
sources: list[CloneSource] = []
packages = device.shell_text("pm list packages -3")
for line in packages.splitlines():
package = line.strip().removeprefix("package:")
if not package.startswith("dkplugin."):
continue
sources.append(
CloneSource(
label=f"clone:{package.removeprefix('dkplugin.')}",
kind="clone",
package=package,
data_dir=f"/data/user/0/{package}/virtual/data/user/0/{HUYA_PACKAGE}",
)
)
# 小X分身自己的「主分身」数据在 com.bly.dkplat/chaos/data/user/<id>/ 下。
glob_cmd = (
f"ls -d /data/user/0/{DKPLAT_PACKAGE}/chaos/data/user/*/{HUYA_PACKAGE} "
"2>/dev/null || true"
)
for raw_path in device.shell_text(glob_cmd, root=True).splitlines():
path = raw_path.strip()
if not path:
continue
marker = f"/{DKPLAT_PACKAGE}/chaos/data/user/"
instance = path.split(marker, 1)[1].split("/", 1)[0] if marker in path else "?"
sources.append(
CloneSource(
label=f"clone:dkplat/{instance}",
kind="clone",
package=DKPLAT_PACKAGE,
data_dir=path,
)
)
if include_real:
sources.append(
CloneSource(
label="real:huya",
kind="real",
package=HUYA_PACKAGE,
data_dir=f"/data/user/0/{HUYA_PACKAGE}",
)
)
found: list[CloneSource] = []
for source in sources:
cookies_path = f"{source.data_dir}/{COOKIE_REL}"
probe = device.shell_text(
f"test -f {cookies_path} && echo yes || echo no", root=True
).strip()
if probe.endswith("yes"):
found.append(
CloneSource(
label=source.label,
kind=source.kind,
package=source.package,
data_dir=source.data_dir,
cookies_path=cookies_path,
)
)
return sorted(found, key=lambda item: item.label)
def read_cookies(blob: bytes) -> tuple[str, list[str]]:
"""把 WebView Cookies SQLite 转成 Cookie 字符串。"""
with tempfile.NamedTemporaryFile(suffix=".db") as handle:
handle.write(blob)
handle.flush()
connection = sqlite3.connect(f"file:{handle.name}?mode=ro", uri=True)
try:
integrity = connection.execute("PRAGMA integrity_check").fetchone()
if not integrity or integrity[0] != "ok":
raise CliError(f"Cookie 数据库完整性检查失败: {integrity}")
rows = list(
connection.execute(
"SELECT host_key, name, value, encrypted_value "
"FROM cookies ORDER BY rowid"
)
)
finally:
connection.close()
pairs: list[tuple[str, str]] = []
encrypted: list[str] = []
for host_key, name, value, encrypted_value in rows:
host = str(host_key or "")
key = str(name or "")
if not key or not (host == ".huya.com" or host.endswith("huya.com")):
continue
if value:
pairs.append((key, str(value)))
elif encrypted_value:
encrypted.append(key)
if not pairs:
raise CliError("Cookie 库里没有非空的 huya.com Cookie。")
return normalize_cookie_pairs(pairs), encrypted
def decrypt_cred0(blob: bytes) -> dict[str, object] | None:
"""解密 ``files/huyaudb/cred0``，取回 App 登录态里的 cred/uid/passport。
cred0 是 AES-128-ECB(零填充) 加密的登录历史 JSON；``apploginData.cred``
就是 ``udb_cred``（152 字符 base64url），WebView Cookie 里永远没有它。
解密失败只返回 None，不影响 Cookie 提取。
"""
if not blob or len(blob) % 16:
return None
try:
from Crypto.Cipher import AES
plain = AES.new(UDB_MASTER_KEY, AES.MODE_ECB).decrypt(blob).rstrip(b"\x00")
payload = json.loads(plain)
history = payload.get("loginHistory") or []
if not history:
return None
login = history[0].get("apploginData") or {}
cred = str(login.get("cred") or "")
if not cred:
return None
cookie = login.get("cookie") or {}
return {
"cred": cred,
"uid": int(login.get("uid") or cookie.get("uid") or 0),
"passport": str(login.get("passport") or cookie.get("passport") or ""),
"biztoken": str(cookie.get("biztoken") or ""),
"status": str(cookie.get("status") or ""),
"version": str(cookie.get("version") or ""),
}
except Exception: # noqa: BLE001 - 解密失败按「没有 cred0」处理
return None
def apply_credential(raw_cookie: str, login: Mapping[str, object]) -> str:
"""把 cred0 里的登录凭据写进 Cookie（覆盖 WebView 里的旧值）。"""
overrides = {
"udb_cred": login.get("cred"),
"udb_uid": login.get("uid"),
"yyuid": login.get("uid"),
"udb_passport": login.get("passport"),
"username": login.get("passport"),
"udb_biztoken": login.get("biztoken"),
"udb_status": login.get("status"),
"udb_version": login.get("version"),
}
pairs = [
(key, value) for key, value in cookie_pairs(raw_cookie) if key not in overrides
]
for key, value in overrides.items():
text = "" if value in (None, 0) else str(value)
if text:
pairs.append((key, text))
return "; ".join(f"{key}={value}" for key, value in pairs)
def build_web_cookie(raw_cookie: str, uid: int) -> str:
"""补齐网页设备态，并把渠道 appId 修正为网页的 5002。"""
filled = fill_web_cookie_fields(raw_cookie, guid="", uid=uid, qimei_feature=None)
pairs = dict(cookie_pairs(filled))
pairs["udb_appid"] = "5002"
return "; ".join(f"{key}={value}" for key, value in pairs.items())
def verify_cookie(cookie: str, uid: int, sid: int, timeout: float) -> dict[str, object]:
from core.huya.http_client import HuyaHttpClient
try:
response = HuyaHttpClient(logger=lambda _message: None).query_user_score(
uid, cookie, sid=sid, timeout=timeout
)
except Exception as exc: # noqa: BLE001 - 验证失败要写进报告而不是中断提取
return {"ok": False, "error": f"{type(exc).__name__}: {exc}"}
return {
"ok": True,
"status": getattr(response, "status", None),
"msg": getattr(response, "msg", None),
}
def extract_one(
device: AdbDevice,
source: CloneSource,
*,
fill_web: bool,
verify_sid: int | None,
timeout: float,
use_cred0: bool = True,
anon_device: bool = False,
) -> Extracted:
blob = device.exec_out(f"cat {source.cookies_path}", root=True)
raw_cookie, encrypted = read_cookies(blob)
uid = int(cookie_value(raw_cookie, "udb_uid") or 0)
if not uid:
uid = int(cookie_value(raw_cookie, "yyuid") or 0)
notes: list[str] = []
if encrypted:
notes.append(
f"有 {len(encrypted)} 个 Cookie 使用 encrypted_value（当前脚本不解密）"
)
# WebView Cookie 只有登录态字段，udb_cred 只存在于 App 本地登录态 cred0。
login: dict[str, object] | None = None
if use_cred0:
cred_path = f"{source.data_dir}/{CRED0_REL}"
probe = device.shell_text(
f"test -f {cred_path} && echo yes || echo no", root=True
).strip()
if probe.endswith("yes"):
login = decrypt_cred0(device.exec_out(f"cat {cred_path}", root=True))
if login:
notes.append("已从 cred0 解出 App 登录凭据（udb_cred）")
if not uid and login.get("uid"):
uid = int(login["uid"]) # type: ignore[arg-type]
else:
notes.append("cred0 存在但解密失败，CK 不含 udb_cred")
else:
notes.append("没有 cred0（该分身可能只有匿名态）")
if not uid:
raise CliError(f"{source.label} 的 Cookie 里没有 udb_uid/yyuid。")
base_cookie = apply_credential(raw_cookie, login) if login else raw_cookie
try:
cookie = build_web_cookie(base_cookie, uid) if fill_web else base_cookie
except Exception as exc: # noqa: BLE001 - 补齐失败时保留原始 Cookie
notes.append(
f"网页设备态补齐失败，退回原始 Cookie：{type(exc).__name__}: {exc}"
)
cookie = base_cookie
if anon_device:
try:
from core.huya.anon_device import fetch_anonymous_device_state
anon = fetch_anonymous_device_state()
missing = {
key: value
for key, value in anon.items()
if value and not cookie_value(cookie, key)
}
if missing:
cookie = "; ".join(
[cookie] + [f"{key}={value}" for key, value in missing.items()]
)
notes.append("已用匿名设备态补齐: " + ", ".join(sorted(missing)))
except Exception as exc: # noqa: BLE001 - 匿名设备态失败不阻断提取
notes.append(f"匿名设备态获取失败: {type(exc).__name__}: {exc}")
missing_login = tuple(
key for key in LOGIN_REQUIRED if not cookie_value(cookie, key)
)
missing_device = tuple(
key for key in DEVICE_REQUIRED if not cookie_value(cookie, key)
)
missing_wss = missing_login + missing_device
if missing_login:
notes.append(
"缺少 WSS 登录字段: "
+ ", ".join(missing_login)
+ "；活动/商城只能走 HTTP 通道"
)
elif missing_device:
notes.append(
"缺少匿名设备字段: "
+ ", ".join(missing_device)
+ "（加 --anon-device 可由协议补齐）"
)
verify = (
verify_cookie(cookie, uid, verify_sid, timeout)
if verify_sid is not None
else None
)
return Extracted(
source=source,
uid=uid,
passport=cookie_value(raw_cookie, "udb_passport")
or cookie_value(raw_cookie, "username"),
cookie=cookie,
raw_cookie=raw_cookie,
has_cred=bool(cookie_value(cookie, "udb_cred")),
missing_wss=missing_wss,
notes=notes,
verify=verify,
)
def public_record(item: Extracted, *, reveal: bool) -> dict[str, object]:
record: dict[str, object] = {
"label": item.source.label,
"kind": item.source.kind,
"package": item.source.package,
"uid": item.uid,
"passport": item.passport,
"account": item.account,
"cookie_fields": len(cookie_pairs(item.cookie)),
"cookie_bytes": len(item.cookie.encode("utf-8")),
"has_cred": item.has_cred,
"wss_ready": item.wss_ready,
"missing_device": [k for k in DEVICE_REQUIRED if k in item.missing_wss],
"missing_wss": list(item.missing_wss),
"notes": item.notes,
}
if reveal:
record["cookie"] = item.cookie
if item.verify is not None:
record["verify"] = item.verify
return record
def print_table(items: list[Extracted]) -> None:
if not items:
print("没有发现可提取的虎牙 Cookie。")
return
headers = ("实例", "UID", "账号", "字段", "WSS", "验证", "说明")
rows = []
for item in items:
verify = "-"
if item.verify is not None:
if item.verify.get("ok"):
verify = f"{item.verify.get('status')} {item.verify.get('msg') or ''}".strip()
else:
verify = str(item.verify.get("error") or "失败")
rows.append(
(
item.source.label,
str(item.uid),
item.account,
str(len(cookie_pairs(item.cookie))),
"OK" if item.wss_ready else "HTTP",
verify,
"；".join(item.notes) or "-",
)
)
widths = [
max(len(headers[index]), *(len(row[index]) for row in rows))
for index in range(len(headers))
]
print(
" ".join(headers[index].ljust(widths[index]) for index in range(len(headers)))
)
for row in rows:
print(
" ".join(row[index].ljust(widths[index]) for index in range(len(headers)))
)
def write_cookie_file(path: Path, items: list[Extracted]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
lines = [f"{item.account}----{item.cookie}" for item in items]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
os.chmod(path, 0o600)
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="从已 root 的安卓设备提取小X分身/虎牙 App 的 CK（只读）"
)
parser.add_argument("--adb", default="adb", help="adb 可执行文件（默认: adb）")
parser.add_argument("--serial", help="目标设备序列号（多设备时必填）")
parser.add_argument(
"--include-real",
action="store_true",
help="同时提取真身 com.duowan.kiwi（默认只提分身）",
)
parser.add_argument(
"--raw",
action="store_true",
help="不补齐网页设备字段（仍会注入 cred0 解出的 udb_cred）",
)
parser.add_argument(
"--no-cred0",
action="store_true",
help="不解密 files/huyaudb/cred0（默认会解密并补上 udb_cred）",
)
parser.add_argument(
"--anon-device",
action="store_true",
help="额外走匿名设备态协议补齐 udb_deviceid/guiddata/anobiztoken（需要外网）",
)
parser.add_argument(
"--out",
type=Path,
help="把完整 CK 写成 `账号----ck` 文件（权限 0600）",
)
parser.add_argument(
"--print",
action="store_true",
help="在终端打印完整 `账号----ck`（默认只打印脱敏摘要）",
)
parser.add_argument("--json", action="store_true", help="输出 JSON 摘要")
parser.add_argument(
"--verify-sid",
type=int,
help="可选：用活动积分接口验证 CK（例如 2203；活动已结束也会返回 603）",
)
parser.add_argument("--timeout", type=float, default=15.0, help="验证超时秒数")
return parser
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
serial = pick_serial(args.adb, args.serial)
device = AdbDevice(args.adb, serial)
device.ensure_root()
sources = discover_sources(device, include_real=args.include_real)
if not sources:
raise CliError(
"没有发现虎牙分身。请确认小X分身已安装虎牙、分身至少启动过一次。"
)
items: list[Extracted] = []
seen: dict[tuple[int, str], str] = {}
for source in sources:
try:
item = extract_one(
device,
source,
fill_web=not args.raw,
verify_sid=args.verify_sid,
timeout=args.timeout,
use_cred0=not args.no_cred0,
anon_device=args.anon_device,
)
except CliError as exc:
print(f"[跳过] {source.label}: {exc}", file=sys.stderr)
continue
key = (item.uid, item.passport)
if key in seen:
print(
f"[去重] {source.label} 与 {seen[key]} 是同一账号 {item.account}",
file=sys.stderr,
)
continue
seen[key] = source.label
items.append(item)
if args.out:
write_cookie_file(args.out, items)
if args.print:
for item in items:
print(f"{item.account}----{item.cookie}")
if args.json:
print(
json.dumps(
[public_record(item, reveal=args.print) for item in items],
ensure_ascii=False,
indent=2,
)
)
elif not args.print:
print_table(items)
if args.out:
print(
f"已写入 {len(items)} 个 CK: {args.out}（{datetime.now(UTC).isoformat()}）",
file=sys.stderr,
)
if not args.print and not args.out and not args.json:
print("提示：加 --out 落盘，或加 --print 在终端输出完整 CK。")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except CliError as exc:
print(f"错误：{exc}", file=sys.stderr)
raise SystemExit(2) from exc
tests/test_huya_collab_orders.py
+31 -2
View File
@@ -795,6 +795,7 @@ def points_center_blob(
product_id: str = "hy-1001",
points: int = 200,
stock: int = 39,
remaining: int | None = None,
category: str = "礼包",
) -> bytes:
"""构造 queryAllocation 的 ``tag5`` 最小商品列表（TAF LIST）。"""
@@ -817,6 +818,13 @@ def points_center_blob(
os.write_string(5, "")
os.write_int64(9, points * 10)
os.write_int64(13, stock)
if remaining is not None:
os.write_int64(14, remaining)
os.write_head(17, TafType.LIST)
os.write_int32(0, 1)
os.write_head(0, TafType.STRUCT_BEGIN)
os.write_int64(3, remaining)
os.write_struct_end()
# 积分 tag75.tag0.tag2
os.write_head(75, TafType.STRUCT_BEGIN)
os.write_head(0, TafType.STRUCT_BEGIN)
@@ -1295,8 +1303,9 @@ class TestCollabCkPreview(TestCollabBase):
"product_id": "hy-1001",
"name": "不朽伊人大礼包",
"price": 200,
"remain_text": "剩39",
"remain_text": "39",
"stock": 39,
"claimed": 39,
"score": 200,
"category_name": "礼包",
"icon": "",
@@ -1392,8 +1401,9 @@ class TestCollabGifts(TestCollabBase):
"product_id": "hy-1001",
"name": "不朽伊人大礼包",
"price": 200,
"remain_text": "剩39",
"remain_text": "39",
"stock": 39,
"claimed": 39,
"score": 200,
"category_name": "礼包",
"icon": "",
@@ -1440,6 +1450,25 @@ class TestCollabGifts(TestCollabBase):
assert gifts[0]["score"] == 888
assert gifts[0]["stock"] == 7
def test_points_center_parser_prefers_remaining_stock(self):
"""2026-10-25 抓包：tag13 是累计兑换，tag14 才是剩余库存。"""
from core.huya.shop_structs import parse_points_center_gifts
gifts = parse_points_center_gifts(
{
"tag2": [
{
"tag5": points_center_blob(
points=600, stock=43194, remaining=5054
)
}
]
}
)
assert gifts[0]["claimed"] == 43194
assert gifts[0]["stock"] == 5054
assert gifts[0]["remain_text"] == "5054"
def test_resolve_gifts_sid_prefers_workbench_config(self):
self.db.add(HuyaConfig(sid="2277"))
self.db.commit()
tests/test_huya_elite_protocol.py
+7 -16
View File
@@ -11,7 +11,7 @@ from core.huya.shop_structs import (
ShopCodeRsp,
)
from core.huya.taf_protocol import TafInputStream, TafOutputStream, TafType
from core.huya.wss_client import TAIL_BYTES, HuyaWssClient, WssCommand, WssMessage
from core.huya.wss_client import HuyaWssClient, WssCommand, WssMessage
from web.backend.services.huya_runner_core import HuyaBatchRunnerCore
@@ -108,9 +108,8 @@ def test_wss_requires_web_device_cookie_fields():
def test_activity_auth_frame_matches_latest_capture_layout():
"""对齐 9.1 活动 AUTH（828 连接）逐字段结构。
9.1: tag3 ZERO 首字段 -> tag0 uid -> tag1 UA -> tag2 cookie(STRING4)
-> tag3 STRING1 guid(32hex) -> tag4 INT8=1 -> tag5 "HUYA&ZH&2052"
-> tag6 "" -> TAIL_BYTES
2026-10-25 对齐 huya-go：tag0 uid -> tag1 UA -> tag2 cookie(STRING4)
-> tag3 STRING1 guid(32hex) -> tag4 INT32=1 -> tag5 "HUYA&ZH&2052"
"""
async def build_frame():
@@ -126,12 +125,9 @@ def test_activity_auth_frame_matches_latest_capture_layout():
message = WssMessage.decode(asyncio.run(build_frame()))
assert message.command == WssCommand.AUTH
assert message.sequence == 0x1D000109
assert message.body.endswith(TAIL_BYTES)
assert message.sequence == 0
stream = TafInputStream(message.body[: -len(TAIL_BYTES)])
# 首字段：tag3 ZERO（9.1 活动帧的占位字段）
assert stream.read_head() == (3, TafType.ZERO)
stream = TafInputStream(message.body)
assert stream.read_int64(0) == 1199664135026
assert stream.read_string(1) == "webh5&0.0.1&websocket&&diypc_52775"
assert stream.read_head() == (2, TafType.STRING4)
@@ -141,12 +137,8 @@ def test_activity_auth_frame_matches_latest_capture_layout():
# tag3 = STRING1 guid（9.1 中为 Cookie 里的 32hex guid）
guid = stream.read_string(3)
assert guid == "0a" * 16
assert stream.read_int8(4) == 1
assert stream.read_int32(4) == 1
assert stream.read_string(5) == "HUYA&ZH&2052"
assert stream.read_string(6) == ""
# body 必须与 9.1 828 AUTH 同样以 TAIL_BYTES 收尾
assert message.body[-len(TAIL_BYTES) :] == TAIL_BYTES
def test_activity_cookie_guid_is_replaced_by_launch_guid():
@@ -173,8 +165,7 @@ def test_activity_auth_accepts_explicit_launch_guid_over_cookie_guid():
return client.ws.send.call_args.args[0]
message = WssMessage.decode(asyncio.run(build_frame()))
stream = TafInputStream(message.body[: -len(TAIL_BYTES)])
assert stream.read_head() == (3, TafType.ZERO)
stream = TafInputStream(message.body)
stream.read_int64(0)
stream.read_string(1)
assert stream.read_head() == (2, TafType.STRING4)
tests/test_huya_points_query.py
+15 -7
View File
@@ -11,10 +11,11 @@
import asyncio
from types import SimpleNamespace
from core.huya.frame_decoder import _decode_wup_body, decode_struct_bytes
from core.huya.frame_decoder import _decode_wup_body, _extract_wup, decode_struct_bytes
from core.huya.http_client import HuyaHttpClient
from core.huya.shop_structs import ShopRawRsp
from core.huya.wss_client import HuyaWssClient, WssMessage
from core.huya.wss_client import HuyaWssClient, WssCommand, WssMessage
from core.huya.wup_protocol import WupResponse
from web.backend.models import HuyaAccount
from web.backend.services.huya_runner import HuyaBatchRunner
@@ -41,10 +42,17 @@ class _FakeWs:
async def send(self, data: bytes):
self.sent.append(data)
while self.client.rpc_queue:
future = self.client.rpc_queue.popleft()
if not future.done():
future.set_result(b"")
message = WssMessage.decode(data)
if message.command != WssCommand.RPC_REQUEST:
return
try:
wup = WupResponse()
wup.decode(_extract_wup(message.body))
except Exception: # noqa: BLE001 测试替身：无法解析的帧不兑现
return
future = self.client._pending.pop(wup.iRequestId, None)
if future is not None and not future.done():
future.set_result(b"")
def last_request(self) -> dict:
return _decode_wup_body(WssMessage.decode(self.sent[-1]).body)
@@ -111,7 +119,7 @@ def test_wss_redeem_need_goods_request_matches_real_call():
message = WssMessage.decode(client.ws.sent[-1])
# 积分中心业务帧统一 0x1d000107（只有下单/支付用 0x1d000108）
assert message.sequence == 0x1D000107
assert message.sequence == 0
decoded = client.ws.last_request()
assert (decoded["servant"], decoded["func"]) == (
tests/test_huya_points_shop_protocol.py
+27 -15
View File
@@ -29,12 +29,13 @@ COOKIE = f"udb_uid={UID}; yyuid={UID}; guid={GUID}"
PID = 1199650619883
# 抓包字面值（不要引用实现常量，否则断言会与实现一起漂移）
POINTS_UA_LITERAL = "webh5&0.0.1&websocket&&h5_/points.html"
# 旧版本在 vData 末尾拼的尾部字节；JCE 帧不需要，保留常量用于确认已移除
TAIL_LITERAL = bytes.fromhex("2c36004c5c6600")
BIZ_PARAMS_LITERAL = '{ "jifencenter": 1 }'
# 抓包实测：RegisterGroup(live:0) 与 UpdateUserExps(版本 map) 的 body 逐字节一致
HEARTBEAT_BODY = bytes.fromhex("09000106066c6976653a3016002c36004c5c6600")
UPDATE_EXPS_BODY = bytes.fromhex("0800010605353836363416063130373236332c36004c5c6600")
HEARTBEAT_BODY = bytes.fromhex("09000106066c6976653a301600")
UPDATE_EXPS_BODY = bytes.fromhex("080001060535383636341606313037323633")
CONFIG_VERSIONS = {"58664": "107263"}
@@ -48,10 +49,17 @@ class _FakeWs:
async def send(self, data: bytes):
self.sent.append(data)
while self.client.rpc_queue:
future = self.client.rpc_queue.popleft()
if not future.done():
future.set_result(self.response)
message = WssMessage.decode(data)
if message.command != WssCommand.RPC_REQUEST:
return
try:
wup = WupResponse()
wup.decode(_extract_wup(message.body))
except Exception: # noqa: BLE001 测试替身：无法解析的帧不兑现
return
future = self.client._pending.pop(wup.iRequestId, None)
if future is not None and not future.done():
future.set_result(self.response)
def last_body(self) -> bytes:
return WssMessage.decode(self.sent[-1]).body
@@ -106,8 +114,9 @@ def test_register_group_frame_matches_capture():
await client.send_heartbeat()
message = WssMessage.decode(client.ws.sent[-1])
assert message.command == WssCommand.HEARTBEAT_SEND
assert message.sequence == 0x1D00000D
assert message.sequence == 0
assert message.body == HEARTBEAT_BODY
assert not message.body.endswith(TAIL_LITERAL)
asyncio.run(run())
@@ -119,8 +128,8 @@ def test_shop_ws_launch_carries_real_identity_and_points_ua():
message = WssMessage.decode(client.ws.sent[-1])
assert message.command == WssCommand.RPC_REQUEST
# 抓包：wsLaunch 帧 seq=0x1d000100，iRequestId=-1
assert message.sequence == 0x1D000100
# JCE 帧 lRequestId 固定 0，真正的请求号是 WUP iRequestId=-1
assert message.sequence == 0
wup = WupResponse()
wup.decode(_extract_wup(message.body))
assert wup.iRequestId == -1
@@ -182,8 +191,7 @@ def test_create_order_v5_matches_points_capture():
)
message = WssMessage.decode(client.ws.sent[-1])
# 抓包：下单帧用 0x1d000108（其余业务帧 0x1d000107）
assert message.sequence == 0x1D000108
assert message.command == WssCommand.RPC_REQUEST
decoded = _decode_wup_body(message.body)
assert (decoded["servant"], decoded["func"]) == (
@@ -223,7 +231,7 @@ def test_pay_order_submit_v5_callback_points_page():
)
message = WssMessage.decode(client.ws.sent[-1])
assert message.sequence == 0x1D000108
assert message.command == WssCommand.RPC_REQUEST
decoded = _decode_wup_body(message.body)
assert (decoded["servant"], decoded["func"]) == (
@@ -338,11 +346,15 @@ def test_new_points_rpcs_match_capture():
asyncio.run(run())
def test_confirm_frame_ends_with_capture_tail():
def test_confirm_frame_carries_version_map_without_legacy_tail():
async def run():
client = _points_client()
await client.send_confirm({})
assert client.ws.last_body().endswith(TAIL_LITERAL)
await client.send_confirm(CONFIG_VERSIONS)
message = WssMessage.decode(client.ws.sent[-1])
assert message.command == WssCommand.CONFIRM_SEND
assert message.sequence == 0
assert message.body == UPDATE_EXPS_BODY
assert not message.body.endswith(TAIL_LITERAL)
asyncio.run(run())