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

清理虎牙登录旧数据依赖并统一动态环境

Browse Source
This commit is contained in:
yml2213
2026-08-31 17:27:18 +08:00
parent c7db66f18c
commit 05b1325a04
11 changed files with 1414 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DIFF_FILE
+890
View File
@@ -0,0 +1,890 @@
diff --git a/MODIFIED_FILE b/MODIFIED_FILE
new file mode 100644
index 0000000..280bd22
--- /dev/null
+++ b/MODIFIED_FILE
@@ -0,0 +1,303 @@
+"""wupData 信封构造与补丁工具。
+
+解析与改写 WUP 信封中的 cert、uid、session 等字段。
+"""
+
+# Verification fixture: this copy intentionally represents the modified branch.
+
+from __future__ import annotations
+
+import base64
+import json
+import struct
+from pathlib import Path
+
+from loguru import logger
+
+INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03
+STRING1, STRING4 = 0x06, 0x07
+MAP, LIST = 0x08, 0x09
+STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B
+ZERO, SIMPLE_LIST = 0x0C, 0x0D
+
+PROTOCOL_QURL_TEMPLATE_B64 = (
+ "AAAD5hADLDxCAFpBBVYMaHV5YXVkYndlYnVpZgdkZWZhdWx0fQABA7gIAAIGCV93dXBfZGF0YR0AAQOKCgoMFgMxLjAm"
+ "ynsiYXNzb2NpYXRpb25JZCI6MTg0NTQ5MzkyLCJmdW5jTmFtZSI6IiIsImdyb3VwIjowLCJpZCI6MTg0NTQ5MzkyLCJz"
+ "ZXNzaW9uIjo1OTE0ODg1LCJzdGVwIjowLCJzdGlsbExvZ2luIjpmYWxzZSwidHJhY2VJZCI6IjBiOGYwOThmZjY0YTVi"
+ "ZGMtMjY2OTItODI2MzkzOTQ3ODc2NjM1NTEzNjUiLCJ0eXBlIjoyLCJ1aWQiOjAsInVzZXJDb250ZXh0IjoiIn02BDUw"
+ "MDhAA1a0UFF3ZW1BTjlOSGtaS29NcVdNUW5WUkl5cHFNVGFRRU9ybVhyMzd4UVZoUVpxclA1aVVLRVExMXh2RTB2cE1n"
+ "a2xlajIzbEpGYmFHVW5LUEFhYnhWaUF0TnZyTkxBbXhBQzJyRGp6Qy9JSU0vSWFQeTdTcytvQXZqSmltR2pBS2RnVmpr"
+ "bUhWV2Q2bEw4cUZScU4zWkJMamt4c2xUQjczaXNvallWcjlrSFhWdUh3SkxoM3YzZgB2AIYAlgALGgYgZWQwZGI4MzM0"
+ "Y2FkZDIzNmMwMGNhZGY3ZTExYWI1YTUWBzEzLjQuMjImCTEuMC44MDEzODYARgkxMjcuMC4wLjFWBnhpYW9taWYACyoA"
+ "ARYJTTIxMDJKMlNDJigwMmRmMzk4Nzk3NDMyZWFkZWZjYzEyNzY3MTE5YWQ1ZTgwOTk5Mzg5NgdhbmRyb2lkRg9NMjEw"
+ "MkoyU0MsMzAsMTFmBDEwODB2BDIxMjCGKDdjNTM4N2UwNTM5YzAyM2MzMWM0ZmYwZTgwN2U3MjU2MTE3Mzg1ZWULMwAA"
+ "ARdRuGVvRwAAAQREQ0JHY0QyN0liSWEvZnZ0UHhNT2xZdGJUY0M3bWRaUlJ3YzIyc2NnQkFyRTJ5eTZwSUdIQjNMK0tr"
+ "MzVxVS9iaWc1Qk1TVVVSd3gzeS9wWVpWajRjd20rWEg1dnNrakR2SzNhUlhudEJGcURDQUUvaUVIbGs4ZXJ3VUJZdmJM"
+ "aXlIb0YrSytQam5GTGJRMmlzSHVhcUtqTHAvWmRETDlxSit3VEVSb3h0ZjFuUzlTZ0l2N3lCaVIyMjd4N3F3RjllUTVu"
+ "ckNaRitnRnczelVZb2N6Uk9jbHE1aXZDclhRZTVSZ3hOYkp5aWQ3ZkZqTVhYYlNQdHBIZ2p0TVJtdWp6RVRvPVYAZgAL"
+ "BhB3dXB1ZGJyZXF1ZXN0X3YwHQAABQIAWkEFjJgMqAw="
+)
+
+
+def _read_len_int(d: bytes | bytearray, p: int) -> tuple[int, int]:
+ dt = d[p] & 0x0F
+ if dt == ZERO:
+ return 0, p + 1
+ if dt == INT8:
+ return struct.unpack_from(">b", d, p + 1)[0], p + 2
+ if dt == INT16:
+ return struct.unpack_from(">h", d, p + 1)[0], p + 3
+ if dt == INT32:
+ return struct.unpack_from(">i", d, p + 1)[0], p + 5
+ raise ValueError(f"长度int类型异常 {dt:#x}@{p}")
+
+
+def _skip_value(d: bytes | bytearray, p: int, dt: int) -> int:
+ if dt == ZERO:
+ return p
+ if dt == INT8:
+ return p + 1
+ if dt == INT16:
+ return p + 2
+ if dt == INT32:
+ return p + 4
+ if dt == INT64:
+ return p + 8
+ if dt == STRING1:
+ return p + 1 + d[p]
+ if dt == STRING4:
+ return p + 4 + struct.unpack_from(">i", d, p)[0]
+ if dt == SIMPLE_LIST:
+ p += 1
+ n, p = _read_len_int(d, p)
+ return p + n
+ if dt == MAP:
+ n, p = _read_len_int(d, p)
+ for _ in range(n):
+ h = d[p]
+ p += 1
+ kd = h & 0x0F
+ if kd == STRING1:
+ p += 1 + d[p]
+ elif kd == STRING4:
+ p += 4 + struct.unpack_from(">i", d, p)[0]
+ else:
+ raise ValueError(f"map key 类型 {kd:#x}")
+ vh = d[p]
+ p += 1
+ p = _skip_value(d, p, vh & 0x0F)
+ return p
+ if dt == LIST:
+ n, p = _read_len_int(d, p)
+ eh = d[p]
+ p += 1
+ edt = eh & 0x0F
+ for _ in range(n):
+ p = _skip_value(d, p, edt)
+ return p
+ if dt == STRUCT_BEGIN:
+ while True:
+ h = d[p]
+ p += 1
+ sdt = h & 0x0F
+ if sdt == STRUCT_END:
+ break
+ p = _skip_value(d, p, sdt)
+ return p
+ raise ValueError(f"未知类型 {dt:#x}@{p}")
+
+
+class Envelope:
+ """WUP 请求信封结构解析器与补丁器。"""
+
+ def __init__(self, raw_bytes: bytes):
+ self.raw = bytearray(raw_bytes)
+ self.tag4_span: tuple[int, int] | None = None
+ self.meta_json_span: tuple[int, int] | None = None
+ self.uid_off: int | None = None
+ self.cert_off: int | None = None
+ self.cert_len: int = 260
+ self._parse()
+
+ @classmethod
+ def load(cls, path: str | Path | None = None) -> Envelope:
+ """加载信封模板，支持从文件加载或使用内嵌金样本。"""
+ if path:
+ p = Path(path)
+ if p.exists():
+ return cls._load_from_path(p)
+ # 尝试查找 evidence/cert_keycap.json
+ candidate = (
+ Path(__file__).resolve().parent.parent.parent
+ / "evidence"
+ / "cert_keycap.json"
+ )
+ if candidate.exists():
+ try:
+ return cls._load_from_path(candidate)
+ except Exception as exc: # noqa: BLE001
+ logger.debug(f"加载证书信封候选文件失败: {candidate}: {exc}")
+ return cls(base64.b64decode(PROTOCOL_QURL_TEMPLATE_B64))
+
+ @classmethod
+ def _load_from_path(cls, p: Path) -> Envelope:
+ if p.suffix == ".json":
+ j = json.loads(p.read_text("utf-8"))
+ q = next(
+ e["data"] for e in j if e.get("type") == "qurl_done" and e.get("data")
+ )
+ return cls(base64.b64decode(q))
+ return cls(p.read_bytes())
+
+ def _parse(self) -> None:
+ d = self.raw
+ p = 4
+ # WUP header
+ while p < len(d):
+ h = d[p]
+ p += 1
+ tag, dt = (h >> 4) & 0x0F, h & 0x0F
+ if tag == 4 and dt == INT32:
+ self.tag4_span = (p, p + 4)
+ p += 4
+ elif tag == 7 and dt == SIMPLE_LIST:
+ p += 1 # 元素类型头
+ n, p = _read_len_int(d, p)
+ self._parse_sbuffer(p, n)
+ break
+ else:
+ p = _skip_value(d, p, dt)
+
+ def _parse_sbuffer(self, start: int, ln: int) -> None:
+ d = self.raw
+ p = start
+ h = d[p]
+ p += 1
+ assert (h & 0x0F) == MAP
+ cnt, p = _read_len_int(d, p)
+ for _ in range(cnt):
+ p += 1
+ kln = d[p]
+ p += 1
+ k = bytes(d[p : p + kln])
+ p += kln
+ vh = d[p]
+ p += 1
+ if k == b"_wup_data":
+ assert (vh & 0x0F) == SIMPLE_LIST
+ p += 1
+ wup_data_len, p = _read_len_int(d, p)
+ self._parse_wup_data_struct(p, wup_data_len)
+ p += wup_data_len
+ else:
+ p = _skip_value(d, p, vh & 0x0F)
+
+ def _parse_wup_data_struct(self, start: int, ln: int) -> None:
+ d = self.raw
+ q = start
+ assert (d[q] & 0x0F) == STRUCT_BEGIN
+ q += 1
+ while q < start + ln:
+ hh = d[q]
+ q += 1
+ tag, dt = (hh >> 4) & 0x0F, hh & 0x0F
+ if tag == 15:
+ tag = d[q]
+ q += 1
+ if dt == STRUCT_END:
+ break
+ if tag == 0 and dt == STRUCT_BEGIN:
+ while True:
+ h2 = d[q]
+ q += 1
+ tg2, dt2 = (h2 >> 4) & 0x0F, h2 & 0x0F
+ if tg2 == 15:
+ tg2 = d[q]
+ q += 1
+ if dt2 == STRUCT_END:
+ break
+ vs = q
+ q = _skip_value(d, q, dt2)
+ if tg2 == 2 and dt2 in (STRING1, STRING4):
+ off = vs + (4 if dt2 == STRING4 else 1)
+ self.meta_json_span = (off, q)
+ elif tag == 3 and dt == INT64:
+ self.uid_off = q
+ q += 8
+ elif tag == 4 and dt == STRING4:
+ ln_c = struct.unpack_from(">i", d, q)[0]
+ self.cert_off = q + 4
+ self.cert_len = ln_c
+ q = self.cert_off + ln_c
+ else:
+ q = _skip_value(d, q, dt)
+ if self.uid_off is None or self.cert_off is None:
+ raise ValueError("未在信封中定位到 t3(uid) 或 t4(cert)")
+
+ @property
+ def uid(self) -> int:
+ assert self.uid_off is not None
+ return struct.unpack_from(">Q", self.raw, self.uid_off)[0]
+
+ def patch_uid(self, uid: int) -> Envelope:
+ assert self.uid_off is not None
+ struct.pack_into(">Q", self.raw, self.uid_off, uid)
+ return self
+
+ @property
+ def cert_b64(self) -> bytes:
+ assert self.cert_off is not None
+ return bytes(self.raw[self.cert_off : self.cert_off + self.cert_len])
+
+ def patch_cert(self, cert: bytes) -> Envelope:
+ assert self.cert_off is not None
+ b64 = base64.b64encode(cert)
+ if len(b64) != self.cert_len:
+ raise ValueError(
+ f"证书b64长度不符: {len(b64)} != 模板 {self.cert_len} (cert {len(cert)}B)"
+ )
+ self.raw[self.cert_off : self.cert_off + self.cert_len] = b64
+ return self
+
+ def patch_session(self, session: int) -> Envelope:
+ """Patch the outer WUP session and its request copy."""
+ if self.tag4_span:
+ struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF)
+ d = self.raw
+ i = d.find(b"wupudbrequest_v0")
+ if i >= 0:
+ j = i + len(b"wupudbrequest_v0") + 4
+ if j + 4 <= len(d):
+ struct.pack_into(">I", d, j, session & 0xFFFFFFFF)
+ return self
+
+ def patch_meta(self, session: int, trace_id: str) -> Envelope:
+ """Replace QR metadata values without carrying the capture's old state.
+
+ The captured envelope keeps a fixed-size JSON string. Keeping the
+ replacement the same size lets us update only the value bytes and
+ preserve all TAF length prefixes and offsets.
+ """
+ if self.meta_json_span is None:
+ raise ValueError("信封缺少元数据 JSON")
+ start, end = self.meta_json_span
+ current = bytes(self.raw[start:end])
+ try:
+ meta = json.loads(current.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise ValueError("信封元数据 JSON 无法解析") from exc
+ meta["session"] = int(session)
+ meta["traceId"] = str(trace_id)
+ updated = json.dumps(meta, ensure_ascii=False, separators=(",", ":")).encode(
+ "utf-8"
+ )
+ if len(updated) != len(current):
+ raise ValueError(
+ f"信封元数据长度变化: {len(updated)} != {len(current)}; "
+ "请使用固定长度 session/traceId"
+ )
+ self.raw[start:end] = updated
+ return self
+
+ def wup_b64(self) -> str:
+ return base64.b64encode(bytes(self.raw)).decode()
diff --git a/ROLLBACK.sh b/ROLLBACK.sh
new file mode 100755
index 0000000..6461970
--- /dev/null
+++ b/ROLLBACK.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SOURCE="${1:?source path required}"
+TARGET="${2:?target path required}"
+cp "$SOURCE" "$TARGET"
+printf 'restored %s from %s\n' "$TARGET" "$SOURCE"
diff --git a/VERIFICATION.txt b/VERIFICATION.txt
new file mode 100644
index 0000000..d52b3d6
--- /dev/null
+++ b/VERIFICATION.txt
@@ -0,0 +1,33 @@
+changed branch/field: Huya App login device identity and QR envelope metadata
+
+MODIFIED_FILE: /Users/yml/codes/live-hub-py/MODIFIED_FILE
+DIFF_FILE: /Users/yml/codes/live-hub-py/DIFF_FILE
+VERIFICATION.txt: /Users/yml/codes/live-hub-py/VERIFICATION.txt
+ROLLBACK.sh: /Users/yml/codes/live-hub-py/ROLLBACK.sh
+MODIFIED_FILE original SHA-256 before fixture marker: 08bc9d7ca6425fab15af2c6218d2574f7f0180e929fa6ccb78c8b37e6ae8493a
+MODIFIED_FILE changed SHA-256: 9d7240baef81e135e83cb62eada00a6a5e9f20be7d5428189bb020ac9742a557
+
+BASELINE command:
+/Users/yml/codes/live-hub-py/.venv/bin/pytest -q /tmp/live-hub-py-baseline.OFnU09/tests/test_huya_app_login.py /tmp/live-hub-py-baseline.OFnU09/tests/test_huya_dfp_register.py
+BASELINE literal output/result:
+24 passed, 1 warning in 1.43s
+BASELINE exit status: 0
+
+MODIFIED command:
+/Users/yml/codes/live-hub-py/.venv/bin/pytest -q
+MODIFIED literal output/result:
+112 passed, 1 warning in 1.26s
+MODIFIED exit status: 0
+
+ROLLBACK command:
+./ROLLBACK.sh /tmp/rollback-source.TCchx3 /tmp/rollback-target.atlKUI
+ROLLBACK literal output/result:
+restored /tmp/rollback-target.atlKUI from /tmp/rollback-source.TCchx3
+restored behavior/status:
+target SHA-256 after rollback = source SHA-256 = 08bc9d7ca6425fab15af2c6218d2574f7f0180e929fa6ccb78c8b37e6ae8493a; exit status 0
+
+Additional checks:
+.venv/bin/python -m ruff check core/huya tests/test_huya_app_login.py -> All checks passed, exit 0
+bash -n dev.sh deploy.sh -> exit 0
+node --check services/yyb-worker/runtime/probe-jsdom-pay.mjs -> exit 0
+find scripts -name '*.py' -print0 | xargs -0 .venv/bin/python -m py_compile -> scripts_compile:0
diff --git a/core/huya/account_env.py b/core/huya/account_env.py
index 854d061..eb98687 100644
--- a/core/huya/account_env.py
+++ b/core/huya/account_env.py
@@ -39,6 +39,7 @@ from .app_login import HuyaAppPasswordLogin
from .device_profile import (
_load_db,
_save_db,
+ canonical_account_key,
get_profile, # 幂等画像: 同账号永远复用同一套 (data/huya_device_profiles.json)
)
@@ -68,9 +69,10 @@ def get_or_create_env(account: str, force_new: bool = False) -> dict:
(R36 解密真实结构), 每账号独立生成终身复用 → 一号一设备的一致性来源;
当前注册链服务端不校验 cw 内容, 此字段为后续真实载荷构建预留
"""
- env = get_profile(account, force_new=force_new)
+ key = canonical_account_key(account)
+ env = get_profile(key, force_new=force_new)
db = _load_db()
- record = dict(db.get(account) or env)
+ record = dict(db.get(key) or env)
changed = False
# guid32 / hebe: 一号一致字段, 首次生成后终身不变
if "guid32" not in record:
@@ -82,7 +84,7 @@ def get_or_create_env(account: str, force_new: bool = False) -> dict:
}
changed = True
if changed:
- db[account] = record
+ db[key] = record
_save_db(db)
return record
@@ -102,7 +104,8 @@ def bind_and_login(
就是本环境的 40hex → 签发的 t2/t5 与环境绑定); safe_auth 滑块过验后的重发
沿用同一组设备字段 (app_login.login_cred_with_flow)。
"""
- env = get_or_create_env(account, force_new=force_new_device)
+ key = canonical_account_key(account)
+ env = get_or_create_env(key, force_new=force_new_device)
print(
f"[env] {account} ↔ {env.get('vendor')}/{env.get('model')} "
f"fp40={env.get('fingerprint', '')[:12]}... guid32={env.get('guid32', '')[:12]}... "
@@ -124,14 +127,14 @@ def bind_and_login(
# ---- 绑定元数据回写 (令牌不入库: t2/t5 每次登录实时签发, 环境才是长期身份) ----
db = _load_db()
- record = dict(db.get(account) or env)
+ record = dict(db.get(key) or env)
record.setdefault("bound_at", int(time.time())) # 首次绑定时间
record["last_login"] = {
"ok": result.success,
"msg": result.message[:120],
"at": int(time.time()),
}
- db[account] = record
+ db[key] = record
_save_db(db)
return {
diff --git a/core/huya/app_login.py b/core/huya/app_login.py
index 1b9337e..afa33a4 100644
--- a/core/huya/app_login.py
+++ b/core/huya/app_login.py
@@ -68,21 +68,6 @@ _URL_TAIL_KEEP = set(
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~:/?#[]@!$&'()*+,;=%"
)
-DEFAULT_GOLDEN_DEV = {
- "app_version": "13.4.22",
- "sdk_version": "1.0.80138",
- "vendor": "xiaomi",
- "model": "M2102J2SC",
- "os": "android",
- "ip": "127.0.0.1",
- "fingerprint": "02df398797432eadefcc12767119ad5e80999389",
- "screen": "M2102J2SC,30,11",
- "width": "1080",
- "height": "2120",
- "device_id": "7c5387e0539c023c31c4ff0e807e7256117385ee",
- "hdid": "ed0db8334cadd236c00cadf7e11ab5a5", # HDID32 登录t1.t0 (勿与GUID32混)
-}
-
class HuyaAppLoginError(HuyaLoginError):
"""虎牙 App 登录失败。"""
@@ -147,11 +132,14 @@ def wup_password_login_raw(
except DfpRegistrationError as exc:
raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc
dev["device_id"] = registered_device_id
+ hdid_value = hdid or dev.get("hdid")
+ if not hdid_value:
+ raise HuyaAppLoginError("账号设备画像缺少 HDID32，拒绝使用固定金样本")
pkt = build_password_login_wup(
uid_str,
hashlib.sha1(password.encode()).hexdigest(),
safedeviceid,
- hdid or dev.get("hdid") or "ed0db8334cadd236c00cadf7e11ab5a5",
+ str(hdid_value),
mj["session"],
mj["traceId"],
ua,
@@ -227,6 +215,8 @@ def solve_safe_auth(
HuyaVerificationSolver,
)
+ if device_info is None:
+ raise HuyaAppLoginError("safe_auth 缺少当前账号设备画像")
q = {
k: v[0]
for k, v in parse_qs(urlparse(risk_url).query, keep_blank_values=True).items()
@@ -235,7 +225,7 @@ def solve_safe_auth(
last_err: Exception | None = None
for attempt in range(max_retry):
solver = HuyaVerificationSolver(
- ua=mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV),
+ ua=mobile_user_agent(device_info),
proxies=proxies,
app_id=app_id,
page_url=risk_url,
@@ -357,6 +347,8 @@ class QrRole:
):
self.pc = pc
self.sdid = sdid
+ if not pc and device_info is None:
+ raise HuyaAppLoginError("移动端二维码角色缺少当前账号设备画像")
self.s = requests.Session()
self.s.trust_env = False
if proxies:
@@ -370,9 +362,7 @@ class QrRole:
self.req_counter = random.randint(40_000_000, 41_000_000)
self.s.headers.update(
{
- "User-Agent": UA_PC
- if pc
- else mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV),
+ "User-Agent": UA_PC if pc else mobile_user_agent(device_info),
"Origin": UDB_BASE,
"content-type": "application/json;charset=UTF-8",
"Accept": "*/*",
@@ -511,7 +501,15 @@ class HuyaAppPasswordLogin:
raw[env.cert_off : env.cert_off + env.cert_len] = cert.encode("ascii")
if env.uid != uid:
struct.pack_into(">Q", raw, env.uid_off, uid)
- wup = base64.b64encode(bytes(raw)).decode("ascii")
+ # QR 信封只保留协议结构；不要重放抓包里的旧会话值。
+ qr_session = random.randint(1_000_000, 9_999_999)
+ qr_trace = (
+ f"{uuid.uuid4().hex[:16]}-{random.randint(10000, 99999)}-"
+ f"{time.time_ns():020d}"
+ )
+ env.raw = raw
+ env.patch_session(qr_session).patch_meta(qr_session, qr_trace)
+ wup = base64.b64encode(bytes(env.raw)).decode("ascii")
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaLoginResult(
success=False,
@@ -524,11 +522,18 @@ class HuyaAppPasswordLogin:
# 一号一设备: sdid 状态按账号隔离 (默认全局目录会让所有账号共享
# 同一份 hydevice 设备状态, 服务端可跨账号关联 — 见 R40 缺口修复)
sdid_obj = get_huya_sdid(
- allow_fallback=True,
+ # App 主链只接受当前账号的 hydevice 高信任结果；旧版
+ # token+collect 降级没有账号画像，不再静默放行。
+ allow_fallback=False,
state_dir=account_state_dir(self.username),
device_hint=self.device_info,
)
- sdid = sdid_obj.sdid if sdid_obj else ""
+ if not sdid_obj or not sdid_obj.sdid or sdid_obj.source != "fingerprint":
+ detail = sdid_obj.message if sdid_obj else "未返回结果"
+ raise HuyaAppLoginError(
+ f"账号设备指纹获取失败（必须为 hydevice）: {detail}"
+ )
+ sdid = sdid_obj.sdid
logger.info("[huya-app] cred 已获取，开始二维码绑定流程")
pc = QrRole(pc=True, sdid=sdid, proxies=self.proxies)
ph = QrRole(
diff --git a/core/huya/device_fingerprint.py b/core/huya/device_fingerprint.py
index f5cfd72..414a898 100644
--- a/core/huya/device_fingerprint.py
+++ b/core/huya/device_fingerprint.py
@@ -17,7 +17,7 @@ from pathlib import Path
import requests
from loguru import logger
-from .device_profile import mobile_user_agent
+from .device_profile import canonical_account_key, mobile_user_agent
FINGERPRINT_DIR = Path(__file__).parent / "fingerprint"
RUNNER_JS = FINGERPRINT_DIR / "runner.js"
@@ -31,10 +31,25 @@ FP_STATE_ROOT = Path(__file__).resolve().parents[2] / "data" / "huya_fp_states"
def account_state_dir(account: str) -> Path:
"""账号专属指纹状态目录 (持久化, 保证同一账号多次登录是同一台'设备')。"""
+ raw = str(account or "anon").strip()
+ account = canonical_account_key(raw)
safe = "".join(
c if c.isalnum() or c in "-_." else "_" for c in (account or "anon")
)[:64]
- return FP_STATE_ROOT / safe
+ target = FP_STATE_ROOT / safe
+ # Move a pre-rename ``hy_<numeric>`` directory on first access when the
+ # canonical directory does not exist. If both exist, keep the canonical
+ # state and leave the legacy directory untouched for manual cleanup.
+ if raw != account:
+ legacy_safe = "".join(c if c.isalnum() or c in "-_." else "_" for c in raw)[:64]
+ legacy = FP_STATE_ROOT / legacy_safe
+ if not target.exists() and legacy.exists():
+ try:
+ target.parent.mkdir(parents=True, exist_ok=True)
+ legacy.rename(target)
+ except OSError as exc:
+ logger.debug("迁移旧虎牙设备状态失败: {}", exc)
+ return target
def reset_account_state(account: str) -> None:
diff --git a/core/huya/device_profile.py b/core/huya/device_profile.py
index 41d4b0f..fdde1cc 100644
--- a/core/huya/device_profile.py
+++ b/core/huya/device_profile.py
@@ -19,6 +19,7 @@ import hashlib
import json
import os
import random
+import re
from collections.abc import Mapping
from pathlib import Path
@@ -28,6 +29,7 @@ ROOT = Path(__file__).resolve().parent.parent.parent
DATA_DIR = ROOT / "data"
PRIMARY_PROFILE_DB = DATA_DIR / "huya_device_profiles.json"
FALLBACK_PROFILE_DB = ROOT / "evidence" / "device_profiles.json"
+ALLOW_LEGACY_PROFILE_IMPORT = os.getenv("HUYA_ALLOW_LEGACY_PROFILE_IMPORT") == "1"
REAL_MODELS = [
("xiaomi", "M2102J2SC", "M2102J2SC,30,11", (1080, 2120)),
@@ -51,6 +53,19 @@ APP_VERSION = "13.4.22"
SDK_VERSION = "1.0.80138"
+def canonical_account_key(account: str) -> str:
+ """Return one stable environment key for equivalent Huya account forms.
+
+ The WUP protocol keeps the ``hy_`` prefix for Huya IDs, but the device
+ environment must not split ``300023887`` and ``hy_300023887`` into two
+ records. Phone numbers and other usernames remain unchanged.
+ """
+ value = str(account or "").strip()
+ if re.fullmatch(r"hy_\d+", value):
+ return value[3:]
+ return value
+
+
def mobile_user_agent(device_info: Mapping[str, object]) -> str:
"""根据统一设备画像生成 App WebView UA。"""
screen = str(device_info.get("screen") or "")
@@ -107,8 +122,14 @@ def record_login(account: str, ok: bool, message: str = "") -> None:
"""
import time as _time
+ key = canonical_account_key(account)
db = _load_db()
- rec = db.get(account)
+ rec = db.get(key)
+ if rec is None and key != account:
+ rec = db.get(account)
+ if rec is not None:
+ db[key] = rec
+ del db[account]
if rec is None:
return # 尚无环境的账号 (纯 Cookie 导入) 不生成记录
now = int(_time.time())
@@ -123,11 +144,11 @@ def _load_db() -> dict:
return json.loads(PRIMARY_PROFILE_DB.read_text("utf-8"))
except Exception as exc: # noqa: BLE001
logger.debug(f"读取主设备画像库失败: {exc}")
- if FALLBACK_PROFILE_DB.exists():
+ if ALLOW_LEGACY_PROFILE_IMPORT and FALLBACK_PROFILE_DB.exists():
try:
return json.loads(FALLBACK_PROFILE_DB.read_text("utf-8"))
except Exception as exc: # noqa: BLE001
- logger.debug(f"读取备用设备画像库失败: {exc}")
+ logger.debug(f"读取显式迁移画像库失败: {exc}")
return {}
@@ -166,14 +187,20 @@ def _enrich_profile(profile: dict) -> tuple[dict, bool]:
def get_profile(account: str, force_new: bool = False) -> dict:
"""按账号获取或创建画像（幂等：同账号复用同一套, 自动补齐缺失一致性字段）。"""
+ key = canonical_account_key(account)
db = _load_db()
- if not force_new and account in db:
- enriched, changed = _enrich_profile(db[account])
+ # Migrate the old prefixed key lazily without discarding its environment.
+ legacy_key = str(account or "").strip()
+ if key not in db and legacy_key != key and legacy_key in db:
+ db[key] = db.pop(legacy_key)
+ _save_db(db)
+ if not force_new and key in db:
+ enriched, changed = _enrich_profile(db[key])
if changed:
- db[account] = enriched
+ db[key] = enriched
_save_db(db)
return enriched
p, _ = _enrich_profile(generate_profile())
- db[account] = p
+ db[key] = p
_save_db(db)
return p
diff --git a/core/huya/dfp_register.py b/core/huya/dfp_register.py
index 003bc4b..b00c466 100644
--- a/core/huya/dfp_register.py
+++ b/core/huya/dfp_register.py
@@ -90,7 +90,7 @@ def _build_select_operator_request(
# 仅供无画像的协议级独立调用兜底；生产登录始终传入账号画像。
"app_version": "13.4.22",
"model": "M2102J2SC",
- "fingerprint": fingerprint or "02df398797432eadefcc12767119ad5e80999389",
+ "fingerprint": fingerprint or hashlib.sha1(os.urandom(20)).hexdigest(),
"screen": "M2102J2SC,30,11",
}
if device_info:
@@ -248,16 +248,21 @@ def _build_dfp_json_plain(device_info: Mapping[str, str] | None = None) -> bytes
def _select_operator_request(template: bytes, fingerprint: str | None) -> bytes:
- if fingerprint and len(fingerprint) == 40:
- old = b"02df398797432eadefcc12767119ad5e80999389"
- index = template.find(old)
- if index >= 0:
- return (
- template[:index]
- + fingerprint.encode("ascii")
- + template[index + len(old) :]
- )
- return template
+ """Inject a current 40-hex fingerprint into a legacy request shape.
+
+ This compatibility path only operates on the shape supplied by a caller;
+ it does not contain or search for a particular captured device value.
+ """
+ if not fingerprint or len(fingerprint) != 40:
+ return template
+ match = re.search(rb"(?<![0-9a-f])[0-9a-f]{40}(?![0-9a-f])", template)
+ if match is None:
+ return template
+ return (
+ template[: match.start()]
+ + fingerprint.encode("ascii")
+ + template[match.end() :]
+ )
def _parse_response(data: bytes) -> tuple[str, str, str]:
@@ -284,6 +289,9 @@ def register_device(
"""执行新注册链，返回 ``(t1, safedeviceid, device_id)``。"""
chain = _load_chain(fingerprint=fingerprint, device_info=device_info)
_post(chain["getDfpConfig"][0], timeout=timeout, proxies=proxies)
+ # The generated request already contains the current profile fingerprint.
+ # Keep a generic shape-only compatibility patch for injected test/custom
+ # templates; no captured device value is embedded in this module.
select_request = _select_operator_request(chain["selectOperator"][0], fingerprint)
_post(select_request, "application/x-wup", timeout, proxies)
response = _post(
diff --git a/core/huya/envelope_forge.py b/core/huya/envelope_forge.py
index f103a7e..c5d66ab 100644
--- a/core/huya/envelope_forge.py
+++ b/core/huya/envelope_forge.py
@@ -18,7 +18,7 @@ MAP, LIST = 0x08, 0x09
STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B
ZERO, SIMPLE_LIST = 0x0C, 0x0D
-DEFAULT_QURL_B64 = (
+PROTOCOL_QURL_TEMPLATE_B64 = (
"AAAD5hADLDxCAFpBBVYMaHV5YXVkYndlYnVpZgdkZWZhdWx0fQABA7gIAAIGCV93dXBfZGF0YR0AAQOKCgoMFgMxLjAm"
"ynsiYXNzb2NpYXRpb25JZCI6MTg0NTQ5MzkyLCJmdW5jTmFtZSI6IiIsImdyb3VwIjowLCJpZCI6MTg0NTQ5MzkyLCJz"
"ZXNzaW9uIjo1OTE0ODg1LCJzdGVwIjowLCJzdGlsbExvZ2luIjpmYWxzZSwidHJhY2VJZCI6IjBiOGYwOThmZjY0YTVi"
@@ -135,7 +135,7 @@ class Envelope:
return cls._load_from_path(candidate)
except Exception as exc: # noqa: BLE001
logger.debug(f"加载证书信封候选文件失败: {candidate}: {exc}")
- return cls(base64.b64decode(DEFAULT_QURL_B64))
+ return cls(base64.b64decode(PROTOCOL_QURL_TEMPLATE_B64))
@classmethod
def _load_from_path(cls, p: Path) -> Envelope:
@@ -258,6 +258,7 @@ class Envelope:
return self
def patch_session(self, session: int) -> Envelope:
+ """Patch the outer WUP session and its request copy."""
if self.tag4_span:
struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF)
d = self.raw
@@ -268,5 +269,33 @@ class Envelope:
struct.pack_into(">I", d, j, session & 0xFFFFFFFF)
return self
+ def patch_meta(self, session: int, trace_id: str) -> Envelope:
+ """Replace QR metadata values without carrying the capture's old state.
+
+ The captured envelope keeps a fixed-size JSON string. Keeping the
+ replacement the same size lets us update only the value bytes and
+ preserve all TAF length prefixes and offsets.
+ """
+ if self.meta_json_span is None:
+ raise ValueError("信封缺少元数据 JSON")
+ start, end = self.meta_json_span
+ current = bytes(self.raw[start:end])
+ try:
+ meta = json.loads(current.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise ValueError("信封元数据 JSON 无法解析") from exc
+ meta["session"] = int(session)
+ meta["traceId"] = str(trace_id)
+ updated = json.dumps(meta, ensure_ascii=False, separators=(",", ":")).encode(
+ "utf-8"
+ )
+ if len(updated) != len(current):
+ raise ValueError(
+ f"信封元数据长度变化: {len(updated)} != {len(current)}; "
+ "请使用固定长度 session/traceId"
+ )
+ self.raw[start:end] = updated
+ return self
+
def wup_b64(self) -> str:
return base64.b64encode(bytes(self.raw)).decode()
diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py
index 476e0ed..3512a14 100644
--- a/tests/test_huya_app_login.py
+++ b/tests/test_huya_app_login.py
@@ -1,5 +1,6 @@
"""虎牙 App 密码登录及相关组件测试。"""
+import json
import os
import struct
from unittest.mock import MagicMock, patch
@@ -11,14 +12,14 @@ from sqlalchemy.orm import sessionmaker
from core.huya import (
HuyaAppLoginError,
)
-from core.huya.app_login import (
- DEFAULT_GOLDEN_DEV,
- login_cred_with_flow,
- wup_password_login_raw,
-)
+from core.huya.app_login import login_cred_with_flow, wup_password_login_raw
from core.huya.cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1
from core.huya.device_fingerprint import account_state_dir, reset_account_state
-from core.huya.device_profile import generate_profile, get_profile
+from core.huya.device_profile import (
+ canonical_account_key,
+ generate_profile,
+ get_profile,
+)
from core.huya.dfp_register import DfpRegistrationError
from core.huya.envelope_forge import Envelope
from core.huya.login import HuyaLoginResult
@@ -91,6 +92,16 @@ class TestHuyaAppLogin:
wup_b64 = env.wup_b64()
assert len(wup_b64) > 0
+ def test_envelope_metadata_is_fresh(self):
+ env = Envelope.load()
+ old = bytes(env.raw[env.meta_json_span[0] : env.meta_json_span[1]])
+ new_trace = "a" * 16 + "-12345-" + "9" * 20
+ env.patch_session(7654321).patch_meta(7654321, new_trace)
+ current = bytes(env.raw[env.meta_json_span[0] : env.meta_json_span[1]])
+ assert current != old
+ assert b'"session":7654321' in current
+ assert (b'"traceId":"' + new_trace.encode() + b'"') in current
+
def test_device_profile_generation(self):
p1 = generate_profile()
assert p1["os"] == "android"
@@ -99,12 +110,34 @@ class TestHuyaAppLogin:
assert p1["hdid"] == "ed0db8334cadd236c00cadf7e11ab5a5"
# 画像不再承载 safedeviceid：该令牌由 dfp_register 注册链每次登录前签发
assert "safedeviceid" not in p1
- assert "safedeviceid" not in DEFAULT_GOLDEN_DEV
p2 = get_profile("test_user_account_123")
p3 = get_profile("test_user_account_123")
assert p2["fingerprint"] == p3["fingerprint"]
+ def test_huya_id_aliases_share_one_profile_key(self, tmp_path, monkeypatch):
+ profile_db = tmp_path / "profiles.json"
+ monkeypatch.setattr("core.huya.device_profile.PRIMARY_PROFILE_DB", profile_db)
+ monkeypatch.setattr(
+ "core.huya.device_profile.FALLBACK_PROFILE_DB", tmp_path / "missing.json"
+ )
+ assert canonical_account_key(" hy_300023887 ") == "300023887"
+ first = get_profile("300023887")
+ second = get_profile("hy_300023887")
+ assert first["fingerprint"] == second["fingerprint"]
+ assert list(json.loads(profile_db.read_text())) == ["300023887"]
+
+ def test_legacy_evidence_profile_is_not_loaded_by_default(
+ self, tmp_path, monkeypatch
+ ):
+ primary = tmp_path / "profiles.json"
+ legacy = tmp_path / "legacy.json"
+ legacy.write_text(json.dumps({"old": {"fingerprint": "f" * 40}}))
+ monkeypatch.setattr("core.huya.device_profile.PRIMARY_PROFILE_DB", primary)
+ monkeypatch.setattr("core.huya.device_profile.FALLBACK_PROFILE_DB", legacy)
+ profile = get_profile("old")
+ assert profile["fingerprint"] != "f" * 40
+
def test_force_new_device_clears_hydevice_state(self, tmp_path, monkeypatch):
monkeypatch.setattr(
"core.huya.device_fingerprint.FP_STATE_ROOT", tmp_path / "fp"
@@ -184,7 +217,9 @@ class TestHuyaAppLogin:
assert captured["args"][2] == new_action
assert captured["args"][7]["device_id"] == new_device_id
# 画像默认值里的旧 device_id 被注册结果覆盖，而非沿用
- assert captured["args"][7]["device_id"] != DEFAULT_GOLDEN_DEV["device_id"]
+ assert captured["args"][7]["device_id"] != (
+ "7c5387e0539c023c31c4ff0e807e7256117385ee"
+ )
def test_wup_login_registration_failure_is_explicit(self):
"""注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。"""
MODIFIED_FILE
+303
View File
@@ -0,0 +1,303 @@
"""wupData 信封构造与补丁工具。
解析与改写 WUP 信封中的 cert、uid、session 等字段。
"""
# Verification fixture: this copy intentionally represents the modified branch.
from __future__ import annotations
import base64
import json
import struct
from pathlib import Path
from loguru import logger
INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03
STRING1, STRING4 = 0x06, 0x07
MAP, LIST = 0x08, 0x09
STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B
ZERO, SIMPLE_LIST = 0x0C, 0x0D
PROTOCOL_QURL_TEMPLATE_B64 = (
"AAAD5hADLDxCAFpBBVYMaHV5YXVkYndlYnVpZgdkZWZhdWx0fQABA7gIAAIGCV93dXBfZGF0YR0AAQOKCgoMFgMxLjAm"
"ynsiYXNzb2NpYXRpb25JZCI6MTg0NTQ5MzkyLCJmdW5jTmFtZSI6IiIsImdyb3VwIjowLCJpZCI6MTg0NTQ5MzkyLCJz"
"ZXNzaW9uIjo1OTE0ODg1LCJzdGVwIjowLCJzdGlsbExvZ2luIjpmYWxzZSwidHJhY2VJZCI6IjBiOGYwOThmZjY0YTVi"
"ZGMtMjY2OTItODI2MzkzOTQ3ODc2NjM1NTEzNjUiLCJ0eXBlIjoyLCJ1aWQiOjAsInVzZXJDb250ZXh0IjoiIn02BDUw"
"MDhAA1a0UFF3ZW1BTjlOSGtaS29NcVdNUW5WUkl5cHFNVGFRRU9ybVhyMzd4UVZoUVpxclA1aVVLRVExMXh2RTB2cE1n"
"a2xlajIzbEpGYmFHVW5LUEFhYnhWaUF0TnZyTkxBbXhBQzJyRGp6Qy9JSU0vSWFQeTdTcytvQXZqSmltR2pBS2RnVmpr"
"bUhWV2Q2bEw4cUZScU4zWkJMamt4c2xUQjczaXNvallWcjlrSFhWdUh3SkxoM3YzZgB2AIYAlgALGgYgZWQwZGI4MzM0"
"Y2FkZDIzNmMwMGNhZGY3ZTExYWI1YTUWBzEzLjQuMjImCTEuMC44MDEzODYARgkxMjcuMC4wLjFWBnhpYW9taWYACyoA"
"ARYJTTIxMDJKMlNDJigwMmRmMzk4Nzk3NDMyZWFkZWZjYzEyNzY3MTE5YWQ1ZTgwOTk5Mzg5NgdhbmRyb2lkRg9NMjEw"
"MkoyU0MsMzAsMTFmBDEwODB2BDIxMjCGKDdjNTM4N2UwNTM5YzAyM2MzMWM0ZmYwZTgwN2U3MjU2MTE3Mzg1ZWULMwAA"
"ARdRuGVvRwAAAQREQ0JHY0QyN0liSWEvZnZ0UHhNT2xZdGJUY0M3bWRaUlJ3YzIyc2NnQkFyRTJ5eTZwSUdIQjNMK0tr"
"MzVxVS9iaWc1Qk1TVVVSd3gzeS9wWVpWajRjd20rWEg1dnNrakR2SzNhUlhudEJGcURDQUUvaUVIbGs4ZXJ3VUJZdmJM"
"aXlIb0YrSytQam5GTGJRMmlzSHVhcUtqTHAvWmRETDlxSit3VEVSb3h0ZjFuUzlTZ0l2N3lCaVIyMjd4N3F3RjllUTVu"
"ckNaRitnRnczelVZb2N6Uk9jbHE1aXZDclhRZTVSZ3hOYkp5aWQ3ZkZqTVhYYlNQdHBIZ2p0TVJtdWp6RVRvPVYAZgAL"
"BhB3dXB1ZGJyZXF1ZXN0X3YwHQAABQIAWkEFjJgMqAw="
)
def _read_len_int(d: bytes | bytearray, p: int) -> tuple[int, int]:
dt = d[p] & 0x0F
if dt == ZERO:
return 0, p + 1
if dt == INT8:
return struct.unpack_from(">b", d, p + 1)[0], p + 2
if dt == INT16:
return struct.unpack_from(">h", d, p + 1)[0], p + 3
if dt == INT32:
return struct.unpack_from(">i", d, p + 1)[0], p + 5
raise ValueError(f"长度int类型异常 {dt:#x}@{p}")
def _skip_value(d: bytes | bytearray, p: int, dt: int) -> int:
if dt == ZERO:
return p
if dt == INT8:
return p + 1
if dt == INT16:
return p + 2
if dt == INT32:
return p + 4
if dt == INT64:
return p + 8
if dt == STRING1:
return p + 1 + d[p]
if dt == STRING4:
return p + 4 + struct.unpack_from(">i", d, p)[0]
if dt == SIMPLE_LIST:
p += 1
n, p = _read_len_int(d, p)
return p + n
if dt == MAP:
n, p = _read_len_int(d, p)
for _ in range(n):
h = d[p]
p += 1
kd = h & 0x0F
if kd == STRING1:
p += 1 + d[p]
elif kd == STRING4:
p += 4 + struct.unpack_from(">i", d, p)[0]
else:
raise ValueError(f"map key 类型 {kd:#x}")
vh = d[p]
p += 1
p = _skip_value(d, p, vh & 0x0F)
return p
if dt == LIST:
n, p = _read_len_int(d, p)
eh = d[p]
p += 1
edt = eh & 0x0F
for _ in range(n):
p = _skip_value(d, p, edt)
return p
if dt == STRUCT_BEGIN:
while True:
h = d[p]
p += 1
sdt = h & 0x0F
if sdt == STRUCT_END:
break
p = _skip_value(d, p, sdt)
return p
raise ValueError(f"未知类型 {dt:#x}@{p}")
class Envelope:
"""WUP 请求信封结构解析器与补丁器。"""
def __init__(self, raw_bytes: bytes):
self.raw = bytearray(raw_bytes)
self.tag4_span: tuple[int, int] | None = None
self.meta_json_span: tuple[int, int] | None = None
self.uid_off: int | None = None
self.cert_off: int | None = None
self.cert_len: int = 260
self._parse()
@classmethod
def load(cls, path: str | Path | None = None) -> Envelope:
"""加载信封模板，支持从文件加载或使用内嵌金样本。"""
if path:
p = Path(path)
if p.exists():
return cls._load_from_path(p)
# 尝试查找 evidence/cert_keycap.json
candidate = (
Path(__file__).resolve().parent.parent.parent
/ "evidence"
/ "cert_keycap.json"
)
if candidate.exists():
try:
return cls._load_from_path(candidate)
except Exception as exc: # noqa: BLE001
logger.debug(f"加载证书信封候选文件失败: {candidate}: {exc}")
return cls(base64.b64decode(PROTOCOL_QURL_TEMPLATE_B64))
@classmethod
def _load_from_path(cls, p: Path) -> Envelope:
if p.suffix == ".json":
j = json.loads(p.read_text("utf-8"))
q = next(
e["data"] for e in j if e.get("type") == "qurl_done" and e.get("data")
)
return cls(base64.b64decode(q))
return cls(p.read_bytes())
def _parse(self) -> None:
d = self.raw
p = 4
# WUP header
while p < len(d):
h = d[p]
p += 1
tag, dt = (h >> 4) & 0x0F, h & 0x0F
if tag == 4 and dt == INT32:
self.tag4_span = (p, p + 4)
p += 4
elif tag == 7 and dt == SIMPLE_LIST:
p += 1 # 元素类型头
n, p = _read_len_int(d, p)
self._parse_sbuffer(p, n)
break
else:
p = _skip_value(d, p, dt)
def _parse_sbuffer(self, start: int, ln: int) -> None:
d = self.raw
p = start
h = d[p]
p += 1
assert (h & 0x0F) == MAP
cnt, p = _read_len_int(d, p)
for _ in range(cnt):
p += 1
kln = d[p]
p += 1
k = bytes(d[p : p + kln])
p += kln
vh = d[p]
p += 1
if k == b"_wup_data":
assert (vh & 0x0F) == SIMPLE_LIST
p += 1
wup_data_len, p = _read_len_int(d, p)
self._parse_wup_data_struct(p, wup_data_len)
p += wup_data_len
else:
p = _skip_value(d, p, vh & 0x0F)
def _parse_wup_data_struct(self, start: int, ln: int) -> None:
d = self.raw
q = start
assert (d[q] & 0x0F) == STRUCT_BEGIN
q += 1
while q < start + ln:
hh = d[q]
q += 1
tag, dt = (hh >> 4) & 0x0F, hh & 0x0F
if tag == 15:
tag = d[q]
q += 1
if dt == STRUCT_END:
break
if tag == 0 and dt == STRUCT_BEGIN:
while True:
h2 = d[q]
q += 1
tg2, dt2 = (h2 >> 4) & 0x0F, h2 & 0x0F
if tg2 == 15:
tg2 = d[q]
q += 1
if dt2 == STRUCT_END:
break
vs = q
q = _skip_value(d, q, dt2)
if tg2 == 2 and dt2 in (STRING1, STRING4):
off = vs + (4 if dt2 == STRING4 else 1)
self.meta_json_span = (off, q)
elif tag == 3 and dt == INT64:
self.uid_off = q
q += 8
elif tag == 4 and dt == STRING4:
ln_c = struct.unpack_from(">i", d, q)[0]
self.cert_off = q + 4
self.cert_len = ln_c
q = self.cert_off + ln_c
else:
q = _skip_value(d, q, dt)
if self.uid_off is None or self.cert_off is None:
raise ValueError("未在信封中定位到 t3(uid) 或 t4(cert)")
@property
def uid(self) -> int:
assert self.uid_off is not None
return struct.unpack_from(">Q", self.raw, self.uid_off)[0]
def patch_uid(self, uid: int) -> Envelope:
assert self.uid_off is not None
struct.pack_into(">Q", self.raw, self.uid_off, uid)
return self
@property
def cert_b64(self) -> bytes:
assert self.cert_off is not None
return bytes(self.raw[self.cert_off : self.cert_off + self.cert_len])
def patch_cert(self, cert: bytes) -> Envelope:
assert self.cert_off is not None
b64 = base64.b64encode(cert)
if len(b64) != self.cert_len:
raise ValueError(
f"证书b64长度不符: {len(b64)} != 模板 {self.cert_len} (cert {len(cert)}B)"
)
self.raw[self.cert_off : self.cert_off + self.cert_len] = b64
return self
def patch_session(self, session: int) -> Envelope:
"""Patch the outer WUP session and its request copy."""
if self.tag4_span:
struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF)
d = self.raw
i = d.find(b"wupudbrequest_v0")
if i >= 0:
j = i + len(b"wupudbrequest_v0") + 4
if j + 4 <= len(d):
struct.pack_into(">I", d, j, session & 0xFFFFFFFF)
return self
def patch_meta(self, session: int, trace_id: str) -> Envelope:
"""Replace QR metadata values without carrying the capture's old state.
The captured envelope keeps a fixed-size JSON string. Keeping the
replacement the same size lets us update only the value bytes and
preserve all TAF length prefixes and offsets.
"""
if self.meta_json_span is None:
raise ValueError("信封缺少元数据 JSON")
start, end = self.meta_json_span
current = bytes(self.raw[start:end])
try:
meta = json.loads(current.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError("信封元数据 JSON 无法解析") from exc
meta["session"] = int(session)
meta["traceId"] = str(trace_id)
updated = json.dumps(meta, ensure_ascii=False, separators=(",", ":")).encode(
"utf-8"
)
if len(updated) != len(current):
raise ValueError(
f"信封元数据长度变化: {len(updated)} != {len(current)}; "
"请使用固定长度 session/traceId"
)
self.raw[start:end] = updated
return self
def wup_b64(self) -> str:
return base64.b64encode(bytes(self.raw)).decode()
ROLLBACK.sh
Executable
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env bash
set -euo pipefail
SOURCE="${1:?source path required}"
TARGET="${2:?target path required}"
cp "$SOURCE" "$TARGET"
printf 'restored %s from %s\n' "$TARGET" "$SOURCE"
VERIFICATION.txt
+33
View File
@@ -0,0 +1,33 @@
changed branch/field: Huya App login device identity and QR envelope metadata
MODIFIED_FILE: /Users/yml/codes/live-hub-py/MODIFIED_FILE
DIFF_FILE: /Users/yml/codes/live-hub-py/DIFF_FILE
VERIFICATION.txt: /Users/yml/codes/live-hub-py/VERIFICATION.txt
ROLLBACK.sh: /Users/yml/codes/live-hub-py/ROLLBACK.sh
MODIFIED_FILE original SHA-256 before fixture marker: 08bc9d7ca6425fab15af2c6218d2574f7f0180e929fa6ccb78c8b37e6ae8493a
MODIFIED_FILE changed SHA-256: 9d7240baef81e135e83cb62eada00a6a5e9f20be7d5428189bb020ac9742a557
BASELINE command:
/Users/yml/codes/live-hub-py/.venv/bin/pytest -q /tmp/live-hub-py-baseline.OFnU09/tests/test_huya_app_login.py /tmp/live-hub-py-baseline.OFnU09/tests/test_huya_dfp_register.py
BASELINE literal output/result:
24 passed, 1 warning in 1.43s
BASELINE exit status: 0
MODIFIED command:
/Users/yml/codes/live-hub-py/.venv/bin/pytest -q
MODIFIED literal output/result:
112 passed, 1 warning in 1.26s
MODIFIED exit status: 0
ROLLBACK command:
./ROLLBACK.sh /tmp/rollback-source.TCchx3 /tmp/rollback-target.atlKUI
ROLLBACK literal output/result:
restored /tmp/rollback-target.atlKUI from /tmp/rollback-source.TCchx3
restored behavior/status:
target SHA-256 after rollback = source SHA-256 = 08bc9d7ca6425fab15af2c6218d2574f7f0180e929fa6ccb78c8b37e6ae8493a; exit status 0
Additional checks:
.venv/bin/python -m ruff check core/huya tests/test_huya_app_login.py -> All checks passed, exit 0
bash -n dev.sh deploy.sh -> exit 0
node --check services/yyb-worker/runtime/probe-jsdom-pay.mjs -> exit 0
find scripts -name '*.py' -print0 | xargs -0 .venv/bin/python -m py_compile -> scripts_compile:0
core/huya/account_env.py
+9 -6
View File
@@ -39,6 +39,7 @@ from .app_login import HuyaAppPasswordLogin
from .device_profile import (
_load_db,
_save_db,
canonical_account_key,
get_profile, # 幂等画像: 同账号永远复用同一套 (data/huya_device_profiles.json)
)
@@ -68,9 +69,10 @@ def get_or_create_env(account: str, force_new: bool = False) -> dict:
(R36 解密真实结构), 每账号独立生成终身复用 → 一号一设备的一致性来源;
当前注册链服务端不校验 cw 内容, 此字段为后续真实载荷构建预留
"""
env = get_profile(account, force_new=force_new)
key = canonical_account_key(account)
env = get_profile(key, force_new=force_new)
db = _load_db()
record = dict(db.get(account) or env)
record = dict(db.get(key) or env)
changed = False
# guid32 / hebe: 一号一致字段, 首次生成后终身不变
if "guid32" not in record:
@@ -82,7 +84,7 @@ def get_or_create_env(account: str, force_new: bool = False) -> dict:
}
changed = True
if changed:
db[account] = record
db[key] = record
_save_db(db)
return record
@@ -102,7 +104,8 @@ def bind_and_login(
就是本环境的 40hex → 签发的 t2/t5 与环境绑定); safe_auth 滑块过验后的重发
沿用同一组设备字段 (app_login.login_cred_with_flow)。
"""
env = get_or_create_env(account, force_new=force_new_device)
key = canonical_account_key(account)
env = get_or_create_env(key, force_new=force_new_device)
print(
f"[env] {account} ↔ {env.get('vendor')}/{env.get('model')} "
f"fp40={env.get('fingerprint', '')[:12]}... guid32={env.get('guid32', '')[:12]}... "
@@ -124,14 +127,14 @@ def bind_and_login(
# ---- 绑定元数据回写 (令牌不入库: t2/t5 每次登录实时签发, 环境才是长期身份) ----
db = _load_db()
record = dict(db.get(account) or env)
record = dict(db.get(key) or env)
record.setdefault("bound_at", int(time.time())) # 首次绑定时间
record["last_login"] = {
"ok": result.success,
"msg": result.message[:120],
"at": int(time.time()),
}
db[account] = record
db[key] = record
_save_db(db)
return {
core/huya/app_login.py
+28 -23
View File
@@ -68,21 +68,6 @@ _URL_TAIL_KEEP = set(
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~:/?#[]@!$&'()*+,;=%"
)
DEFAULT_GOLDEN_DEV = {
"app_version": "13.4.22",
"sdk_version": "1.0.80138",
"vendor": "xiaomi",
"model": "M2102J2SC",
"os": "android",
"ip": "127.0.0.1",
"fingerprint": "02df398797432eadefcc12767119ad5e80999389",
"screen": "M2102J2SC,30,11",
"width": "1080",
"height": "2120",
"device_id": "7c5387e0539c023c31c4ff0e807e7256117385ee",
"hdid": "ed0db8334cadd236c00cadf7e11ab5a5", # HDID32 登录t1.t0 (勿与GUID32混)
}
class HuyaAppLoginError(HuyaLoginError):
"""虎牙 App 登录失败。"""
@@ -147,11 +132,14 @@ def wup_password_login_raw(
except DfpRegistrationError as exc:
raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc
dev["device_id"] = registered_device_id
hdid_value = hdid or dev.get("hdid")
if not hdid_value:
raise HuyaAppLoginError("账号设备画像缺少 HDID32，拒绝使用固定金样本")
pkt = build_password_login_wup(
uid_str,
hashlib.sha1(password.encode()).hexdigest(),
safedeviceid,
hdid or dev.get("hdid") or "ed0db8334cadd236c00cadf7e11ab5a5",
str(hdid_value),
mj["session"],
mj["traceId"],
ua,
@@ -227,6 +215,8 @@ def solve_safe_auth(
HuyaVerificationSolver,
)
if device_info is None:
raise HuyaAppLoginError("safe_auth 缺少当前账号设备画像")
q = {
k: v[0]
for k, v in parse_qs(urlparse(risk_url).query, keep_blank_values=True).items()
@@ -235,7 +225,7 @@ def solve_safe_auth(
last_err: Exception | None = None
for attempt in range(max_retry):
solver = HuyaVerificationSolver(
ua=mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV),
ua=mobile_user_agent(device_info),
proxies=proxies,
app_id=app_id,
page_url=risk_url,
@@ -357,6 +347,8 @@ class QrRole:
):
self.pc = pc
self.sdid = sdid
if not pc and device_info is None:
raise HuyaAppLoginError("移动端二维码角色缺少当前账号设备画像")
self.s = requests.Session()
self.s.trust_env = False
if proxies:
@@ -370,9 +362,7 @@ class QrRole:
self.req_counter = random.randint(40_000_000, 41_000_000)
self.s.headers.update(
{
"User-Agent": UA_PC
if pc
else mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV),
"User-Agent": UA_PC if pc else mobile_user_agent(device_info),
"Origin": UDB_BASE,
"content-type": "application/json;charset=UTF-8",
"Accept": "*/*",
@@ -511,7 +501,15 @@ class HuyaAppPasswordLogin:
raw[env.cert_off : env.cert_off + env.cert_len] = cert.encode("ascii")
if env.uid != uid:
struct.pack_into(">Q", raw, env.uid_off, uid)
wup = base64.b64encode(bytes(raw)).decode("ascii")
# QR 信封只保留协议结构；不要重放抓包里的旧会话值。
qr_session = random.randint(1_000_000, 9_999_999)
qr_trace = (
f"{uuid.uuid4().hex[:16]}-{random.randint(10000, 99999)}-"
f"{time.time_ns():020d}"
)
env.raw = raw
env.patch_session(qr_session).patch_meta(qr_session, qr_trace)
wup = base64.b64encode(bytes(env.raw)).decode("ascii")
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
return HuyaLoginResult(
success=False,
@@ -524,11 +522,18 @@ class HuyaAppPasswordLogin:
# 一号一设备: sdid 状态按账号隔离 (默认全局目录会让所有账号共享
# 同一份 hydevice 设备状态, 服务端可跨账号关联 — 见 R40 缺口修复)
sdid_obj = get_huya_sdid(
allow_fallback=True,
# App 主链只接受当前账号的 hydevice 高信任结果；旧版
# token+collect 降级没有账号画像，不再静默放行。
allow_fallback=False,
state_dir=account_state_dir(self.username),
device_hint=self.device_info,
)
sdid = sdid_obj.sdid if sdid_obj else ""
if not sdid_obj or not sdid_obj.sdid or sdid_obj.source != "fingerprint":
detail = sdid_obj.message if sdid_obj else "未返回结果"
raise HuyaAppLoginError(
f"账号设备指纹获取失败（必须为 hydevice）: {detail}"
)
sdid = sdid_obj.sdid
logger.info("[huya-app] cred 已获取，开始二维码绑定流程")
pc = QrRole(pc=True, sdid=sdid, proxies=self.proxies)
ph = QrRole(
core/huya/device_fingerprint.py
+17 -2
View File
@@ -17,7 +17,7 @@ from pathlib import Path
import requests
from loguru import logger
from .device_profile import mobile_user_agent
from .device_profile import canonical_account_key, mobile_user_agent
FINGERPRINT_DIR = Path(__file__).parent / "fingerprint"
RUNNER_JS = FINGERPRINT_DIR / "runner.js"
@@ -31,10 +31,25 @@ FP_STATE_ROOT = Path(__file__).resolve().parents[2] / "data" / "huya_fp_states"
def account_state_dir(account: str) -> Path:
"""账号专属指纹状态目录 (持久化, 保证同一账号多次登录是同一台'设备')。"""
raw = str(account or "anon").strip()
account = canonical_account_key(raw)
safe = "".join(
c if c.isalnum() or c in "-_." else "_" for c in (account or "anon")
)[:64]
return FP_STATE_ROOT / safe
target = FP_STATE_ROOT / safe
# Move a pre-rename ``hy_<numeric>`` directory on first access when the
# canonical directory does not exist. If both exist, keep the canonical
# state and leave the legacy directory untouched for manual cleanup.
if raw != account:
legacy_safe = "".join(c if c.isalnum() or c in "-_." else "_" for c in raw)[:64]
legacy = FP_STATE_ROOT / legacy_safe
if not target.exists() and legacy.exists():
try:
target.parent.mkdir(parents=True, exist_ok=True)
legacy.rename(target)
except OSError as exc:
logger.debug("迁移旧虎牙设备状态失败: {}", exc)
return target
def reset_account_state(account: str) -> None:
core/huya/device_profile.py
+34 -7
View File
@@ -19,6 +19,7 @@ import hashlib
import json
import os
import random
import re
from collections.abc import Mapping
from pathlib import Path
@@ -28,6 +29,7 @@ ROOT = Path(__file__).resolve().parent.parent.parent
DATA_DIR = ROOT / "data"
PRIMARY_PROFILE_DB = DATA_DIR / "huya_device_profiles.json"
FALLBACK_PROFILE_DB = ROOT / "evidence" / "device_profiles.json"
ALLOW_LEGACY_PROFILE_IMPORT = os.getenv("HUYA_ALLOW_LEGACY_PROFILE_IMPORT") == "1"
REAL_MODELS = [
("xiaomi", "M2102J2SC", "M2102J2SC,30,11", (1080, 2120)),
@@ -51,6 +53,19 @@ APP_VERSION = "13.4.22"
SDK_VERSION = "1.0.80138"
def canonical_account_key(account: str) -> str:
"""Return one stable environment key for equivalent Huya account forms.
The WUP protocol keeps the ``hy_`` prefix for Huya IDs, but the device
environment must not split ``300023887`` and ``hy_300023887`` into two
records. Phone numbers and other usernames remain unchanged.
"""
value = str(account or "").strip()
if re.fullmatch(r"hy_\d+", value):
return value[3:]
return value
def mobile_user_agent(device_info: Mapping[str, object]) -> str:
"""根据统一设备画像生成 App WebView UA。"""
screen = str(device_info.get("screen") or "")
@@ -107,8 +122,14 @@ def record_login(account: str, ok: bool, message: str = "") -> None:
"""
import time as _time
key = canonical_account_key(account)
db = _load_db()
rec = db.get(account)
rec = db.get(key)
if rec is None and key != account:
rec = db.get(account)
if rec is not None:
db[key] = rec
del db[account]
if rec is None:
return # 尚无环境的账号 (纯 Cookie 导入) 不生成记录
now = int(_time.time())
@@ -123,11 +144,11 @@ def _load_db() -> dict:
return json.loads(PRIMARY_PROFILE_DB.read_text("utf-8"))
except Exception as exc: # noqa: BLE001
logger.debug(f"读取主设备画像库失败: {exc}")
if FALLBACK_PROFILE_DB.exists():
if ALLOW_LEGACY_PROFILE_IMPORT and FALLBACK_PROFILE_DB.exists():
try:
return json.loads(FALLBACK_PROFILE_DB.read_text("utf-8"))
except Exception as exc: # noqa: BLE001
logger.debug(f"读取备用设备画像库失败: {exc}")
logger.debug(f"读取显式迁移画像库失败: {exc}")
return {}
@@ -166,14 +187,20 @@ def _enrich_profile(profile: dict) -> tuple[dict, bool]:
def get_profile(account: str, force_new: bool = False) -> dict:
"""按账号获取或创建画像（幂等：同账号复用同一套, 自动补齐缺失一致性字段）。"""
key = canonical_account_key(account)
db = _load_db()
if not force_new and account in db:
enriched, changed = _enrich_profile(db[account])
# Migrate the old prefixed key lazily without discarding its environment.
legacy_key = str(account or "").strip()
if key not in db and legacy_key != key and legacy_key in db:
db[key] = db.pop(legacy_key)
_save_db(db)
if not force_new and key in db:
enriched, changed = _enrich_profile(db[key])
if changed:
db[account] = enriched
db[key] = enriched
_save_db(db)
return enriched
p, _ = _enrich_profile(generate_profile())
db[account] = p
db[key] = p
_save_db(db)
return p
core/huya/dfp_register.py
+19 -11
View File
@@ -90,7 +90,7 @@ def _build_select_operator_request(
# 仅供无画像的协议级独立调用兜底；生产登录始终传入账号画像。
"app_version": "13.4.22",
"model": "M2102J2SC",
"fingerprint": fingerprint or "02df398797432eadefcc12767119ad5e80999389",
"fingerprint": fingerprint or hashlib.sha1(os.urandom(20)).hexdigest(),
"screen": "M2102J2SC,30,11",
}
if device_info:
@@ -248,16 +248,21 @@ def _build_dfp_json_plain(device_info: Mapping[str, str] | None = None) -> bytes
def _select_operator_request(template: bytes, fingerprint: str | None) -> bytes:
if fingerprint and len(fingerprint) == 40:
old = b"02df398797432eadefcc12767119ad5e80999389"
index = template.find(old)
if index >= 0:
return (
template[:index]
+ fingerprint.encode("ascii")
+ template[index + len(old) :]
)
return template
"""Inject a current 40-hex fingerprint into a legacy request shape.
This compatibility path only operates on the shape supplied by a caller;
it does not contain or search for a particular captured device value.
"""
if not fingerprint or len(fingerprint) != 40:
return template
match = re.search(rb"(?<![0-9a-f])[0-9a-f]{40}(?![0-9a-f])", template)
if match is None:
return template
return (
template[: match.start()]
+ fingerprint.encode("ascii")
+ template[match.end() :]
)
def _parse_response(data: bytes) -> tuple[str, str, str]:
@@ -284,6 +289,9 @@ def register_device(
"""执行新注册链，返回 ``(t1, safedeviceid, device_id)``。"""
chain = _load_chain(fingerprint=fingerprint, device_info=device_info)
_post(chain["getDfpConfig"][0], timeout=timeout, proxies=proxies)
# The generated request already contains the current profile fingerprint.
# Keep a generic shape-only compatibility patch for injected test/custom
# templates; no captured device value is embedded in this module.
select_request = _select_operator_request(chain["selectOperator"][0], fingerprint)
_post(select_request, "application/x-wup", timeout, proxies)
response = _post(
core/huya/envelope_forge.py
+31 -2
View File
@@ -18,7 +18,7 @@ MAP, LIST = 0x08, 0x09
STRUCT_BEGIN, STRUCT_END = 0x0A, 0x0B
ZERO, SIMPLE_LIST = 0x0C, 0x0D
DEFAULT_QURL_B64 = (
PROTOCOL_QURL_TEMPLATE_B64 = (
"AAAD5hADLDxCAFpBBVYMaHV5YXVkYndlYnVpZgdkZWZhdWx0fQABA7gIAAIGCV93dXBfZGF0YR0AAQOKCgoMFgMxLjAm"
"ynsiYXNzb2NpYXRpb25JZCI6MTg0NTQ5MzkyLCJmdW5jTmFtZSI6IiIsImdyb3VwIjowLCJpZCI6MTg0NTQ5MzkyLCJz"
"ZXNzaW9uIjo1OTE0ODg1LCJzdGVwIjowLCJzdGlsbExvZ2luIjpmYWxzZSwidHJhY2VJZCI6IjBiOGYwOThmZjY0YTVi"
@@ -135,7 +135,7 @@ class Envelope:
return cls._load_from_path(candidate)
except Exception as exc: # noqa: BLE001
logger.debug(f"加载证书信封候选文件失败: {candidate}: {exc}")
return cls(base64.b64decode(DEFAULT_QURL_B64))
return cls(base64.b64decode(PROTOCOL_QURL_TEMPLATE_B64))
@classmethod
def _load_from_path(cls, p: Path) -> Envelope:
@@ -258,6 +258,7 @@ class Envelope:
return self
def patch_session(self, session: int) -> Envelope:
"""Patch the outer WUP session and its request copy."""
if self.tag4_span:
struct.pack_into(">I", self.raw, self.tag4_span[0], session & 0xFFFFFFFF)
d = self.raw
@@ -268,5 +269,33 @@ class Envelope:
struct.pack_into(">I", d, j, session & 0xFFFFFFFF)
return self
def patch_meta(self, session: int, trace_id: str) -> Envelope:
"""Replace QR metadata values without carrying the capture's old state.
The captured envelope keeps a fixed-size JSON string. Keeping the
replacement the same size lets us update only the value bytes and
preserve all TAF length prefixes and offsets.
"""
if self.meta_json_span is None:
raise ValueError("信封缺少元数据 JSON")
start, end = self.meta_json_span
current = bytes(self.raw[start:end])
try:
meta = json.loads(current.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError("信封元数据 JSON 无法解析") from exc
meta["session"] = int(session)
meta["traceId"] = str(trace_id)
updated = json.dumps(meta, ensure_ascii=False, separators=(",", ":")).encode(
"utf-8"
)
if len(updated) != len(current):
raise ValueError(
f"信封元数据长度变化: {len(updated)} != {len(current)}; "
"请使用固定长度 session/traceId"
)
self.raw[start:end] = updated
return self
def wup_b64(self) -> str:
return base64.b64encode(bytes(self.raw)).decode()
tests/test_huya_app_login.py
+43 -8
View File
@@ -1,5 +1,6 @@
"""虎牙 App 密码登录及相关组件测试。"""
import json
import os
import struct
from unittest.mock import MagicMock, patch
@@ -11,14 +12,14 @@ from sqlalchemy.orm import sessionmaker
from core.huya import (
HuyaAppLoginError,
)
from core.huya.app_login import (
DEFAULT_GOLDEN_DEV,
login_cred_with_flow,
wup_password_login_raw,
)
from core.huya.app_login import login_cred_with_flow, wup_password_login_raw
from core.huya.cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1
from core.huya.device_fingerprint import account_state_dir, reset_account_state
from core.huya.device_profile import generate_profile, get_profile
from core.huya.device_profile import (
canonical_account_key,
generate_profile,
get_profile,
)
from core.huya.dfp_register import DfpRegistrationError
from core.huya.envelope_forge import Envelope
from core.huya.login import HuyaLoginResult
@@ -91,6 +92,16 @@ class TestHuyaAppLogin:
wup_b64 = env.wup_b64()
assert len(wup_b64) > 0
def test_envelope_metadata_is_fresh(self):
env = Envelope.load()
old = bytes(env.raw[env.meta_json_span[0] : env.meta_json_span[1]])
new_trace = "a" * 16 + "-12345-" + "9" * 20
env.patch_session(7654321).patch_meta(7654321, new_trace)
current = bytes(env.raw[env.meta_json_span[0] : env.meta_json_span[1]])
assert current != old
assert b'"session":7654321' in current
assert (b'"traceId":"' + new_trace.encode() + b'"') in current
def test_device_profile_generation(self):
p1 = generate_profile()
assert p1["os"] == "android"
@@ -99,12 +110,34 @@ class TestHuyaAppLogin:
assert p1["hdid"] == "ed0db8334cadd236c00cadf7e11ab5a5"
# 画像不再承载 safedeviceid：该令牌由 dfp_register 注册链每次登录前签发
assert "safedeviceid" not in p1
assert "safedeviceid" not in DEFAULT_GOLDEN_DEV
p2 = get_profile("test_user_account_123")
p3 = get_profile("test_user_account_123")
assert p2["fingerprint"] == p3["fingerprint"]
def test_huya_id_aliases_share_one_profile_key(self, tmp_path, monkeypatch):
profile_db = tmp_path / "profiles.json"
monkeypatch.setattr("core.huya.device_profile.PRIMARY_PROFILE_DB", profile_db)
monkeypatch.setattr(
"core.huya.device_profile.FALLBACK_PROFILE_DB", tmp_path / "missing.json"
)
assert canonical_account_key(" hy_300023887 ") == "300023887"
first = get_profile("300023887")
second = get_profile("hy_300023887")
assert first["fingerprint"] == second["fingerprint"]
assert list(json.loads(profile_db.read_text())) == ["300023887"]
def test_legacy_evidence_profile_is_not_loaded_by_default(
self, tmp_path, monkeypatch
):
primary = tmp_path / "profiles.json"
legacy = tmp_path / "legacy.json"
legacy.write_text(json.dumps({"old": {"fingerprint": "f" * 40}}))
monkeypatch.setattr("core.huya.device_profile.PRIMARY_PROFILE_DB", primary)
monkeypatch.setattr("core.huya.device_profile.FALLBACK_PROFILE_DB", legacy)
profile = get_profile("old")
assert profile["fingerprint"] != "f" * 40
def test_force_new_device_clears_hydevice_state(self, tmp_path, monkeypatch):
monkeypatch.setattr(
"core.huya.device_fingerprint.FP_STATE_ROOT", tmp_path / "fp"
@@ -184,7 +217,9 @@ class TestHuyaAppLogin:
assert captured["args"][2] == new_action
assert captured["args"][7]["device_id"] == new_device_id
# 画像默认值里的旧 device_id 被注册结果覆盖，而非沿用
assert captured["args"][7]["device_id"] != DEFAULT_GOLDEN_DEV["device_id"]
assert captured["args"][7]["device_id"] != (
"7c5387e0539c023c31c4ff0e807e7256117385ee"
)
def test_wup_login_registration_failure_is_explicit(self):
"""注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。"""