From 95ad900ef0936b3ee6cff0fca1c637ad9eee4284 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Wed, 2 Sep 2026 18:11:41 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E6=8E=A8=E9=80=81=E5=89=8DPy?= =?UTF-8?q?thon=E4=BB=A3=E7=A0=81=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/huya/browser_harvest.py | 185 +++++++++++++++++++-------- core/huya/elite_session.py | 18 ++- core/huya/web_cookie_fields.py | 47 ++++--- core/huya/wss_client.py | 86 ++++++++++--- tests/test_huya_elite_protocol.py | 19 +-- tests/test_huya_web_cookie_fields.py | 34 ++--- 6 files changed, 265 insertions(+), 124 deletions(-) diff --git a/core/huya/browser_harvest.py b/core/huya/browser_harvest.py index 51aed2d..e044412 100755 --- a/core/huya/browser_harvest.py +++ b/core/huya/browser_harvest.py @@ -14,6 +14,7 @@ 无第三方依赖：标准库 + websockets（直接走 CDP 协议）。 """ + from __future__ import annotations import argparse @@ -29,8 +30,7 @@ from pathlib import Path import websockets CHROME = ( - Path.home() - / "Library/Caches/ms-playwright/chromium-1234/chrome-mac-arm64/" + Path.home() / "Library/Caches/ms-playwright/chromium-1234/chrome-mac-arm64/" "Google Chrome for Testing.app/Contents/MacOS/Google Chrome for Testing" ) @@ -41,8 +41,13 @@ PAGES = [ # 与 core/huya/app_login.py::WEB_DEVICE_COOKIE_KEYS 一致 WEB_DEVICE_KEYS = ( - "guid", "udb_guiddata", "udb_deviceid", "game_did", - "_qimei_uuid42", "udb_anobiztoken", "__yamid_new", + "guid", + "udb_guiddata", + "udb_deviceid", + "game_did", + "_qimei_uuid42", + "udb_anobiztoken", + "__yamid_new", ) OUT_DIR = Path("evidence/browser_cookie_harvest") @@ -83,20 +88,29 @@ class Cdp: self._id += 1 fut = asyncio.get_running_loop().create_future() self._pending[self._id] = fut - await self.ws.send(json.dumps({"id": self._id, "method": method, "params": params or {}})) + await self.ws.send( + json.dumps({"id": self._id, "method": method, "params": params or {}}) + ) resp = await fut if "error" in resp: raise RuntimeError(f"{method}: {resp['error']}") return resp.get("result", {}) async def eval_js(self, expr): - r = await self.call("Runtime.evaluate", { - "expression": expr, "returnByValue": True, "awaitPromise": True, - }) + r = await self.call( + "Runtime.evaluate", + { + "expression": expr, + "returnByValue": True, + "awaitPromise": True, + }, + ) return r.get("result", {}).get("value") -def launch(profile: Path, port: int, headless: bool = True) -> tuple[subprocess.Popen, str]: +def launch( + profile: Path, port: int, headless: bool = True +) -> tuple[subprocess.Popen, str]: """启动用于 CDP 抓取的 Chrome。 默认使用无头模式，避免登录/抓取流程弹出窗口；传入 ``headless=False`` @@ -104,9 +118,13 @@ def launch(profile: Path, port: int, headless: bool = True) -> tuple[subprocess. """ args = [ str(CHROME), - "--no-first-run", "--no-default-browser-check", - "--disable-background-networking", "--disable-component-update", - "--disable-sync", "--metrics-recording-only", "--disable-extensions", + "--no-first-run", + "--no-default-browser-check", + "--disable-background-networking", + "--disable-component-update", + "--disable-sync", + "--metrics-recording-only", + "--disable-extensions", "--disable-features=OptimizationHints,MediaRouter", f"--user-data-dir={profile}", f"--remote-debugging-port={port}", @@ -133,7 +151,11 @@ def launch(profile: Path, port: int, headless: bool = True) -> tuple[subprocess. f"http://127.0.0.1:{port}/json/list", timeout=2 ) as resp: targets = json.load(resp) - pages = {t for t in targets if t.get("type") == "page" and t.get("webSocketDebuggerUrl")} + pages = { + t + for t in targets + if t.get("type") == "page" and t.get("webSocketDebuggerUrl") + } if pages: ws_url = pages[0]["webSocketDebuggerUrl"] break @@ -204,11 +226,13 @@ async def capture_round(profile: Path, tag: str, port: int, *, headless: bool = else: entry["seq"] = min(ack_order, req_seq) req_seq += 1 - entry.update({ - "url": url, - "host": urllib.parse.urlparse(url).netloc, - "type": params.get("type", ""), - }) + entry.update( + { + "url": url, + "host": urllib.parse.urlparse(url).netloc, + "type": params.get("type", ""), + } + ) # 部分请求的 Cookie 头在 requestWillBeSent 就直接可见 seen = set(entry.get("cookie_names", [])) for name in cookie_names_from_header(req.get("headers", {})): @@ -249,12 +273,16 @@ async def capture_round(profile: Path, tag: str, port: int, *, headless: bool = entry = requests.setdefault(rid, {"url": "?"}) url = entry.get("url", "?") headers = params.get("headers", {}) - set_cookie = headers.get("Set-Cookie") or headers.get("set-cookie") or [] + set_cookie = ( + headers.get("Set-Cookie") or headers.get("set-cookie") or [] + ) if isinstance(set_cookie, str): set_cookie = [set_cookie] for sc in set_cookie: name = sc.split("=", 1)[0].strip() - resp_extra.append({"cookie": name, "url": url, "meta": sc[:160]}) + resp_extra.append( + {"cookie": name, "url": url, "meta": sc[:160]} + ) cdp.listeners.append(on_resp_extra) @@ -268,10 +296,19 @@ async def capture_round(profile: Path, tag: str, port: int, *, headless: bool = } await navigate(cdp, url) doc_cookies[name]["after"] = await cdp.eval_js("document.cookie") - snapshots.append({"page": name, "url": url, "time": ts, - "cookies": await cdp.call("Network.getAllCookies")}) - print(f"[harvest] {tag}/{name} done: {url} doc_cookie_len=" - f"{len(doc_cookies[name]['after'] or '')}", flush=True) + snapshots.append( + { + "page": name, + "url": url, + "time": ts, + "cookies": await cdp.call("Network.getAllCookies"), + } + ) + print( + f"[harvest] {tag}/{name} done: {url} doc_cookie_len=" + f"{len(doc_cookies[name]['after'] or '')}", + flush=True, + ) # 按事件到达次序输出请求列表 req_list = sorted(requests.values(), key=lambda r: r.get("seq", 10**9)) @@ -330,14 +367,21 @@ async def probe_anon(profile: Path, port: int, *, headless: bool = True): "url": url, "method": req.get("method", ""), "postData": req.get("postData", ""), - "headers": {k: v for k, v in req.get("headers", {}).items() - if k.lower() in ("cookie", "content-type", "referer", "origin")}, + "headers": { + k: v + for k, v in req.get("headers", {}).items() + if k.lower() + in ("cookie", "content-type", "referer", "origin") + }, "body": None, "status": None, } rid_map[params.get("requestId", "")] = entry anon.append(entry) - print(f"[probe] 抓到匿名请求: {req.get('method')} {url[:120]}", flush=True) + print( + f"[probe] 抓到匿名请求: {req.get('method')} {url[:120]}", + flush=True, + ) cdp.listeners.append(on_anon_req) @@ -357,10 +401,14 @@ async def probe_anon(profile: Path, port: int, *, headless: bool = True): # 抓响应体 for rid, entry in rid_map.items(): try: - body = await cdp.call("Network.getResponseBody", {"requestId": rid}) + body = await cdp.call( + "Network.getResponseBody", {"requestId": rid} + ) entry["body"] = body.get("body", "") - print(f"[probe] 响应体 {len(entry['body'])}B status={entry['status']}", - flush=True) + print( + f"[probe] 响应体 {len(entry['body'])}B status={entry['status']}", + flush=True, + ) except Exception as exc: # noqa: BLE001 print(f"[probe] 响应体获取失败: {exc}", flush=True) @@ -389,16 +437,20 @@ async def probe_anon(profile: Path, port: int, *, headless: bool = True): replay = [] for entry in anon: host = urllib.parse.urlparse(entry["url"]).netloc - headers = {"User-Agent": ( - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/149.0.0.0 Safari/537.36" - )} + headers = { + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/149.0.0.0 Safari/537.36" + ) + } ck = cookie_header(host) if ck: headers["Cookie"] = ck if entry["postData"]: - headers.setdefault("Content-Type", "application/x-www-form-urlencoded") + headers.setdefault( + "Content-Type", "application/x-www-form-urlencoded" + ) try: resp = await asyncio.to_thread( pyrequests.request, @@ -409,18 +461,22 @@ async def probe_anon(profile: Path, port: int, *, headless: bool = True): timeout=15, allow_redirects=False, ) - replay.append({ - "url": entry["url"][:160], - "method": entry["method"], - "status": resp.status_code, - "set_cookie": resp.headers.get("Set-Cookie", ""), - "location": resp.headers.get("Location", ""), - "body_head": resp.text[:600], - "body_len": len(resp.content), - }) - print(f"[probe] 重放 {entry['method']} {entry['url'][:90]} " - f"-> {resp.status_code} setcookie={bool(resp.headers.get('Set-Cookie'))}", - flush=True) + replay.append( + { + "url": entry["url"][:160], + "method": entry["method"], + "status": resp.status_code, + "set_cookie": resp.headers.get("Set-Cookie", ""), + "location": resp.headers.get("Location", ""), + "body_head": resp.text[:600], + "body_len": len(resp.content), + } + ) + print( + f"[probe] 重放 {entry['method']} {entry['url'][:90]} " + f"-> {resp.status_code} setcookie={bool(resp.headers.get('Set-Cookie'))}", + flush=True, + ) except Exception as exc: # noqa: BLE001 replay.append({"url": entry["url"][:160], "error": str(exc)}) print(f"[probe] 重放失败: {exc}", flush=True) @@ -449,9 +505,17 @@ def build_report(): sys.exit(2) req1 = json.loads((r1 / "requests.json").read_text()) - req2 = json.loads((r2 / "requests.json").read_text()) if (r2 / "requests.json").exists() else [] + req2 = ( + json.loads((r2 / "requests.json").read_text()) + if (r2 / "requests.json").exists() + else [] + ) setc1 = json.loads((r1 / "set_cookies.json").read_text()) - snap2 = json.loads((r2 / "snapshots.json").read_text()) if (r2 / "snapshots.json").exists() else [] + snap2 = ( + json.loads((r2 / "snapshots.json").read_text()) + if (r2 / "snapshots.json").exists() + else [] + ) doc1 = json.loads((r1 / "doc_cookies.json").read_text()) # 每个 cookie 名在 round1 的首次出现位置（按请求顺序） @@ -459,7 +523,12 @@ def build_report(): for i, req in enumerate(req1): for name in req.get("cookie_names", []): if name not in first_seen: - first_seen[name] = {"idx": i, "host": req["host"], "url": req["url"], "type": req["type"]} + first_seen[name] = { + "idx": i, + "host": req["host"], + "url": req["url"], + "type": req["type"], + } # Set-Cookie 写入的 cookie set_from: dict[str, str] = {} @@ -483,7 +552,7 @@ def build_report(): def fmt_first(info: dict | None): if not info: return "未见来源" - return f'#{info["idx"]} {info["host"]} ({info["type"]})' + return f"#{info['idx']} {info['host']} ({info['type']})" lines = [ "# 浏览器 Cookie 字段来源分析（真实 Chrome 抓包，未登录）", @@ -515,7 +584,7 @@ def build_report(): for i, req in enumerate(req1[:60]): names = ", ".join(req.get("cookie_names", [])[:12]) names_all = req.get("cookie_names", []) - extra = f" +{len(names_all)-12}" if len(names_all) > 12 else "" + extra = f" +{len(names_all) - 12}" if len(names_all) > 12 else "" lines.append(f"| {i} | {req['host']} | {req['type']} | {names}{extra} |") lines += [ @@ -537,8 +606,10 @@ def build_report(): names = [p.split("=", 1)[0].strip() for p in after.split(";") if p.strip()] diff = [n for n in names if n not in seen] seen.update(names) - lines.append(f"- {page}: before={bool(v.get('before'))} after={len(names)} 字段" - f" 新增={', '.join(diff) if diff else '无'}") + lines.append( + f"- {page}: before={bool(v.get('before'))} after={len(names)} 字段" + f" 新增={', '.join(diff) if diff else '无'}" + ) lines += ["", "## round2 重启后首请求即携带的字段（浏览器持久态）", ""] if req2: pk = [] @@ -582,7 +653,9 @@ def main(): if args.mode == "round2" and not (profile / "Default").exists(): print("round2 需要先运行 round1（同一 profile）") sys.exit(2) - asyncio.run(capture_round(profile, args.mode, args.port, headless=not args.headed)) + asyncio.run( + capture_round(profile, args.mode, args.port, headless=not args.headed) + ) elif args.mode == "probe": # 匿名流程只在"无 udb_deviceid 持久态"的全新 profile 触发； # probe 模式内部使用临时全新 profile，跑完自动清理。 diff --git a/core/huya/elite_session.py b/core/huya/elite_session.py index 0eb740c..3926db4 100644 --- a/core/huya/elite_session.py +++ b/core/huya/elite_session.py @@ -58,7 +58,9 @@ class HuyaEliteWssSession: probe = HuyaWssClient(baseinfo=baseinfo, logger=self.logger) try: self.loop.run_until_complete( - probe.connect(host=connect_host, origin=origin, cookie=self.cookie) + probe.connect( + host=connect_host, origin=origin, cookie=self.cookie + ) ) self.loop.run_until_complete( probe.call_ws_launch_activity(self.uid, self.guid) @@ -75,9 +77,7 @@ class HuyaEliteWssSession: self.logger("[WSS] 预取探针清理失败") from .web_cookie_fields import fill_web_cookie_fields - self.cookie = fill_web_cookie_fields( - self.cookie, launch_guid, self.uid - ) + self.cookie = fill_web_cookie_fields(self.cookie, launch_guid, self.uid) self.guid = HuyaHttpClient._resolve_cookie_guid(self.cookie) baseinfo = urllib.parse.unquote( generate_http_baseinfo(self.uid, self.guid, "") @@ -85,7 +85,9 @@ class HuyaEliteWssSession: self.client = HuyaWssClient(baseinfo=baseinfo, logger=self.logger) self.loop.run_until_complete( - self.client.connect(host=connect_host, origin=origin, cookie=self.cookie) + self.client.connect( + host=connect_host, origin=origin, cookie=self.cookie + ) ) initialized = self.loop.run_until_complete( self.client.initialize_activity(self.uid, self.guid, self.cookie) @@ -117,7 +119,11 @@ class HuyaEliteWssSession: wsLaunch 取回会话 guid 并补齐字段（见 __init__），这里把缺字段的 输入替换为补齐后的 self.cookie。 """ - if not cookie or cookie == self._original_cookie or missing_web_cookie_fields(cookie): + if ( + not cookie + or cookie == self._original_cookie + or missing_web_cookie_fields(cookie) + ): return self.cookie return cookie diff --git a/core/huya/web_cookie_fields.py b/core/huya/web_cookie_fields.py index 6c375ca..0036aea 100644 --- a/core/huya/web_cookie_fields.py +++ b/core/huya/web_cookie_fields.py @@ -42,9 +42,7 @@ _FILL_MAP: tuple[tuple[str, str], ...] = ( _SDK_EPOCH_1582 = -12216643543000 # 浏览器实测 JS Date(1582,10,15) 本地时区 ms _BASE16 = "0123456789ABCDEF" -_GAME_DID_CHARS = ( - "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" -) +_GAME_DID_CHARS = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_" _DEFAULT_QIMEI_FEATURE = { "screen": {"resolution": "800x600", "availResolution": "800x600"}, "navigator": { @@ -100,18 +98,20 @@ def _yamid_new_generate32(rng=None) -> str: r = rng if rng is not None else random.Random() now_ms = int(time.time() * 1000) c = now_ms - _SDK_EPOCH_1582 - return "".join(( - _generate_bits(c, 0, 31), - _generate_bits(c, 32, 47), - _generate_bits(c, 48, 59) + "1", - _generate_bits(r.randrange(4095), 0, 7), - _generate_bits(r.randrange(4095), 0, 7), - _generate_bits(r.randrange(8191), 0, 7) - + _generate_bits(r.randrange(8191), 8, 15) - + _generate_bits(r.randrange(8191), 0, 7) - + _generate_bits(r.randrange(8191), 8, 15) - + _generate_bits(r.randrange(8191), 0, 15), - )) + return "".join( + ( + _generate_bits(c, 0, 31), + _generate_bits(c, 32, 47), + _generate_bits(c, 48, 59) + "1", + _generate_bits(r.randrange(4095), 0, 7), + _generate_bits(r.randrange(4095), 0, 7), + _generate_bits(r.randrange(8191), 0, 7) + + _generate_bits(r.randrange(8191), 8, 15) + + _generate_bits(r.randrange(8191), 0, 7) + + _generate_bits(r.randrange(8191), 8, 15) + + _generate_bits(r.randrange(8191), 0, 15), + ) + ) def _game_did_generate(length: int = 35) -> str: @@ -295,8 +295,7 @@ def _qimei_uuid42_generate( random_source = random.random if random_value is None else lambda: random_value random_segment = "".join( - f"{min(15, max(0, int(float(random_source()) * 16))):x}" - for _ in range(20) + f"{min(15, max(0, int(float(random_source()) * 16))):x}" for _ in range(20) ) template = prefix + "----" + random_segment + "----" replacement = f"{zlib.crc32(template.encode('ascii')) & 0xFFFFFFFF:08x}" @@ -390,9 +389,17 @@ def missing_web_cookie_fields(cookie: str) -> tuple[str, ...]: from .cookie_utils import cookie_value core = ( - "guid", "_qimei_uuid42", "__yamid_new", "game_did", - "_qimei_fingerprint", "_qimei_h38", "__yasmid", "__yamid_tt1", - "udb_passdata", "_yasids", "HMACCOUNT", + "guid", + "_qimei_uuid42", + "__yamid_new", + "game_did", + "_qimei_fingerprint", + "_qimei_h38", + "__yasmid", + "__yamid_tt1", + "udb_passdata", + "_yasids", + "HMACCOUNT", ) return tuple(key for key in core if not cookie_value(cookie, key)) diff --git a/core/huya/wss_client.py b/core/huya/wss_client.py index c665afd..511c385 100644 --- a/core/huya/wss_client.py +++ b/core/huya/wss_client.py @@ -57,6 +57,7 @@ class _ActivityConfigReq(TafStruct): def read_from(self, _ins): pass + # seq 按消息类型固定 (HAR 实证，跨两次抓包一致) SEQ_HEARTBEAT = 0x1D00000D SEQ_WSLAUNCH = 0x1D000078 @@ -426,7 +427,9 @@ class HuyaWssClient: await asyncio.sleep(0.1) return True - async def call_ws_launch_activity(self, uid: int, guid: str = "", timeout: float = 10.0): + async def call_ws_launch_activity( + self, uid: int, guid: str = "", timeout: float = 10.0 + ): ua = WSS_COOKIE_UA os = TafOutputStream() os.write_struct_begin(0) @@ -449,7 +452,9 @@ class HuyaWssClient: msg = WssMessage(WssCommand.RPC_REQUEST, SEQ_ACTIVITY_WSLAUNCH, body) future = asyncio.Future() self.rpc_queue.append(future) - self.logger(format_wss_log(body, WssCommand.RPC_REQUEST, SEQ_ACTIVITY_WSLAUNCH, "发")) + self.logger( + format_wss_log(body, WssCommand.RPC_REQUEST, SEQ_ACTIVITY_WSLAUNCH, "发") + ) await self.ws.send(msg.encode()) try: response = await asyncio.wait_for(future, timeout) @@ -457,7 +462,11 @@ class HuyaWssClient: if future in self.rpc_queue: self.rpc_queue.remove(future) return None - self.logger(format_wss_log(response, WssCommand.RPC_RESPONSE, SEQ_ACTIVITY_WSLAUNCH, "收")) + self.logger( + format_wss_log( + response, WssCommand.RPC_RESPONSE, SEQ_ACTIVITY_WSLAUNCH, "收" + ) + ) self._parse_launch_response(response) return response @@ -623,7 +632,11 @@ class HuyaWssClient: os.write_string(1, WSS_COOKIE_UA) os.write_string(2, HuyaWssClient._normalize_biz_cookie(cookie)) guid_match = re.search(r"(?:^|;\s*)guid=([^;]+)", cookie or "") - auth_guid = (guid or "").strip() or (guid_match.group(1).strip() if guid_match else "") or self._launch_guid + auth_guid = ( + (guid or "").strip() + or (guid_match.group(1).strip() if guid_match else "") + or self._launch_guid + ) os.write_string(3, auth_guid) os.write_int8(4, 1) os.write_string(5, "HUYA&ZH&2052") @@ -836,7 +849,9 @@ class HuyaWssClient: req = GetActTaskDetailReq() req.userId = self._build_activity_user(uid, cookie) req.actId = int(act_id or 0) - return await self.call_rpc("webActUI", "getActTaskDetail", req, GetActTaskDetailResp) + return await self.call_rpc( + "webActUI", "getActTaskDetail", req, GetActTaskDetailResp + ) async def get_act_user_task_detail(self, uid: int, cookie: str, act_id: int): from .activity_structs import GetActUserTaskDetailReq, GetActUserTaskDetailResp @@ -844,7 +859,9 @@ class HuyaWssClient: req = GetActUserTaskDetailReq() req.userId = self._build_activity_user(uid, cookie) req.actId = int(act_id or 0) - return await self.call_rpc("webActUI", "getActUserTaskDetail", req, GetActUserTaskDetailResp) + return await self.call_rpc( + "webActUI", "getActUserTaskDetail", req, GetActUserTaskDetailResp + ) async def get_user_score(self, uid: int, cookie: str, sid: int): from .activity_structs import GetUserScoreReq, GetUserScoreResp @@ -860,7 +877,9 @@ class HuyaWssClient: req = GetActPrizeListReq() req.userId = self._build_activity_user(uid, cookie) req.sid = int(sid or 0) - return await self.call_rpc("webActUI", "getActPrizeList", req, GetActPrizeListResp) + return await self.call_rpc( + "webActUI", "getActPrizeList", req, GetActPrizeListResp + ) async def get_act_prize_detail(self, uid: int, cookie: str, sid: int, pid: int): from .activity_structs import GetActPrizeDetailReq, GetActPrizeDetailResp @@ -869,7 +888,9 @@ class HuyaWssClient: req.userId = self._build_activity_user(uid, cookie) req.sid = int(sid or 0) req.pid = int(pid or 0) - return await self.call_rpc("webActUI", "getActPrizeDetail", req, GetActPrizeDetailResp) + return await self.call_rpc( + "webActUI", "getActPrizeDetail", req, GetActPrizeDetailResp + ) async def score_exchange_prize(self, uid: int, cookie: str, sid: int, pid: int): from .activity_structs import ScoreExchangePrizeReq, ScoreExchangePrizeResp @@ -878,7 +899,9 @@ class HuyaWssClient: req.userId = self._build_activity_user(uid, cookie) req.sid = int(sid or 0) req.pid = int(pid or 0) - return await self.call_rpc("webActUI", "scoreExchangePrize", req, ScoreExchangePrizeResp) + return await self.call_rpc( + "webActUI", "scoreExchangePrize", req, ScoreExchangePrizeResp + ) async def get_user_prize_records(self, uid: int, cookie: str, sid: int): from .activity_structs import GetUserPrizeRecordsReq, GetUserPrizeRecordsResp @@ -886,11 +909,19 @@ class HuyaWssClient: req = GetUserPrizeRecordsReq() req.userId = self._build_activity_user(uid, cookie) req.sid = int(sid or 0) - return await self.call_rpc("webActUI", "getUserPrizeRecords", req, GetUserPrizeRecordsResp) + return await self.call_rpc( + "webActUI", "getUserPrizeRecords", req, GetUserPrizeRecordsResp + ) async def check_user_bind_game_account( - self, uid: int, cookie: str, b_act_id: int, is_use_outer_act_id: int = 1, - gid: int = 0, outer_act_id: str = "", scene: str = "", + self, + uid: int, + cookie: str, + b_act_id: int, + is_use_outer_act_id: int = 1, + gid: int = 0, + outer_act_id: str = "", + scene: str = "", ): from .activity_structs import ( CheckUserBindGameAccountReq, @@ -926,8 +957,13 @@ class HuyaWssClient: ) async def get_live_link_param( - self, uid: int, cookie: str, b_act_id: int, gid: int = 0, - outer_act_id: int = 0, game_auth_scene: str = "", + self, + uid: int, + cookie: str, + b_act_id: int, + gid: int = 0, + outer_act_id: int = 0, + game_auth_scene: str = "", ): from .activity_structs import GetLiveLinkParamReq, GetLiveLinkParamResp @@ -937,9 +973,13 @@ class HuyaWssClient: req.outerActId = int(outer_act_id or 0) req.gameAuthScene = game_auth_scene or "" req.bActId = int(b_act_id or 0) - return await self.call_rpc("webActUI", "getLiveLinkParam", req, GetLiveLinkParamResp) + return await self.call_rpc( + "webActUI", "getLiveLinkParam", req, GetLiveLinkParamResp + ) - async def get_user_profile_batch(self, uid: int, cookie: str, target_uids: list[int]): + async def get_user_profile_batch( + self, uid: int, cookie: str, target_uids: list[int] + ): from .activity_structs import GetUserProfileBatchReq, GetUserProfileBatchResp req = GetUserProfileBatchReq() @@ -952,7 +992,12 @@ class HuyaWssClient: ) async def list_pay_channels( - self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int = 0, + self, + uid: int, + guid: str, + cookie: str, + spu_id: str, + sku_id: int = 0, supplier_id: int = 87401, ): from .shop_structs import ListPayChannelReq, ListPayChannelRsp @@ -978,7 +1023,12 @@ class HuyaWssClient: ) async def check_user_buy_auth( - self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int, + self, + uid: int, + guid: str, + cookie: str, + spu_id: str, + sku_id: int, ): from .shop_structs import CheckUserBuyAuthReq, ShopCodeRsp diff --git a/tests/test_huya_elite_protocol.py b/tests/test_huya_elite_protocol.py index ce4360d..71ef66f 100644 --- a/tests/test_huya_elite_protocol.py +++ b/tests/test_huya_elite_protocol.py @@ -95,14 +95,14 @@ def test_shop_preflight_response_contracts(): def test_wss_requires_web_device_cookie_fields(): - assert not HuyaBatchRunnerCore._has_web_wss_cookie( - "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1" - ) - assert HuyaBatchRunnerCore._has_web_wss_cookie( - "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1; sdid=s; guid=g; " - "udb_guiddata=gd; udb_deviceid=w_1234567890123456789; game_did=did; " - "_qimei_uuid42=q; udb_anobiztoken=ab; __yamid_new=y" - ) + assert not HuyaBatchRunnerCore._has_web_wss_cookie( + "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1" + ) + assert HuyaBatchRunnerCore._has_web_wss_cookie( + "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1; sdid=s; guid=g; " + "udb_guiddata=gd; udb_deviceid=w_1234567890123456789; game_did=did; " + "_qimei_uuid42=q; udb_anobiztoken=ab; __yamid_new=y" + ) def test_activity_auth_frame_matches_latest_capture_layout(): @@ -112,6 +112,7 @@ def test_activity_auth_frame_matches_latest_capture_layout(): -> tag3 STRING1 guid(32hex) -> tag4 INT8=1 -> tag5 "HUYA&ZH&2052" -> tag6 "" -> TAIL_BYTES """ + async def build_frame(): client = HuyaWssClient() client._activity_mode = True @@ -145,7 +146,7 @@ def test_activity_auth_frame_matches_latest_capture_layout(): assert stream.read_string(6) == "" # body 必须与 9.1 828 AUTH 同样以 TAIL_BYTES 收尾 - assert message.body[-len(TAIL_BYTES):] == TAIL_BYTES + assert message.body[-len(TAIL_BYTES) :] == TAIL_BYTES def test_activity_cookie_guid_is_replaced_by_launch_guid(): diff --git a/tests/test_huya_web_cookie_fields.py b/tests/test_huya_web_cookie_fields.py index 807a911..362db0d 100644 --- a/tests/test_huya_web_cookie_fields.py +++ b/tests/test_huya_web_cookie_fields.py @@ -56,9 +56,7 @@ def test_qimei_h38_aliases_uuid42_without_overwriting_existing_value(): assert fields["_qimei_h38"] == uuid42 fields = cookie_fields( - fill_web_cookie_fields( - f"_qimei_uuid42={uuid42}; _qimei_h38=historical", "" - ) + fill_web_cookie_fields(f"_qimei_uuid42={uuid42}; _qimei_h38=historical", "") ) assert fields["_qimei_h38"] == "historical" @@ -123,12 +121,14 @@ def test_sdk_uuid42_trace_vectors_have_confirmed_outer_layout(): def test_device_profile_builds_stable_webview_feature_approximation(): - feature = qimei_feature_from_device_profile({ - "vendor": "vivo", - "model": "V2370A", - "width": "1080", - "height": "2412", - }) + feature = qimei_feature_from_device_profile( + { + "vendor": "vivo", + "model": "V2370A", + "width": "1080", + "height": "2412", + } + ) assert feature["screen"] == { "resolution": "1080x2412", "availResolution": "1080x2412", @@ -136,11 +136,15 @@ def test_device_profile_builds_stable_webview_feature_approximation(): assert feature["navigator"]["pdfViewerEnabled"] is False assert "Adreno" in str(feature["webgl"]) first = qimei_fingerprint_generate(feature) - second = qimei_fingerprint_generate(qimei_feature_from_device_profile({ - "vendor": "vivo", - "model": "V2370A", - "width": "1080", - "height": "2412", - })) + second = qimei_fingerprint_generate( + qimei_feature_from_device_profile( + { + "vendor": "vivo", + "model": "V2370A", + "width": "1080", + "height": "2412", + } + ) + ) assert first == second assert first != qimei_fingerprint_generate()