diff --git a/DIFF_FILE b/DIFF_FILE index cc16d1a..73a3f3c 100644 --- a/DIFF_FILE +++ b/DIFF_FILE @@ -1,890 +1,293 @@ -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 +index afa33a4..86e18a9 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 +@@ -9,7 +9,8 @@ + 6. POST /web/cookie/verify 兑换获取全套网页 Cookie + 注册链不再重放旧 dfpReport 密文。登录帧的 32hex hdid 仍是服务端硬锚， +-当前继续使用已注册样本；新注册链动态更新的是 safedeviceid 和 device_id。 ++登录帧中的 HDID32 是 app 版本级协议常量；二维码信封的 ACTION、设备字段、 ++UID、证书和会话元数据均在本次登录中动态生成或由注册链签发。 + 注册链（``core/huya/dfp_register``）每次登录前执行，失败即抛错终止（``HuyaAppLoginError``）， + 不读取画像里的旧固定值，也不静默回退旧链。 +@@ -35,7 +36,7 @@ from urllib.parse import parse_qs, quote, urlparse 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(): +-from .cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1 ++from .cert_forge import build_p1, forge_cert + from .cookie_utils import normalize_huya_cookie + from .device_fingerprint import account_state_dir, get_huya_sdid, reset_account_state + from .device_profile import get_profile, mobile_user_agent +@@ -270,7 +271,8 @@ def login_cred_with_flow( + max_rounds: int = 3, + device_info: dict | None = None, + proxies: dict | None = None, +-) -> tuple[bytes, int]: ++ include_device_token: bool = False, ++) -> tuple[bytes, int] | tuple[bytes, int, str]: + """新注册设备后登录，返回 ``(新鲜 cred, 真实 uid)``。 + 注册只执行一次；safe_auth 通过后的重发继续使用同一组设备字段。 +@@ -305,6 +307,8 @@ def login_cred_with_flow( + ) + if cred: + uid = parse_real_uid(resp) ++ if include_device_token: ++ return cred, uid, safedeviceid + return cred, uid + if risk_url: + kind = ( +@@ -462,11 +466,12 @@ class HuyaAppPasswordLogin: + # 1) 获取新鲜 cred 与 真实 uid (自动过 safe_auth 滑块) 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. +- cred, uid = login_cred_with_flow( ++ cred, uid, safedeviceid = login_cred_with_flow( + acct, + self.password, + device_info=self.device_info, + proxies=self.proxies, ++ include_device_token=True, + ) + except HuyaAppQrAuthRequiredError as exc: + return HuyaLoginResult( +@@ -485,30 +490,27 @@ class HuyaAppPasswordLogin: + # 2) 本地生成 nonce 铸造证书 (P1 指纹与该账号设备画像一致) + try: +- env = Envelope.load() +- orig = base64.b64decode(env.cert_b64) +- f = parse_p1(decrypt_cert(orig)) + st = int(time.time() * 1000) + rnd = gen_nonce(uid, K1_DEFAULT, service_time_ms=st, counter=0) + fp_bytes = self.device_info["fingerprint"].encode("ascii") +- p1 = build_p1(f["app_id"], fp_bytes, cred, rnd=rnd) +- cert = base64.b64encode(forge_cert(p1, key_idx=orig[1])).decode() +- +- # 3) 信封补丁 +- raw = bytearray(env.raw) +- if env.cert_off is None or env.uid_off is None: +- raise ValueError("信封缺少证书或 uid 偏移") +- 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) +- # QR 信封只保留协议结构；不要重放抓包里的旧会话值。 ++ p1 = build_p1(b"5008", fp_bytes, cred, rnd=rnd) ++ # 证书头和 app_id 是协议版本常量，不从抓包信封读取。 ++ cert = base64.b64encode(forge_cert(p1, key_idx=0x20)).decode() + -+ 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( ++ # 3) 按当前账号画像和本次注册令牌重编码二维码信封。 + 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) ++ env = Envelope.build_qr( ++ uid=uid, ++ cert_b64=cert, ++ safedeviceid=safedeviceid, ++ session=qr_session, ++ trace_id=qr_trace, ++ device_info=self.device_info, ++ ) + wup = base64.b64encode(bytes(env.raw)).decode("ascii") + except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 + return HuyaLoginResult( diff --git a/core/huya/envelope_forge.py b/core/huya/envelope_forge.py -index f103a7e..c5d66ab 100644 +index c5d66ab..0d2e8b1 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)) - +@@ -1,6 +1,7 @@ + """wupData 信封构造与补丁工具。 +-解析与改写 WUP 信封中的 cert、uid、session 等字段。 ++生产二维码信封由 :meth:`Envelope.build_qr` 按当前账号状态编码； ++:meth:`Envelope.load` 仅保留协议结构模板兼容测试和离线分析。 + """ + from __future__ import annotations +@@ -10,7 +11,8 @@ import json + import struct + from pathlib import Path +-from loguru import logger ++from .taf_protocol import TafOutputStream ++from .wup_protocol import WupRequest + INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03 + STRING1, STRING4 = 0x06, 0x07 +@@ -119,24 +121,106 @@ class Envelope: + @classmethod + def load(cls, path: str | Path | None = None) -> Envelope: +- """加载信封模板，支持从文件加载或使用内嵌金样本。""" ++ """加载显式模板，或使用内嵌协议结构模板。 ++ ++ 不自动扫描 ``evidence/``，避免历史抓包成为隐式生产输入。 ++ """ + 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 build_qr( ++ cls, ++ *, ++ uid: int, ++ cert_b64: str, ++ safedeviceid: str, ++ session: int, ++ trace_id: str, ++ device_info: dict[str, str], ++ ) -> Envelope: ++ """按当前账号状态编码二维码绑定信封。 ++ ++ 生产二维码请求不应从抓包信封复制字段。这里仅复用已知的 TAF/WUP ++ 字段布局；证书、ACTION、UID、会话和全部设备画像都由本次登录提供。 ++ """ ++ if len(cert_b64) != 260: ++ raise ValueError(f"证书 base64 长度异常: {len(cert_b64)} != 260") ++ if len(safedeviceid) != 180: ++ raise ValueError(f"safedeviceid 长度异常: {len(safedeviceid)} != 180") ++ dev = {k: str(v) for k, v in (device_info or {}).items()} ++ meta = json.dumps( ++ { ++ "associationId": 184549392, ++ "funcName": "", ++ "group": 0, ++ "id": 184549392, ++ "session": int(session), ++ "step": 0, ++ "stillLogin": False, ++ "traceId": str(trace_id), ++ "type": 2, ++ "uid": 0, ++ "userContext": "", ++ }, ++ ensure_ascii=False, ++ separators=(",", ":"), ++ ) ++ body = TafOutputStream() ++ body.write_struct_begin(0) ++ body.write_struct_begin(0) ++ body.write_int8(0, 0) ++ body.write_string(1, "1.0") ++ body.write_string(2, meta) ++ body.write_string(3, "5008") ++ body.write_int8(4, 3) ++ body.write_string(5, safedeviceid) ++ body.write_string(6, "") ++ body.write_string(7, "") ++ body.write_string(8, "") ++ body.write_string(9, "") ++ body.write_struct_end() ++ ++ body.write_struct_begin(1) ++ body.write_string(0, dev.get("hdid", "")) ++ body.write_string(1, dev.get("app_version", "13.4.22")) ++ body.write_string(2, dev.get("sdk_version", "1.0.80138")) ++ body.write_string(3, "") ++ body.write_string(4, dev.get("ip", "127.0.0.1")) ++ body.write_string(5, dev.get("vendor", "android")) ++ body.write_string(6, "") ++ body.write_struct_end() ++ ++ body.write_struct_begin(2) ++ body.write_int8(0, 1) ++ body.write_string(1, dev.get("model", "")) ++ body.write_string(2, dev.get("fingerprint", "")) ++ body.write_string(3, dev.get("os", "android")) ++ body.write_string(4, dev.get("screen", "")) ++ body.write_string(6, dev.get("width", "1080")) ++ body.write_string(7, dev.get("height", "2120")) ++ body.write_string(8, dev.get("device_id", "")) ++ body.write_struct_end() ++ body.write_uint64(3, int(uid)) ++ body.write_string(4, cert_b64) ++ body.write_string(5, "") ++ body.write_string(6, "") ++ body.write_struct_end() ++ ++ req = TafOutputStream() ++ req.write_int32(0, int(session)) ++ wup = WupRequest() ++ wup.iTimeout = 0 ++ wup.setRequestId(int(session)) ++ wup.setServant("huyaudbwebui") ++ wup.setFunc("default") ++ wup.newdata["_wup_data"] = body.get_bytes() ++ wup.newdata["wupudbrequest_v0"] = req.get_bytes() ++ return cls(wup.encode()) ++ @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() + if p.suffix == ".json": diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py -index 476e0ed..3512a14 100644 +index 3512a14..0c897bb 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 +@@ -26,6 +26,7 @@ from core.huya.login import HuyaLoginResult + from core.huya.nonce_forge import K1_DEFAULT, gen_nonce + from core.huya.udb_aes import udb_decrypt, udb_encrypt + from core.huya.wup_encoder import _make_name, build_password_login_wup ++from core.huya.wup_protocol import WupResponse + from web.backend.database import Base + from web.backend.models import HuyaAccount, User + from web.backend.routers.huya import ( +@@ -102,6 +103,38 @@ class TestHuyaAppLogin: + assert b'"session":7654321' in current + assert (b'"traceId":"' + new_trace.encode() + b'"') in current ++ def test_qr_envelope_is_fully_dynamic(self): ++ profile = { ++ "hdid": "h" * 32, ++ "app_version": "13.4.22", ++ "sdk_version": "1.0.80138", ++ "ip": "10.0.0.8", ++ "vendor": "vivo", ++ "model": "V2370A", ++ "fingerprint": "f" * 40, ++ "os": "android", ++ "screen": "V2370A,34,13", ++ "width": "1080", ++ "height": "2412", ++ "device_id": "d" * 40, ++ } ++ env = Envelope.build_qr( ++ uid=1199664135026, ++ cert_b64="A" * 260, ++ safedeviceid="B" * 180, ++ session=7654321, ++ trace_id="t" * 16 + "-12345-" + "9" * 20, ++ device_info=profile, ++ ) ++ wup = WupResponse() ++ wup.decode(bytes(env.raw)) ++ body = wup.newdata["_wup_data"] ++ for value in (b"B" * 180, b"h" * 32, b"f" * 40, b"V2370A", b"d" * 40): ++ assert value in body ++ assert b"PQwemAN9NHkZKoMq" not in body ++ assert b"02df398797432eadefcc12767119ad5e80999389" not in body ++ assert b"7c5387e0539c023c31c4ff0e807e7256117385ee" not in body + 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): - """注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。""" +@@ -271,10 +304,13 @@ class TestHuyaAppLogin: + patch("core.huya.app_login.solve_safe_auth", return_value={"authId": "id"}), + patch("core.huya.app_login.parse_real_uid", return_value=1199666914671), + ): +- cred, uid = login_cred_with_flow("300023887", "pw") ++ cred, uid, safedeviceid = login_cred_with_flow( ++ "300023887", "pw", include_device_token=True ++ ) + assert cred == b"c" * 114 + assert uid == 1199666914671 ++ assert safedeviceid == "A" * 180 + assert calls == [assets, assets] + def test_login_cred_flow_maps_invalid_password_response(self): diff --git a/MODIFIED_FILE b/MODIFIED_FILE index 280bd22..0d2e8b1 100644 --- a/MODIFIED_FILE +++ b/MODIFIED_FILE @@ -1,10 +1,9 @@ """wupData 信封构造与补丁工具。 -解析与改写 WUP 信封中的 cert、uid、session 等字段。 +生产二维码信封由 :meth:`Envelope.build_qr` 按当前账号状态编码； +:meth:`Envelope.load` 仅保留协议结构模板兼容测试和离线分析。 """ -# Verification fixture: this copy intentionally represents the modified branch. - from __future__ import annotations import base64 @@ -12,7 +11,8 @@ import json import struct from pathlib import Path -from loguru import logger +from .taf_protocol import TafOutputStream +from .wup_protocol import WupRequest INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03 STRING1, STRING4 = 0x06, 0x07 @@ -121,24 +121,106 @@ class Envelope: @classmethod def load(cls, path: str | Path | None = None) -> Envelope: - """加载信封模板，支持从文件加载或使用内嵌金样本。""" + """加载显式模板，或使用内嵌协议结构模板。 + + 不自动扫描 ``evidence/``，避免历史抓包成为隐式生产输入。 + """ 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 build_qr( + cls, + *, + uid: int, + cert_b64: str, + safedeviceid: str, + session: int, + trace_id: str, + device_info: dict[str, str], + ) -> Envelope: + """按当前账号状态编码二维码绑定信封。 + + 生产二维码请求不应从抓包信封复制字段。这里仅复用已知的 TAF/WUP + 字段布局；证书、ACTION、UID、会话和全部设备画像都由本次登录提供。 + """ + if len(cert_b64) != 260: + raise ValueError(f"证书 base64 长度异常: {len(cert_b64)} != 260") + if len(safedeviceid) != 180: + raise ValueError(f"safedeviceid 长度异常: {len(safedeviceid)} != 180") + dev = {k: str(v) for k, v in (device_info or {}).items()} + meta = json.dumps( + { + "associationId": 184549392, + "funcName": "", + "group": 0, + "id": 184549392, + "session": int(session), + "step": 0, + "stillLogin": False, + "traceId": str(trace_id), + "type": 2, + "uid": 0, + "userContext": "", + }, + ensure_ascii=False, + separators=(",", ":"), + ) + body = TafOutputStream() + body.write_struct_begin(0) + body.write_struct_begin(0) + body.write_int8(0, 0) + body.write_string(1, "1.0") + body.write_string(2, meta) + body.write_string(3, "5008") + body.write_int8(4, 3) + body.write_string(5, safedeviceid) + body.write_string(6, "") + body.write_string(7, "") + body.write_string(8, "") + body.write_string(9, "") + body.write_struct_end() + + body.write_struct_begin(1) + body.write_string(0, dev.get("hdid", "")) + body.write_string(1, dev.get("app_version", "13.4.22")) + body.write_string(2, dev.get("sdk_version", "1.0.80138")) + body.write_string(3, "") + body.write_string(4, dev.get("ip", "127.0.0.1")) + body.write_string(5, dev.get("vendor", "android")) + body.write_string(6, "") + body.write_struct_end() + + body.write_struct_begin(2) + body.write_int8(0, 1) + body.write_string(1, dev.get("model", "")) + body.write_string(2, dev.get("fingerprint", "")) + body.write_string(3, dev.get("os", "android")) + body.write_string(4, dev.get("screen", "")) + body.write_string(6, dev.get("width", "1080")) + body.write_string(7, dev.get("height", "2120")) + body.write_string(8, dev.get("device_id", "")) + body.write_struct_end() + body.write_uint64(3, int(uid)) + body.write_string(4, cert_b64) + body.write_string(5, "") + body.write_string(6, "") + body.write_struct_end() + + req = TafOutputStream() + req.write_int32(0, int(session)) + wup = WupRequest() + wup.iTimeout = 0 + wup.setRequestId(int(session)) + wup.setServant("huyaudbwebui") + wup.setFunc("default") + wup.newdata["_wup_data"] = body.get_bytes() + wup.newdata["wupudbrequest_v0"] = req.get_bytes() + return cls(wup.encode()) + @classmethod def _load_from_path(cls, p: Path) -> Envelope: if p.suffix == ".json": diff --git a/VERIFICATION.txt b/VERIFICATION.txt index d52b3d6..1c37d55 100644 --- a/VERIFICATION.txt +++ b/VERIFICATION.txt @@ -1,33 +1,35 @@ -changed branch/field: Huya App login device identity and QR envelope metadata +changed branch/field: Huya App login QR envelope and device-token propagation 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 +MODIFIED_FILE original SHA-256 before fixture update: 08bc9d7ca6425fab15af2c6218d2574f7f0180e929fa6ccb78c8b37e6ae8493a +MODIFIED_FILE changed SHA-256: e15f95ce30ca0e0faa6e3c8c6cdf4725e4984434df0f39cf237ee0b7197375b1 -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 exact command: +set -e; BASE=/tmp/live-hub-py-baseline.PJsjpJ; git archive HEAD | tar -x -C "$BASE"; .venv/bin/pytest -q "$BASE/tests/test_huya_app_login.py" "$BASE/tests/test_huya_dfp_register.py" BASELINE literal output/result: -24 passed, 1 warning in 1.43s +27 passed, 1 warning in 0.62s BASELINE exit status: 0 -MODIFIED command: -/Users/yml/codes/live-hub-py/.venv/bin/pytest -q +MODIFIED exact command: +.venv/bin/pytest -q MODIFIED literal output/result: -112 passed, 1 warning in 1.26s +113 passed, 1 warning in 1.13s MODIFIED exit status: 0 -ROLLBACK command: -./ROLLBACK.sh /tmp/rollback-source.TCchx3 /tmp/rollback-target.atlKUI +ROLLBACK exact command: +set -e; SRC=/tmp/rollback-source.mKNKoP; TGT=/tmp/rollback-target.hHBqnQ; cp MODIFIED_FILE "$SRC"; printf '\ncorrupt\n' >> "$TGT"; ./ROLLBACK.sh "$SRC" "$TGT"; cmp -s "$SRC" "$TGT" 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 +restored /tmp/rollback-target.hHBqnQ from /tmp/rollback-source.mKNKoP +rollback_status=restored +source_sha256=e15f95ce30ca0e0faa6e3c8c6cdf4725e4984434df0f39cf237ee0b7197375b1 +target_sha256=e15f95ce30ca0e0faa6e3c8c6cdf4725e4984434df0f39cf237ee0b7197375b1 +ROLLBACK exit status: 0 +restored behavior/status: target byte-for-byte equals source; MODIFIED_FILE remains changed. Additional checks: .venv/bin/python -m ruff check core/huya tests/test_huya_app_login.py -> All checks passed, exit 0 +git diff --check -> 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/app_login.py b/core/huya/app_login.py index afa33a4..86e18a9 100644 --- a/core/huya/app_login.py +++ b/core/huya/app_login.py @@ -9,7 +9,8 @@ 6. POST /web/cookie/verify 兑换获取全套网页 Cookie 注册链不再重放旧 dfpReport 密文。登录帧的 32hex hdid 仍是服务端硬锚， -当前继续使用已注册样本；新注册链动态更新的是 safedeviceid 和 device_id。 +登录帧中的 HDID32 是 app 版本级协议常量；二维码信封的 ACTION、设备字段、 +UID、证书和会话元数据均在本次登录中动态生成或由注册链签发。 注册链（``core/huya/dfp_register``）每次登录前执行，失败即抛错终止（``HuyaAppLoginError``）， 不读取画像里的旧固定值，也不静默回退旧链。 @@ -35,7 +36,7 @@ from urllib.parse import parse_qs, quote, urlparse import requests from loguru import logger -from .cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1 +from .cert_forge import build_p1, forge_cert from .cookie_utils import normalize_huya_cookie from .device_fingerprint import account_state_dir, get_huya_sdid, reset_account_state from .device_profile import get_profile, mobile_user_agent @@ -270,7 +271,8 @@ def login_cred_with_flow( max_rounds: int = 3, device_info: dict | None = None, proxies: dict | None = None, -) -> tuple[bytes, int]: + include_device_token: bool = False, +) -> tuple[bytes, int] | tuple[bytes, int, str]: """新注册设备后登录，返回 ``(新鲜 cred, 真实 uid)``。 注册只执行一次；safe_auth 通过后的重发继续使用同一组设备字段。 @@ -305,6 +307,8 @@ def login_cred_with_flow( ) if cred: uid = parse_real_uid(resp) + if include_device_token: + return cred, uid, safedeviceid return cred, uid if risk_url: kind = ( @@ -462,11 +466,12 @@ class HuyaAppPasswordLogin: # 1) 获取新鲜 cred 与 真实 uid (自动过 safe_auth 滑块) try: - cred, uid = login_cred_with_flow( + cred, uid, safedeviceid = login_cred_with_flow( acct, self.password, device_info=self.device_info, proxies=self.proxies, + include_device_token=True, ) except HuyaAppQrAuthRequiredError as exc: return HuyaLoginResult( @@ -485,30 +490,27 @@ class HuyaAppPasswordLogin: # 2) 本地生成 nonce 铸造证书 (P1 指纹与该账号设备画像一致) try: - env = Envelope.load() - orig = base64.b64decode(env.cert_b64) - f = parse_p1(decrypt_cert(orig)) st = int(time.time() * 1000) rnd = gen_nonce(uid, K1_DEFAULT, service_time_ms=st, counter=0) fp_bytes = self.device_info["fingerprint"].encode("ascii") - p1 = build_p1(f["app_id"], fp_bytes, cred, rnd=rnd) - cert = base64.b64encode(forge_cert(p1, key_idx=orig[1])).decode() + p1 = build_p1(b"5008", fp_bytes, cred, rnd=rnd) + # 证书头和 app_id 是协议版本常量，不从抓包信封读取。 + cert = base64.b64encode(forge_cert(p1, key_idx=0x20)).decode() - # 3) 信封补丁 - raw = bytearray(env.raw) - if env.cert_off is None or env.uid_off is None: - raise ValueError("信封缺少证书或 uid 偏移") - 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) - # QR 信封只保留协议结构；不要重放抓包里的旧会话值。 + # 3) 按当前账号画像和本次注册令牌重编码二维码信封。 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) + env = Envelope.build_qr( + uid=uid, + cert_b64=cert, + safedeviceid=safedeviceid, + session=qr_session, + trace_id=qr_trace, + device_info=self.device_info, + ) wup = base64.b64encode(bytes(env.raw)).decode("ascii") except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底 return HuyaLoginResult( diff --git a/core/huya/envelope_forge.py b/core/huya/envelope_forge.py index c5d66ab..0d2e8b1 100644 --- a/core/huya/envelope_forge.py +++ b/core/huya/envelope_forge.py @@ -1,6 +1,7 @@ """wupData 信封构造与补丁工具。 -解析与改写 WUP 信封中的 cert、uid、session 等字段。 +生产二维码信封由 :meth:`Envelope.build_qr` 按当前账号状态编码； +:meth:`Envelope.load` 仅保留协议结构模板兼容测试和离线分析。 """ from __future__ import annotations @@ -10,7 +11,8 @@ import json import struct from pathlib import Path -from loguru import logger +from .taf_protocol import TafOutputStream +from .wup_protocol import WupRequest INT8, INT16, INT32, INT64 = 0x00, 0x01, 0x02, 0x03 STRING1, STRING4 = 0x06, 0x07 @@ -119,24 +121,106 @@ class Envelope: @classmethod def load(cls, path: str | Path | None = None) -> Envelope: - """加载信封模板，支持从文件加载或使用内嵌金样本。""" + """加载显式模板，或使用内嵌协议结构模板。 + + 不自动扫描 ``evidence/``，避免历史抓包成为隐式生产输入。 + """ 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 build_qr( + cls, + *, + uid: int, + cert_b64: str, + safedeviceid: str, + session: int, + trace_id: str, + device_info: dict[str, str], + ) -> Envelope: + """按当前账号状态编码二维码绑定信封。 + + 生产二维码请求不应从抓包信封复制字段。这里仅复用已知的 TAF/WUP + 字段布局；证书、ACTION、UID、会话和全部设备画像都由本次登录提供。 + """ + if len(cert_b64) != 260: + raise ValueError(f"证书 base64 长度异常: {len(cert_b64)} != 260") + if len(safedeviceid) != 180: + raise ValueError(f"safedeviceid 长度异常: {len(safedeviceid)} != 180") + dev = {k: str(v) for k, v in (device_info or {}).items()} + meta = json.dumps( + { + "associationId": 184549392, + "funcName": "", + "group": 0, + "id": 184549392, + "session": int(session), + "step": 0, + "stillLogin": False, + "traceId": str(trace_id), + "type": 2, + "uid": 0, + "userContext": "", + }, + ensure_ascii=False, + separators=(",", ":"), + ) + body = TafOutputStream() + body.write_struct_begin(0) + body.write_struct_begin(0) + body.write_int8(0, 0) + body.write_string(1, "1.0") + body.write_string(2, meta) + body.write_string(3, "5008") + body.write_int8(4, 3) + body.write_string(5, safedeviceid) + body.write_string(6, "") + body.write_string(7, "") + body.write_string(8, "") + body.write_string(9, "") + body.write_struct_end() + + body.write_struct_begin(1) + body.write_string(0, dev.get("hdid", "")) + body.write_string(1, dev.get("app_version", "13.4.22")) + body.write_string(2, dev.get("sdk_version", "1.0.80138")) + body.write_string(3, "") + body.write_string(4, dev.get("ip", "127.0.0.1")) + body.write_string(5, dev.get("vendor", "android")) + body.write_string(6, "") + body.write_struct_end() + + body.write_struct_begin(2) + body.write_int8(0, 1) + body.write_string(1, dev.get("model", "")) + body.write_string(2, dev.get("fingerprint", "")) + body.write_string(3, dev.get("os", "android")) + body.write_string(4, dev.get("screen", "")) + body.write_string(6, dev.get("width", "1080")) + body.write_string(7, dev.get("height", "2120")) + body.write_string(8, dev.get("device_id", "")) + body.write_struct_end() + body.write_uint64(3, int(uid)) + body.write_string(4, cert_b64) + body.write_string(5, "") + body.write_string(6, "") + body.write_struct_end() + + req = TafOutputStream() + req.write_int32(0, int(session)) + wup = WupRequest() + wup.iTimeout = 0 + wup.setRequestId(int(session)) + wup.setServant("huyaudbwebui") + wup.setFunc("default") + wup.newdata["_wup_data"] = body.get_bytes() + wup.newdata["wupudbrequest_v0"] = req.get_bytes() + return cls(wup.encode()) + @classmethod def _load_from_path(cls, p: Path) -> Envelope: if p.suffix == ".json": diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py index 3512a14..0c897bb 100644 --- a/tests/test_huya_app_login.py +++ b/tests/test_huya_app_login.py @@ -26,6 +26,7 @@ from core.huya.login import HuyaLoginResult from core.huya.nonce_forge import K1_DEFAULT, gen_nonce from core.huya.udb_aes import udb_decrypt, udb_encrypt from core.huya.wup_encoder import _make_name, build_password_login_wup +from core.huya.wup_protocol import WupResponse from web.backend.database import Base from web.backend.models import HuyaAccount, User from web.backend.routers.huya import ( @@ -102,6 +103,38 @@ class TestHuyaAppLogin: assert b'"session":7654321' in current assert (b'"traceId":"' + new_trace.encode() + b'"') in current + def test_qr_envelope_is_fully_dynamic(self): + profile = { + "hdid": "h" * 32, + "app_version": "13.4.22", + "sdk_version": "1.0.80138", + "ip": "10.0.0.8", + "vendor": "vivo", + "model": "V2370A", + "fingerprint": "f" * 40, + "os": "android", + "screen": "V2370A,34,13", + "width": "1080", + "height": "2412", + "device_id": "d" * 40, + } + env = Envelope.build_qr( + uid=1199664135026, + cert_b64="A" * 260, + safedeviceid="B" * 180, + session=7654321, + trace_id="t" * 16 + "-12345-" + "9" * 20, + device_info=profile, + ) + wup = WupResponse() + wup.decode(bytes(env.raw)) + body = wup.newdata["_wup_data"] + for value in (b"B" * 180, b"h" * 32, b"f" * 40, b"V2370A", b"d" * 40): + assert value in body + assert b"PQwemAN9NHkZKoMq" not in body + assert b"02df398797432eadefcc12767119ad5e80999389" not in body + assert b"7c5387e0539c023c31c4ff0e807e7256117385ee" not in body + def test_device_profile_generation(self): p1 = generate_profile() assert p1["os"] == "android" @@ -271,10 +304,13 @@ class TestHuyaAppLogin: patch("core.huya.app_login.solve_safe_auth", return_value={"authId": "id"}), patch("core.huya.app_login.parse_real_uid", return_value=1199666914671), ): - cred, uid = login_cred_with_flow("300023887", "pw") + cred, uid, safedeviceid = login_cred_with_flow( + "300023887", "pw", include_device_token=True + ) assert cred == b"c" * 114 assert uid == 1199666914671 + assert safedeviceid == "A" * 180 assert calls == [assets, assets] def test_login_cred_flow_maps_invalid_password_response(self):