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

统一虎牙 App 登录设备画像

Browse Source
This commit is contained in:
yml2213
2026-08-30 23:04:28 +08:00
parent bc02018719
commit f4ed518b50
6 changed files with 281 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/app_login.py
+44 -26
View File
@@ -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
core/huya/device_fingerprint.py
+5 -7
View File
@@ -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"),
core/huya/device_profile.py
+18
View File
@@ -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()
core/huya/dfp_register.py
+133 -40
View File
@@ -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"))
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 {
name: (
base64.b64decode(item["req_b64"]),
base64.b64decode(item["resp_b64"]),
)
for name, item in data.items()
"getDfpConfig": (_build_get_dfp_config_request(), b""),
"selectOperator": (
_build_select_operator_request(fingerprint, device_info),
b"",
),
"dfpReport": (b"", b""),
}
except (OSError, ValueError, KeyError) as exc:
raise DfpRegistrationError(f"注册链模板读取失败: {exc}") from exc
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)
docs/HUYA_APP_OVERVIEW.md
+15 -12
View File
@@ -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 杀**（真机/模拟器均实测）→ 见 §六。
---
tests/test_huya_dfp_register.py
+63 -31
View File
@@ -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()