From 05b1325a0454a57d79e466cc7dead69098dae3e4 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 31 Aug 2026 17:24:37 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B8=85=E7=90=86=E8=99=8E=E7=89=99=E7=99=BB?= =?UTF-8?q?=E5=BD=95=E6=97=A7=E6=95=B0=E6=8D=AE=E4=BE=9D=E8=B5=96=E5=B9=B6?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E5=8A=A8=E6=80=81=E7=8E=AF=E5=A2=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DIFF_FILE | 890 ++++++++++++++++++++++++++++++++ MODIFIED_FILE | 303 +++++++++++ ROLLBACK.sh | 7 + VERIFICATION.txt | 33 ++ core/huya/account_env.py | 15 +- core/huya/app_login.py | 51 +- core/huya/device_fingerprint.py | 19 +- core/huya/device_profile.py | 41 +- core/huya/dfp_register.py | 30 +- core/huya/envelope_forge.py | 33 +- tests/test_huya_app_login.py | 51 +- 11 files changed, 1414 insertions(+), 59 deletions(-) create mode 100644 DIFF_FILE create mode 100644 MODIFIED_FILE create mode 100755 ROLLBACK.sh create mode 100644 VERIFICATION.txt diff --git a/DIFF_FILE b/DIFF_FILE new file mode 100644 index 0000000..cc16d1a --- /dev/null +++ b/DIFF_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_`` 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"(? 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): + """注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。""" 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_`` 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"(? 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): """注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。"""