173 lines
7.0 KiB
Python
173 lines
7.0 KiB
Python
"""虎牙 App 新设备注册链（零设备 dfpReport 生成 + 注册响应解析）。
|
||
|
||
每次登录前执行 ``getDfpConfig -> selectOperator -> dfpReport``：
|
||
|
||
* ``dfpReport`` 使用已验证的 4146 字节随机 ``cw``（零设备生成，不重放旧设备报文）；
|
||
* 服务端返回新的 ``safedeviceid``（t2）和 ``device_id``（t5）—— 这两个字段
|
||
就是 WUP 登录帧的设备字段签发源；
|
||
* 请求模板只承载 TAF/WUP 协议形状，旧的 dfpReport 密文不会被使用。
|
||
|
||
请求模板来自 ``evidence/dfp_chain_golden.json``，与现有项目的证书/信封样本
|
||
一样作为协议模板使用。模板中的设备字段会在 selectOperator 步骤按当前画像更新。
|
||
|
||
错误语义: 链上任何一步失败（模板缺失/损坏、HTTP 异常、响应缺字段）都会抛
|
||
``DfpRegistrationError``。调用方（core/huya/app_login）应把注册失败当作明确失败
|
||
终止登录；禁止静默回退到画像里的旧固定 safedeviceid/device_id。
|
||
|
||
⚠ 名称混淆提示（详见 ``docs/HUYA_APP_OVERVIEW.md`` §二 / R15, R39 定案）：
|
||
- 这里签发的 t2 = 登录帧 safedeviceid、t5 = 登录帧 device_id（40hex）；
|
||
- 登录帧 t1.t0 (32hex appSign) **已破解且与设备无关**: md5("5008_13.4.22_" + k1)
|
||
= ed0db8334cadd236c00cadf7e11ab5a5 (k1 为 datadiv 内嵌常量, 同版本所有真机同值),
|
||
由 core/huya/app_login.py 直接使用; 本模块响应里的 t1 (服务端回显) 在
|
||
app_login 中被丢弃 (_t1), 注册链结果不依赖它;
|
||
- ``_random_triple`` 的 64hex 占位只是 cw JSON 模板形态 — 真实 dfpReport 指纹
|
||
JSON (R36 解密 12 样本) 无 hdid 字段 (顶层 Athena/deviceinfo/terminal/version)。
|
||
"""
|
||
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 requests
|
||
|
||
WSAPI = "https://wsapi.huya.com"
|
||
UA = "okhttp/3.14.9"
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
CHAIN_FILE = ROOT / "evidence" / "dfp_chain_golden.json"
|
||
|
||
TAF_HEAD = bytes.fromhex(
|
||
"10032c3c4c56"
|
||
"0c687579617564627765627569"
|
||
"66"
|
||
"096466705265706f7274"
|
||
"7d00011056"
|
||
"0800010604"
|
||
"74526571"
|
||
"1d00011048"
|
||
"0a060016"
|
||
"07616e64726f6964"
|
||
"2d0001"
|
||
"1032"
|
||
)
|
||
MAGIC = bytes.fromhex("571882cf664bb39401ee")
|
||
CW_TAIL = bytes.fromhex("3600400c0b8c980ca80c")
|
||
CW_JSON_LEN = 586
|
||
CW_COLL_LEN = 3548
|
||
CW_LEN = 2 + CW_JSON_LEN + CW_COLL_LEN + len(CW_TAIL)
|
||
|
||
|
||
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 _post(body: bytes, content_type: str = "application/octet-stream",
|
||
timeout: float = 20, proxies: Mapping[str, str] | None = None) -> bytes:
|
||
try:
|
||
response = requests.post(
|
||
WSAPI,
|
||
data=body,
|
||
headers={"Content-Type": content_type, "User-Agent": UA,
|
||
"Accept-Encoding": "gzip"},
|
||
timeout=timeout,
|
||
proxies=proxies,
|
||
)
|
||
response.raise_for_status()
|
||
return response.content
|
||
except requests.RequestException as exc:
|
||
raise DfpRegistrationError(f"注册链 HTTP 请求失败: {exc}") from exc
|
||
|
||
|
||
def _random_triple() -> tuple[str, str, str]:
|
||
"""生成 dfp JSON 中的 40hex/40hex/64hex 三元组。"""
|
||
return (
|
||
hashlib.sha256(os.urandom(32) + b"hdid").hexdigest(),
|
||
hashlib.sha256(os.urandom(32) + b"devid").hexdigest(),
|
||
hashlib.sha256(os.urandom(32) + b"appkey").hexdigest(),
|
||
)
|
||
|
||
|
||
def _build_random_dfp_body() -> bytes:
|
||
"""构造服务端接受的随机 dfpReport 请求体。
|
||
|
||
cw 中 JSON 段明文仅承载真实请求的段长度与三元组形态，实证（2026-08-27）
|
||
服务端不校验 cw 内容，随机 cw 照样 200 + 新签发 t2/t5，因此三元组
|
||
（hdid/device_id/appkey，各 64hex sha256）只为凑齐模板形态，属诊断用途。
|
||
"""
|
||
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_sec = os.urandom(CW_JSON_LEN)
|
||
if len(seed) <= CW_JSON_LEN:
|
||
mask = os.urandom(len(seed))
|
||
json_sec = bytes(a ^ b for a, b in zip(seed, mask)) + json_sec[len(seed):]
|
||
cw = os.urandom(2) + json_sec + os.urandom(CW_COLL_LEN) + CW_TAIL
|
||
if len(cw) != CW_LEN:
|
||
raise DfpRegistrationError(f"dfpReport cw 长度异常: {len(cw)}")
|
||
body = struct.pack(">I", len(TAF_HEAD) + len(MAGIC) + len(cw) + 4) + TAF_HEAD + MAGIC + cw
|
||
if len(body) != 4226:
|
||
raise DfpRegistrationError(f"dfpReport body 长度异常: {len(body)}")
|
||
return body
|
||
|
||
|
||
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
|
||
|
||
|
||
def _parse_response(data: bytes) -> tuple[str, str, str]:
|
||
try:
|
||
t1 = re.search(rb"\x16\x20([0-9a-f]{32})", data).group(1).decode()
|
||
t2 = re.search(rb"\x26\xb4([A-Za-z0-9+/=]{180})", data).group(1).decode("latin1")
|
||
t5 = re.search(rb"\x56\x28([0-9a-f]{40})", data).group(1).decode()
|
||
except AttributeError as exc:
|
||
raise DfpRegistrationError("dfpReport 响应缺少 t1/t2/t5") from exc
|
||
return t1, t2, t5
|
||
|
||
|
||
def register_device(fingerprint: str | None = None,
|
||
proxies: Mapping[str, str] | None = None,
|
||
timeout: float = 20) -> tuple[str, str, str]:
|
||
"""执行新注册链，返回 ``(t1, safedeviceid, device_id)``。"""
|
||
chain = _load_chain()
|
||
_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)
|
||
return _parse_response(response)
|
||
|