From 03bc0818dd175647077ec6930987b79f85afc647 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 26 Oct 2026 11:23:29 +0800 Subject: [PATCH] =?UTF-8?q?feat(huya):=20=E5=88=86=E8=BA=AB=20CK=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=20cred0=20=E8=A7=A3=E5=AF=86=E8=A1=A5?= =?UTF-8?q?=E5=85=A8=EF=BC=8CWSS=20=E5=B8=A7=E6=94=B9=E4=B8=BA=20JCE=20Web?= =?UTF-8?q?SocketCommand?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- core/huya/shop_structs.py | 20 +- core/huya/wss_client.py | 265 +++++----- docs/虎牙/HAR提取CK与积分中心.md | 127 +++++ docs/虎牙/分身CK提取.md | 190 ++++++++ scripts/huya_clone_ck.py | 622 ++++++++++++++++++++++++ tests/test_huya_collab_orders.py | 33 +- tests/test_huya_elite_protocol.py | 23 +- tests/test_huya_points_query.py | 22 +- tests/test_huya_points_shop_protocol.py | 42 +- 9 files changed, 1188 insertions(+), 156 deletions(-) create mode 100644 docs/虎牙/HAR提取CK与积分中心.md create mode 100644 docs/虎牙/分身CK提取.md create mode 100644 scripts/huya_clone_ck.py diff --git a/core/huya/shop_structs.py b/core/huya/shop_structs.py index e731b59..dbb242a 100644 --- a/core/huya/shop_structs.py +++ b/core/huya/shop_structs.py @@ -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 diff --git a/core/huya/wss_client.py b/core/huya/wss_client.py index 1fd8836..4138ab8 100644 --- a/core/huya/wss_client.py +++ b/core/huya/wss_client.py @@ -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: ``::0:``。""" 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 diff --git a/docs/虎牙/HAR提取CK与积分中心.md b/docs/虎牙/HAR提取CK与积分中心.md new file mode 100644 index 0000000..80f29a6 --- /dev/null +++ b/docs/虎牙/HAR提取CK与积分中心.md @@ -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`。 diff --git a/docs/虎牙/分身CK提取.md b/docs/虎牙/分身CK提取.md new file mode 100644 index 0000000..0e056d1 --- /dev/null +++ b/docs/虎牙/分身CK提取.md @@ -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 --json + +# 完整 WSS CK：再加匿名设备态四字段（43 字段） +.venv/bin/python scripts/huya_clone_ck.py --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 个商品。 diff --git a/scripts/huya_clone_ck.py b/scripts/huya_clone_ck.py new file mode 100644 index 0000000..dd234c3 --- /dev/null +++ b/scripts/huya_clone_ck.py @@ -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// 下。 + 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 diff --git a/tests/test_huya_collab_orders.py b/tests/test_huya_collab_orders.py index 9a62167..f8e5218 100644 --- a/tests/test_huya_collab_orders.py +++ b/tests/test_huya_collab_orders.py @@ -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() diff --git a/tests/test_huya_elite_protocol.py b/tests/test_huya_elite_protocol.py index 71ef66f..9c0544e 100644 --- a/tests/test_huya_elite_protocol.py +++ b/tests/test_huya_elite_protocol.py @@ -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) diff --git a/tests/test_huya_points_query.py b/tests/test_huya_points_query.py index 300eadd..edd6906 100644 --- a/tests/test_huya_points_query.py +++ b/tests/test_huya_points_query.py @@ -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"]) == ( diff --git a/tests/test_huya_points_shop_protocol.py b/tests/test_huya_points_shop_protocol.py index 00e3eeb..66b2d04 100644 --- a/tests/test_huya_points_shop_protocol.py +++ b/tests/test_huya_points_shop_protocol.py @@ -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())