diff --git a/core/huya/app_login.py b/core/huya/app_login.py index 6cdc9dd..2dbc562 100644 --- a/core/huya/app_login.py +++ b/core/huya/app_login.py @@ -15,7 +15,7 @@ ⚠ 风险与边界（详见 ``docs/HUYA_APP_OVERVIEW.md``）： - hdid 全账号固定共享 -> 服务端可设备维度关联账号； - - 滑块 UA / session / traceId 用金样本常量 -> 与随机机型画像不自洽； + - 滑块 UA / session / traceId / userAction 均按当前画像和本次请求动态生成； - qr_auth(扫码)/dx_auth(短信) 无自动闭环 -> 碰上直接失败（QR_AUTH_REQUIRED）。 """ @@ -29,7 +29,7 @@ import re import struct import time import uuid -from typing import Mapping +from collections.abc import Mapping from urllib.parse import parse_qs, quote, urlparse import requests @@ -38,10 +38,10 @@ from loguru import logger from .cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1 from .cookie_utils import normalize_huya_cookie from .device_fingerprint import account_state_dir, get_huya_sdid -from .device_profile import get_profile +from .device_profile import get_profile, mobile_user_agent from .dfp_register import DfpRegistrationError, register_device from .envelope_forge import Envelope -from .login import HuyaCredentialError, HuyaLoginError, HuyaLoginResult +from .login import HuyaLoginError, HuyaLoginResult from .nonce_forge import K1_DEFAULT, gen_nonce from .wup_encoder import build_password_login_wup @@ -91,30 +91,28 @@ class HuyaAppQrAuthRequiredError(HuyaAppLoginError): def _golden_session_assets() -> tuple[dict, str, str]: + """生成本次登录的会话元数据和行为轨迹。""" + from .wup_encoder import make_trace_id, make_user_action + mj = { "associationId": 8193, "funcName": "hypasswordLogin", "group": 1, "id": 4097, - "session": 3251699, + "session": random.randint(1_000_000, 9_999_999), "step": 1, "stillLogin": False, - "traceId": "0b8f098ff64a5bdc-23473-94783833787595172451", + "traceId": make_trace_id(), "type": 3, "uid": 0, "userContext": "", } - ua = ( - '{"curl":"登录页","furl":"我的","latitude":"-1.0","longitude":"-1.0","ssid":"",' - '"user_action":[{"id":"24","time":"1787595171950","x":"277","y":"1057"},' - '{"id":"11","time":"1787595172444","x":"296","y":"952"}]}' - ) - sd = ( - "PQwemAN9NHkZKoMqVTFUZBIypqMTaQEOrmXr37xQVhQZqrL/gUKEQ11xvE0ju48V8O/" - "t9UBGSp27m4+6bP4IiAEnpaR5Rj1kHEfN2SPLPqYZW9vroxUSoAvjJn6ezTP9jWGxxlRDCbt" - "Py4Rd6MencYT/pNImVIWK+YbNKZt1O05bHUFhqHf3" - ) - return mj, ua, sd + return mj, make_user_action(), "" + + +def _android_version(device_info: Mapping[str, object]) -> str: + parts = str(device_info.get("screen") or "").split(",") + return parts[2] if len(parts) >= 3 and parts[2] else "11" def wup_password_login_raw( @@ -133,14 +131,15 @@ def wup_password_login_raw( 注册链失败抛 ``HuyaAppLoginError``，绝不静默回退旧固定值。 """ uid_str = account[3:] if account.startswith("hy_") else account + dev = dict(device_info) if device_info is not None else get_profile(account) mj, ua, _old_sd = _golden_session_assets() - dev = dict(device_info or DEFAULT_GOLDEN_DEV) if not safedeviceid: try: _t1, safedeviceid, registered_device_id = register_device( fingerprint=dev.get("fingerprint"), proxies=proxies, timeout=timeout, + device_info=dev, ) except DfpRegistrationError as exc: raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc @@ -160,7 +159,7 @@ def wup_password_login_raw( data=pkt, headers={ "Content-Type": "application/multipart-formdata; charset=UTF-8", - "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 11)", + "User-Agent": f"Dalvik/2.1.0 (Linux; U; Android {_android_version(dev)})", "Accept-Encoding": "gzip", }, proxies=proxies, @@ -203,7 +202,12 @@ def parse_risk_url(resp: bytes) -> str | None: return (pt or urls)[0] -def solve_safe_auth(risk_url: str, proxies=None, max_retry: int = 3) -> dict: +def solve_safe_auth( + risk_url: str, + proxies=None, + max_retry: int = 3, + device_info: Mapping[str, object] | None = None, +) -> dict: """解 safe_auth 风控滑块。""" from .verification.solver import ( HuyaQrAuthRequiredError, @@ -218,7 +222,7 @@ def solve_safe_auth(risk_url: str, proxies=None, max_retry: int = 3) -> dict: last_err: Exception | None = None for attempt in range(max_retry): solver = HuyaVerificationSolver( - ua=APP_UA_MOBILE, + ua=mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV), proxies=proxies, app_id=app_id, page_url=risk_url, @@ -268,11 +272,12 @@ def login_cred_with_flow( 注册只执行一次；safe_auth 通过后的重发继续使用同一组设备字段。 """ - dev = dict(device_info or DEFAULT_GOLDEN_DEV) + dev = dict(device_info) if device_info is not None else get_profile(account) try: _t1, safedeviceid, registered_device_id = register_device( fingerprint=dev.get("fingerprint"), proxies=proxies, + device_info=dev, ) except DfpRegistrationError as exc: raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc @@ -301,7 +306,7 @@ def login_cred_with_flow( raise HuyaAppQrAuthRequiredError( f"该账号 App 渠道要求扫码验证(qr_auth)，请先在手机虎牙 App 上正常登录一次建立设备信任。" ) - solve_safe_auth(risk_url, proxies=proxies) + solve_safe_auth(risk_url, proxies=proxies, device_info=dev) logger.info("[huya-app] safe_auth 滑块过验成功，重发 WUP 登录...") continue raise HuyaAppLoginError("登录未返回凭据也无风控URL（密码错误或账号状态异常）") @@ -311,7 +316,13 @@ def login_cred_with_flow( class QrRole: """udblgn.huya.com/qrLgn/* 的 JSON 协议封装。""" - def __init__(self, pc: bool, sdid: str, proxies: dict | None = None): + def __init__( + self, + pc: bool, + sdid: str, + proxies: dict | None = None, + device_info: Mapping[str, object] | None = None, + ): self.pc = pc self.sdid = sdid self.s = requests.Session() @@ -327,7 +338,9 @@ class QrRole: self.req_counter = random.randint(40_000_000, 41_000_000) self.s.headers.update( { - "User-Agent": UA_PC if pc else APP_UA_MOBILE, + "User-Agent": UA_PC + if pc + else mobile_user_agent(device_info or DEFAULT_GOLDEN_DEV), "Origin": UDB_BASE, "content-type": "application/json;charset=UTF-8", "Accept": "*/*", @@ -483,7 +496,12 @@ class HuyaAppPasswordLogin: ) sdid = sdid_obj.sdid if sdid_obj else "" pc = QrRole(pc=True, sdid=sdid, proxies=self.proxies) - ph = QrRole(pc=False, sdid=sdid, proxies=self.proxies) + ph = QrRole( + pc=False, + sdid=sdid, + proxies=self.proxies, + device_info=self.device_info, + ) beh, page = web_behavior() # 4.1 获取 qrId diff --git a/core/huya/device_fingerprint.py b/core/huya/device_fingerprint.py index cce1155..1925de1 100644 --- a/core/huya/device_fingerprint.py +++ b/core/huya/device_fingerprint.py @@ -6,10 +6,10 @@ from __future__ import annotations +import json import shutil import subprocess import tempfile -import json from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path @@ -17,6 +17,8 @@ from pathlib import Path import requests from loguru import logger +from .device_profile import mobile_user_agent + FINGERPRINT_DIR = Path(__file__).parent / "fingerprint" RUNNER_JS = FINGERPRINT_DIR / "runner.js" SDID_PREFIX = "__SDID__" @@ -75,6 +77,7 @@ def _run_node_runner( capture_output=True, text=True, timeout=total_timeout, + check=False, cwd=str(FINGERPRINT_DIR), ) except subprocess.TimeoutExpired as exc: @@ -123,12 +126,7 @@ def write_device_hint(state_dir: str | Path, device_info: Mapping) -> Path: screen = str(device_info.get("screen", "")) parts = screen.split(",") android_ver = parts[2] if len(parts) >= 3 else "11" - ua = ( - f"Mozilla/5.0 (Linux; Android {android_ver}; {device_info.get('model', '')} " - "Build/RKQ1.200826.002; wv) AppleWebKit/537.36 (KHTML, like Gecko) " - "Version/4.0 Chrome/149.0.7827.159 Mobile Safari/537.36 " - f"huya adr/13.4.22/{device_info.get('vendor', 'android')}/30" - ) + ua = mobile_user_agent(device_info) hint = { "model": device_info.get("model", ""), "vendor": device_info.get("vendor", "android"), diff --git a/core/huya/device_profile.py b/core/huya/device_profile.py index f1a5664..8b026e2 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 +from collections.abc import Mapping from pathlib import Path ROOT = Path(__file__).resolve().parent.parent.parent @@ -48,6 +49,23 @@ APP_VERSION = "13.4.22" SDK_VERSION = "1.0.80138" +def mobile_user_agent(device_info: Mapping[str, object]) -> str: + """根据统一设备画像生成 App WebView UA。""" + screen = str(device_info.get("screen") or "") + parts = screen.split(",") + android_ver = parts[2] if len(parts) >= 3 and parts[2] else "11" + api_level = parts[1] if len(parts) >= 2 and parts[1] else "30" + model = str(device_info.get("model") or "M2102J2SC") + vendor = str(device_info.get("vendor") or "android") + app_version = str(device_info.get("app_version") or APP_VERSION) + return ( + f"Mozilla/5.0 (Linux; Android {android_ver}; {model} " + "Build/RKQ1.200826.002; wv) AppleWebKit/537.36 (KHTML, like Gecko) " + f"Version/4.0 Chrome/149.0.7827.159 Mobile Safari/537.36 huya adr/" + f"{app_version}/{vendor}/{api_level}" + ) + + def _rand_sha1_hex() -> str: return hashlib.sha1(os.urandom(20)).hexdigest() diff --git a/core/huya/dfp_register.py b/core/huya/dfp_register.py index 15a2405..53fc233 100644 --- a/core/huya/dfp_register.py +++ b/core/huya/dfp_register.py @@ -7,10 +7,10 @@ 就是 WUP 登录帧的设备字段签发源； * 请求模板只承载 TAF/WUP 协议形状，旧的 dfpReport 密文不会被使用。 -请求模板来自 ``evidence/dfp_chain_golden.json``，与现有项目的证书/信封样本 -一样作为协议模板使用。模板中的设备字段会在 selectOperator 步骤按当前画像更新。 +请求按已验证的 TAF/WUP 字段布局动态编码；不依赖抓包证据文件。selectOperator +中的 RSA 字段使用协议固定格式的密文种子，设备画像字段在每次请求时写入。 -错误语义: 链上任何一步失败（模板缺失/损坏、HTTP 异常、响应缺字段）都会抛 +错误语义: 链上任何一步失败（请求编码、HTTP 异常、响应缺字段）都会抛 ``DfpRegistrationError``。调用方（core/huya/app_login）应把注册失败当作明确失败 终止登录；禁止静默回退到画像里的旧固定 safedeviceid/device_id。 @@ -26,23 +26,25 @@ from __future__ import annotations -import base64 import hashlib -import json import os import re -import socket -import ssl import struct -from pathlib import Path -from typing import Mapping +import time +from collections.abc import Mapping import requests +from .taf_protocol import TafOutputStream +from .wup_protocol import WupRequest + WSAPI = "https://wsapi.huya.com" UA = "okhttp/3.14.9" -ROOT = Path(__file__).resolve().parents[2] -CHAIN_FILE = ROOT / "evidence" / "dfp_chain_golden.json" +SELECT_OPERATOR_SDID = ( + "PQwemAN9NHkZKoMqVeJmXhIypqMTaQEOrmXr37xQVhQZqrPxjEKEQ11xvE02qawcys/" + "iwl5gSIa6nYz0L5tNl0VdpZFyNjNHCTnQtzCwD5RHKKOL+DYohiPvAFPblxznlGyB63NAY74u" + "2q9Niej6bLDarc9xAKbAuLzRb7hHIGxCTBp5uGf3" +) TAF_HEAD = bytes.fromhex( "10032c3c4c56" @@ -69,20 +71,86 @@ class DfpRegistrationError(RuntimeError): """新设备注册链失败。""" -def _load_chain() -> dict[str, tuple[bytes, bytes]]: - if not CHAIN_FILE.exists(): - raise DfpRegistrationError(f"注册链模板不存在: {CHAIN_FILE}") - try: - data = json.loads(CHAIN_FILE.read_text(encoding="utf-8")) - return { - name: ( - base64.b64decode(item["req_b64"]), - base64.b64decode(item["resp_b64"]), - ) - for name, item in data.items() - } - except (OSError, ValueError, KeyError) as exc: - raise DfpRegistrationError(f"注册链模板读取失败: {exc}") from exc +def _build_get_dfp_config_request() -> bytes: + """编码 getDfpConfig 请求（67 字节 TAF/WUP）。""" + wup = WupRequest() + wup.setServant("huyaudbwebui") + wup.setFunc("getDfpConfig") + wup.iTimeout = 0 + wup.writeStruct("tReq", {"app_id": "5008"}) + return wup.encode() + + +def _build_select_operator_request( + fingerprint: str | None = None, + device_info: Mapping[str, str] | None = None, +) -> bytes: + """按 selectOperator 的四段结构动态编码请求。""" + dev = { + "app_version": "13.4.22", + "model": "M2102J2SC", + "fingerprint": fingerprint or "02df398797432eadefcc12767119ad5e80999389", + "screen": "M2102J2SC,30,11", + } + if device_info: + for key in dev: + if device_info.get(key): + dev[key] = str(device_info[key]) + + 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, f"hyudb_{int(time.time() * 1000)}") + body.write_string(3, "5008") + body.write_int8(4, 0) + body.write_string(5, SELECT_OPERATOR_SDID) + for tag in range(6, 10): + body.write_string(tag, "") + body.write_struct_end() + + body.write_struct_begin(1) + body.write_string(0, "") + body.write_string(1, dev["app_version"]) + body.write_string(2, "2.0.8") + for tag in range(3, 7): + body.write_string(tag, "") + body.write_struct_end() + + body.write_struct_begin(2) + body.write_int8(0, 1) + body.write_string(1, dev["model"]) + body.write_string(2, dev["fingerprint"]) + body.write_string(3, "android") + body.write_string(4, dev["screen"]) + body.write_string(6, "") + body.write_string(7, "") + body.write_struct_end() + body.write_list(3, [0, 1, 3]) + body.write_struct_end() + + wup = WupRequest() + wup.setServant("huyaudbwebui") + wup.setFunc("selectOperator") + wup.iTimeout = 0 + wup.newdata["_wup_data"] = body.get_bytes() + return wup.encode() + + +def _load_chain( + fingerprint: str | None = None, + device_info: Mapping[str, str] | None = None, +) -> dict[str, tuple[bytes, bytes]]: + """生成注册链请求，保留 tuple 形状兼容旧调用方和测试。""" + return { + "getDfpConfig": (_build_get_dfp_config_request(), b""), + "selectOperator": ( + _build_select_operator_request(fingerprint, device_info), + b"", + ), + "dfpReport": (b"", b""), + } def _post( @@ -118,26 +186,21 @@ def _random_triple() -> tuple[str, str, str]: ) -def _build_random_dfp_body() -> bytes: +def _build_random_dfp_body( + device_info: Mapping[str, str] | None = None, +) -> bytes: """构造服务端接受的随机 dfpReport 请求体。 cw 中 JSON 段明文仅承载真实请求的段长度与三元组形态，实证（2026-08-27） 服务端不校验 cw 内容，随机 cw 照样 200 + 新签发 t2/t5，因此三元组 - （hdid/device_id/appkey，各 64hex sha256）只为凑齐模板形态，属诊断用途。 + （hdid/device_id/appkey，各 64hex sha256）只为凑齐模板形态，属诊断用途； + device_id、hdid 在画像存在时分别使用画像 device_id、guid32。 """ - hdid, device_id, appkey = _random_triple() - json_plain = ( - '{"appId":"5008","appVer":"13.4.22","appkey":"%s",' - '"channel":"xiaomi","deviceId":"%s",' - '"deviceName":"M2102J2SC","hdid":"%s",' - '"heightPixels":"2120","isCloud":0,"isForbidLog":1,"isHome":0,' - '"isPre":0,"openAppId":"","savePath":"/data/user/0/com.duowan.kiwi/files/huyaudb",' - '"sdkVer":"1.0.80138","servantName":"huyaudbwebui",' - '"shareAppDataPath":"/data/user/0/com.duowan.kiwi/files/huyaudb",' - '"systemInfo":"android","systemVer":"M2102J2SC,30,11",' - '"terminalType":1,"testEnv":0,"widthPixels":"1080"}' - ) % (appkey, device_id, hdid) - seed = json_plain.encode("utf-8") + json_plain = _build_dfp_json_plain(device_info) + seed = json_plain + if len(seed) > CW_JSON_LEN: + # 保持已验证的固定 cw 段长度，同时让完整画像参与本次随机报文。 + seed = hashlib.sha256(seed).digest() json_sec = os.urandom(CW_JSON_LEN) if len(seed) <= CW_JSON_LEN: mask = os.urandom(len(seed)) @@ -156,6 +219,33 @@ def _build_random_dfp_body() -> bytes: return body +def _build_dfp_json_plain(device_info: Mapping[str, str] | None = None) -> bytes: + """生成 dfpReport 加密段使用的画像 JSON 明文。""" + hdid, device_id, appkey = _random_triple() + dev = device_info or {} + app_version = str(dev.get("app_version") or "13.4.22") + vendor = str(dev.get("vendor") or "xiaomi") + model = str(dev.get("model") or "M2102J2SC") + screen = str(dev.get("screen") or "M2102J2SC,30,11") + width = str(dev.get("width") or "1080") + height = str(dev.get("height") or "2120") + # 注册前的 deviceId/guid 使用账号画像；缺失时才生成一次性形态值。 + device_id = str(dev.get("device_id") or device_id) + hdid = str(dev.get("guid32") or hdid) + json_plain = ( + f'{{"appId":"5008","appVer":"{app_version}","appkey":"{appkey}",' + f'"channel":"{vendor}","deviceId":"{device_id}",' + f'"deviceName":"{model}","hdid":"{hdid}",' + f'"heightPixels":"{height}","isCloud":0,"isForbidLog":1,"isHome":0,' + '"isPre":0,"openAppId":"","savePath":"/data/user/0/com.duowan.kiwi/files/huyaudb",' + '"sdkVer":"1.0.80138","servantName":"huyaudbwebui",' + '"shareAppDataPath":"/data/user/0/com.duowan.kiwi/files/huyaudb",' + f'"systemInfo":"android","systemVer":"{screen}",' + f'"terminalType":1,"testEnv":0,"widthPixels":"{width}"}}' + ) + return json_plain.encode("utf-8") + + def _select_operator_request(template: bytes, fingerprint: str | None) -> bytes: if fingerprint and len(fingerprint) == 40: old = b"02df398797432eadefcc12767119ad5e80999389" @@ -188,11 +278,14 @@ def register_device( fingerprint: str | None = None, proxies: Mapping[str, str] | None = None, timeout: float = 20, + device_info: Mapping[str, str] | None = None, ) -> tuple[str, str, str]: """执行新注册链，返回 ``(t1, safedeviceid, device_id)``。""" - chain = _load_chain() + chain = _load_chain(fingerprint=fingerprint, device_info=device_info) _post(chain["getDfpConfig"][0], timeout=timeout, proxies=proxies) select_request = _select_operator_request(chain["selectOperator"][0], fingerprint) _post(select_request, "application/x-wup", timeout, proxies) - response = _post(_build_random_dfp_body(), timeout=timeout, proxies=proxies) + response = _post( + _build_random_dfp_body(device_info), timeout=timeout, proxies=proxies + ) return _parse_response(response) diff --git a/docs/HUYA_APP_OVERVIEW.md b/docs/HUYA_APP_OVERVIEW.md index 68e02a2..23af751 100644 --- a/docs/HUYA_APP_OVERVIEW.md +++ b/docs/HUYA_APP_OVERVIEW.md @@ -21,20 +21,24 @@ Web UI 通过 `web/backend/routers/huya.py` 调用它。 ``` ┌──────────────────────────────────────────────────────────────────────┐ │ 每次 WUP 登录前 │ -│ ① 新设备注册链 core/huya/dfp_register.register_device() │ +│ ① 读取/创建账号设备画像 core/huya/device_profile.get_profile() │ +│ 同一账号复用 vendor/model/screen/width/height/fingerprint/guid32 │ +│ │ +│ ② 新设备注册链 core/huya/dfp_register.register_device() │ │ getDfpConfig → selectOperator → dfpReport(随机cw) │ -│ 返回 (t1, safedeviceid, device_id) ← 全新，非画像旧值 │ +│ 三步请求的设备字段均来自①，返回 (t1, safedeviceid, device_id) │ +│ 返回字段为本次登录专用的新令牌，非画像旧值 │ │ 失败 → 抛 DfpRegistrationError → HuyaAppLoginError，终止，不回退 │ │ │ -│ ② WUP 密码登录 wup_password_login_raw() │ +│ ③ WUP 密码登录 wup_password_login_raw() │ │ POST wup.huya.com hypasswordLogin（safedeviceid/device_id 用①的）│ │ ├─ 正常 → 解出 114B cred + 真实 uid │ │ └─ pt_auth 滑块 → solve_safe_auth() 自动过验 → 重发登录（≤3轮） │ │ │ -│ ③ 本地铸证：nonce_forge(uid+k1) → cert_forge 铸证书 → envelope_forge │ +│ ④ 本地铸证：nonce_forge(uid+k1) → cert_forge 铸证书 → envelope_forge │ │ 补丁 WUP 信封（证书 + UID） │ -│ ④ 模拟扫码绑定四步流 QrRole（getQrId→scan→bind→tryQrLogin）→ biztoken│ -│ ⑤ POST /web/cookie/verify 兑换 → 全套网页 Cookie │ +│ ⑤ 模拟扫码绑定四步流 QrRole（getQrId→scan→bind→tryQrLogin）→ biztoken│ +│ ⑥ POST /web/cookie/verify 兑换 → 全套网页 Cookie │ └──────────────────────────────────────────────────────────────────────┘ ``` @@ -91,8 +95,8 @@ Web UI 通过 `web/backend/routers/huya.py` 调用它。 | fingerprint | **随机**（每账号） | `generate_profile` sha1 | 低 | | 机型 / 屏幕 / 宽高 | **随机**（每账号从 `REAL_MODELS` 选） | `generate_profile` | 低 | | app_version / sdk_version | **固定** | 常量 `13.4.22` / `1.0.80138` | 低（需与 hdid 配套） | -| 滑块 UA | **固定** 小米 M2102J2SC | `app_login.APP_UA_MOBILE` | ⚠ 与随机机型画像不一致 | -| session / traceId / 登录行为 JSON | **固定金样本** | `_golden_session_assets()` | ⚠ 多账号相同，异常画像 | +| 滑块/扫码 WebView UA | **按画像派生** | `device_profile.mobile_user_agent()` | 与机型/厂商/系统版本一致 | +| session / traceId / 登录行为 JSON | **每次请求动态** | `wup_encoder.make_*` + `_golden_session_assets()` | 仅协议字段固定结构 | | QrRole context / reqid / page_id | **随机** | `QrRole`（uuid/random） | 低 | --- @@ -100,10 +104,9 @@ Web UI 通过 `web/backend/routers/huya.py` 调用它。 ## 五、风险与未解决项（诚实清单） 1. **hdid 固定共享** → 服务端可在设备维度把多个账号关联起来（见 §2 #1）。纯代码无法铸造新 hdid；换掉即 `APP_SIGN_NOT_MATCH`。 -2. **滑块 UA / session / traceId 固定** → 与随机机型画像不自洽，长期易形成异常画像。 -3. **qr_auth / dx_auth 无自动闭环** → 碰上直接失败（`QR_AUTH_REQUIRED`）。 -4. **无稳定性数据** → 没有 20/50/100 次连续成功率、多账号/IP/时段矩阵。 -5. **Frida attach 主进程必被 msaoaidsec 杀**（真机/模拟器均实测）→ 见 §六。 +2. **qr_auth / dx_auth 无自动闭环** → 碰上直接失败（`QR_AUTH_REQUIRED`）。 +3. **无稳定性数据** → 没有 20/50/100 次连续成功率、多账号/IP/时段矩阵。 +4. **Frida attach 主进程必被 msaoaidsec 杀**（真机/模拟器均实测）→ 见 §六。 --- diff --git a/tests/test_huya_dfp_register.py b/tests/test_huya_dfp_register.py index 0c60bc8..fdf1caa 100644 --- a/tests/test_huya_dfp_register.py +++ b/tests/test_huya_dfp_register.py @@ -8,23 +8,24 @@ * 失败语义: 模板缺失/HTTP 异常 -> DfpRegistrationError (不静默、不发登录帧)。 """ -import pytest -import base64 -import json -import pytest -from pathlib import Path +import hashlib from unittest.mock import patch +import pytest import requests from core.huya.dfp_register import ( - CHAIN_FILE, CW_LEN, + SELECT_OPERATOR_SDID, DfpRegistrationError, + _build_dfp_json_plain, + _build_get_dfp_config_request, _build_random_dfp_body, + _build_select_operator_request, _parse_response, register_device, ) +from core.huya.wup_protocol import WupResponse TAIL = bytes.fromhex("3600400c0b8c980ca80c") @@ -59,6 +60,40 @@ class TestRandomDfpBody: b2 = _build_random_dfp_body() assert b1 != b2, "每次注册应生成不同的随机 cw" + def test_body_uses_device_profile_fields(self): + profile = { + "app_version": "13.4.22", + "vendor": "vivo", + "model": "V2370A", + "screen": "V2370A,34,13", + "width": "1080", + "height": "2412", + "device_id": "d" * 64, + "guid32": "g" * 64, + } + with ( + patch( + "core.huya.dfp_register._random_triple", + return_value=("h" * 64, "r" * 64, "k" * 64), + ), + ): + plaintext = _build_dfp_json_plain(profile) + assert b'"channel":"vivo"' in plaintext + assert b'"deviceName":"V2370A"' in plaintext + assert b'"deviceId":"' + b"d" * 64 in plaintext + assert b'"systemVer":"V2370A,34,13"' in plaintext + with ( + patch( + "core.huya.dfp_register._random_triple", + return_value=("h" * 64, "r" * 64, "k" * 64), + ), + patch( + "core.huya.dfp_register.os.urandom", side_effect=lambda n: b"\0" * n + ), + ): + body = _build_random_dfp_body(profile) + assert hashlib.sha256(plaintext).digest() in body[-CW_LEN + 2 :] + class TestParseResponse: def test_parse_ok(self): @@ -67,17 +102,6 @@ class TestParseResponse: assert t2 == "A" * 180 assert t5 == "b" * 40 - def test_parse_golden_evidence(self): - """真实抓包模板响应必须可解析 (证据: evidence/dfp_chain_golden.json)。""" - if not CHAIN_FILE.exists(): - pytest.skip("缺失 golden 注册链模板") - data = json.loads(CHAIN_FILE.read_text(encoding="utf-8")) - resp = base64.b64decode(data["dfpReport"]["resp_b64"]) - t1, t2, t5 = _parse_response(resp) - assert len(t1) == 32 - assert len(t2) == 180 - assert len(t5) == 40 - def test_parse_missing_fields_raises(self): with pytest.raises(DfpRegistrationError): _parse_response(b"\x16\x20" + b"a" * 32) # 缺 t2/t5 @@ -88,6 +112,27 @@ class TestParseResponse: class TestRegisterDevice: + def test_get_dfp_config_request_is_generated(self): + req = _build_get_dfp_config_request() + assert len(req) == 67 + wup = WupResponse() + wup.decode(req) + assert wup.sServantName == "huyaudbwebui" + assert wup.sFuncName == "getDfpConfig" + assert wup.readStruct("tReq") == bytes.fromhex("0a0604353030380b") + + def test_select_operator_request_is_generated(self): + fp = "0" * 40 + req = _build_select_operator_request(fp) + assert len(req) == 419 + assert fp.encode() in req + assert SELECT_OPERATOR_SDID.encode() in req + wup = WupResponse() + wup.decode(req) + assert wup.sServantName == "huyaudbwebui" + assert wup.sFuncName == "selectOperator" + assert "_wup_data" in wup.newdata + def test_flow_three_steps_and_returns_triple(self): calls = [] @@ -134,19 +179,6 @@ class TestRegisterDevice: assert len(dfp_body) == 4226 assert dfp_ct == "application/octet-stream" - def test_chain_template_missing_raises(self): - """模板缺失 -> 注册链直接报错, 不应发起任何网络请求。""" - with ( - patch( - "core.huya.dfp_register.CHAIN_FILE", - Path("/tmp/definitely_missing_chain.json"), - ), - patch("core.huya.dfp_register._post") as m_post, - ): - with pytest.raises(DfpRegistrationError): - register_device() - m_post.assert_not_called() - def test_http_failure_raises(self): """HTTP 层异常由 _post 转换为 DfpRegistrationError 后向上传播 (不静默)。""" with ( @@ -155,6 +187,6 @@ class TestRegisterDevice: "core.huya.dfp_register.requests.post", side_effect=requests.RequestException("connection reset"), ), + pytest.raises(DfpRegistrationError), ): - with pytest.raises(DfpRegistrationError): - register_device() + register_device()