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

统一推送前Python代码格式

Browse Source
This commit is contained in:
yml2213
2026-09-02 18:11:41 +08:00
parent aaaa568916
commit 95ad900ef0
6 changed files with 265 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/browser_harvest.py
+129 -56
View File
@@ -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，跑完自动清理。
core/huya/elite_session.py
+12 -6
View File
@@ -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
core/huya/web_cookie_fields.py
+27 -20
View File
@@ -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))
core/huya/wss_client.py
+68 -18
View File
@@ -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
tests/test_huya_elite_protocol.py
+10 -9
View File
@@ -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():
tests/test_huya_web_cookie_fields.py
+19 -15
View File
@@ -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()