diff --git a/core/huya/app_login.py b/core/huya/app_login.py index 0ff0bba..fd953d5 100644 --- a/core/huya/app_login.py +++ b/core/huya/app_login.py @@ -1,11 +1,17 @@ """虎牙 App 渠道协议登录模块。 流程： - 1. 账号+密码 -> 独立设备画像 -> WUP 密码登录 (POST wup.huya.com) - 2. safe_auth 滑块自动过验 -> 提取 fresh cred 与 真实 uid - 3. 本地 XXTEA 算 nonce -> 铸造登录证书 (cert_forge) -> 补丁 WUP 信封 (envelope_forge) - 4. 模拟扫码绑定四步流 (getQrId -> scanQrPicNotify -> bindQrLoginUser -> tryQrLogin) 获取 biztoken - 5. POST /web/cookie/verify 兑换获取全套网页 Cookie + 1. 零设备注册链生成随机 dfpReport，获取新 safedeviceid/device_id + 2. 账号+密码+新注册字段 -> WUP 密码登录 (POST wup.huya.com) + 3. safe_auth 滑块自动过验 -> 提取 fresh cred 与真实 uid + 4. 本地 XXTEA 算 nonce -> 铸造登录证书 (cert_forge) -> 补丁 WUP 信封 (envelope_forge) + 5. 模拟扫码绑定四步流 (getQrId -> scanQrPicNotify -> bindQrLoginUser -> tryQrLogin) 获取 biztoken + 6. POST /web/cookie/verify 兑换获取全套网页 Cookie + +注册链不再重放旧 dfpReport 密文。登录帧的 32hex hdid 仍是服务端硬锚， +当前继续使用已注册样本；新注册链动态更新的是 safedeviceid 和 device_id。 +注册链（``core/huya/dfp_register``）每次登录前执行，失败即抛错终止（``HuyaAppLoginError``）， +不读取画像里的旧固定值，也不静默回退旧链。 """ from __future__ import annotations @@ -27,6 +33,7 @@ from .cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1 from .cookie_utils import normalize_huya_cookie from .device_fingerprint import get_huya_sdid from .device_profile import get_profile +from .dfp_register import DfpRegistrationError, register_device from .envelope_forge import Envelope from .login import HuyaCredentialError, HuyaLoginError, HuyaLoginResult from .nonce_forge import K1_DEFAULT, gen_nonce @@ -66,11 +73,6 @@ DEFAULT_GOLDEN_DEV = { "height": "2120", "device_id": "7c5387e0539c023c31c4ff0e807e7256117385ee", "hdid": "ed0db8334cadd236c00cadf7e11ab5a5", - "safedeviceid": ( - "PQwemAN9NHkZKoMqVTFUZBIypqMTaQEOrmXr37xQVhQZqrL/gUKEQ11xvE0ju48V8O/" - "t9UBGSp27m4+6bP4IiAEnpaR5Rj1kHEfN2SPLPqYZW9vroxUSoAvjJn6ezTP9jWGxxlRDCbt" - "Py4Rd6MencYT/pNImVIWK+YbNKZt1O05bHUFhqHf3" - ), } @@ -118,14 +120,29 @@ def wup_password_login_raw( hdid: str | None = None, proxies: dict | None = None, ) -> bytes: - """发送 WUP 密码登录，返回原始响应字节。""" + """发送 WUP 密码登录，返回原始响应字节。 + + 未显式传入 ``safedeviceid`` 时会先执行新设备注册链，不再回退旧的 + 固定 action/device_id。风控重试调用方应显式复用同一注册结果。 + 注册链失败抛 ``HuyaAppLoginError``，绝不静默回退旧固定值。 + """ uid_str = account[3:] if account.startswith("hy_") else account - mj, ua, sd = _golden_session_assets() - dev = device_info or DEFAULT_GOLDEN_DEV + 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, + ) + except DfpRegistrationError as exc: + raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc + dev["device_id"] = registered_device_id pkt = build_password_login_wup( uid_str, hashlib.sha1(password.encode()).hexdigest(), - safedeviceid or dev.get("safedeviceid") or sd, + safedeviceid, hdid or dev.get("hdid") or "ed0db8334cadd236c00cadf7e11ab5a5", mj["session"], mj["traceId"], @@ -238,9 +255,27 @@ def login_cred_with_flow( device_info: dict | None = None, proxies: dict | None = None, ) -> tuple[bytes, int]: - """账号密码 -> (新鲜cred, 真实uid)。自动过 safe_auth 滑块。""" + """新注册设备后登录，返回 ``(新鲜 cred, 真实 uid)``。 + + 注册只执行一次；safe_auth 通过后的重发继续使用同一组设备字段。 + """ + dev = dict(device_info or DEFAULT_GOLDEN_DEV) + try: + _t1, safedeviceid, registered_device_id = register_device( + fingerprint=dev.get("fingerprint"), + proxies=proxies, + ) + except DfpRegistrationError as exc: + raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc + dev["device_id"] = registered_device_id for rnd in range(max_rounds): - resp = wup_password_login_raw(account, password, device_info=device_info, proxies=proxies) + resp = wup_password_login_raw( + account, + password, + device_info=dev, + safedeviceid=safedeviceid, + proxies=proxies, + ) cred = parse_cred(resp) if cred: uid = parse_real_uid(resp) diff --git a/core/huya/device_profile.py b/core/huya/device_profile.py index 5617635..996083f 100644 --- a/core/huya/device_profile.py +++ b/core/huya/device_profile.py @@ -1,6 +1,11 @@ """虎牙多账号设备画像生成与管理。 为每个账号生成并持久化独立的设备身份画像（机型、屏幕、指纹、设备ID等）。 + +注意：画像只承载 *soft* 设备字段（机型/屏幕/随机指纹/随机 device_id）。 +``safedeviceid``（RSA action 令牌）与登录帧 ``device_id`` 由 +:mod:`core.huya.dfp_register` 在每次 WUP 登录前经新设备注册链实时签发， +不再写入画像，也不再允许任何固定金样本令牌被多账号复用。 """ from __future__ import annotations @@ -33,11 +38,6 @@ REAL_MODELS = [ HDID = "ed0db8334cadd236c00cadf7e11ab5a5" APP_VERSION = "13.4.22" SDK_VERSION = "1.0.80138" -SAFEDEVICEID_DEFAULT = ( - "PQwemAN9NHkZKoMqVTFUZBIypqMTaQEOrmXr37xQVhQZqrL/gUKEQ11xvE0ju48V8O/" - "t9UBGSp27m4+6bP4IiAEnpaR5Rj1kHEfN2SPLPqYZW9vroxUSoAvjJn6ezTP9jWGxxlRDCbt" - "Py4Rd6MencYT/pNImVIWK+YbNKZt1O05bHUFhqHf3" -) def _rand_sha1_hex() -> str: @@ -62,7 +62,8 @@ def generate_profile(model_pick=None) -> dict: "height": str(h), "device_id": _rand_sha1_hex(), "hdid": HDID, - "safedeviceid": SAFEDEVICEID_DEFAULT, + # 注: 不含 safedeviceid —— 每次登录前由 dfp_register 注册链签发, + # 画像不再持久化固定令牌 (见模块注释)。 } diff --git a/core/huya/dfp_register.py b/core/huya/dfp_register.py new file mode 100644 index 0000000..b1c98f5 --- /dev/null +++ b/core/huya/dfp_register.py @@ -0,0 +1,163 @@ +"""虎牙 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。 +""" +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) + diff --git a/tests/test_huya_app_login.py b/tests/test_huya_app_login.py index b824746..cbb7240 100644 --- a/tests/test_huya_app_login.py +++ b/tests/test_huya_app_login.py @@ -19,8 +19,14 @@ from core.huya import ( login_huya_app_password, login_huya_password, ) +from core.huya.app_login import ( + DEFAULT_GOLDEN_DEV, + 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_profile import generate_profile, get_profile +from core.huya.dfp_register import DfpRegistrationError from core.huya.envelope_forge import Envelope from core.huya.login import HuyaLoginResult from core.huya.nonce_forge import K1_DEFAULT, gen_nonce @@ -97,6 +103,9 @@ class TestHuyaAppLogin(unittest.TestCase): self.assertEqual(len(p1["fingerprint"]), 40) self.assertEqual(len(p1["device_id"]), 40) self.assertEqual(p1["hdid"], "ed0db8334cadd236c00cadf7e11ab5a5") + # 画像不再承载 safedeviceid：该令牌由 dfp_register 注册链每次登录前签发 + self.assertNotIn("safedeviceid", p1) + self.assertNotIn("safedeviceid", DEFAULT_GOLDEN_DEV) p2 = get_profile("test_user_account_123") p3 = get_profile("test_user_account_123") @@ -107,7 +116,8 @@ class TestHuyaAppLogin(unittest.TestCase): pkt = build_password_login_wup( uid_str="300023887", sha1_password="772ed992b0e161276f44ec63671e60155c506294", - safedeviceid=dev["safedeviceid"], + # 注册链签发格式: 180B base64 action 令牌 + safedeviceid="A" * 180, hdid=dev["hdid"], session=3251699, trace_id="test-trace-id", @@ -118,6 +128,60 @@ class TestHuyaAppLogin(unittest.TestCase): total_len = struct.unpack(">I", pkt[:4])[0] self.assertEqual(total_len, len(pkt)) + # ---- 新设备注册链 (core/huya/dfp_register) 生产接入测试 ---- + + def test_wup_login_skips_registration_when_safedeviceid_given(self): + """显式传入 safedeviceid 时不再触发注册链（风控重试复用同一注册结果）。""" + with patch("core.huya.app_login.register_device") as m_reg, \ + patch("core.huya.app_login.build_password_login_wup", return_value=b"pkt"), \ + patch("core.huya.app_login.requests.post", + return_value=MagicMock(status_code=200, content=b"")): + wup_password_login_raw("300023887", "pw", safedeviceid="A" * 180) + m_reg.assert_not_called() + + def test_wup_login_registers_fresh_device_when_safedeviceid_missing(self): + """未传 safedeviceid 时先注册，注册签发的 action/device_id 进 WUP 帧。""" + new_action = "B" * 180 + new_device_id = "c" * 40 + captured = {} + + def fake_build(*args, **kwargs): + captured["args"] = args + return b"pkt" + + with patch("core.huya.app_login.register_device", + return_value=("a" * 32, new_action, new_device_id)) as m_reg, \ + patch("core.huya.app_login.build_password_login_wup", side_effect=fake_build), \ + patch("core.huya.app_login.requests.post", + return_value=MagicMock(status_code=200, content=b"")): + wup_password_login_raw("300023887", "pw") + m_reg.assert_called_once() + # args: uid_str, sha1, safedeviceid, hdid, session, traceId, ua, dev + self.assertEqual(captured["args"][2], new_action) + self.assertEqual(captured["args"][7]["device_id"], new_device_id) + # 画像默认值里的旧 device_id 被注册结果覆盖，而非沿用 + self.assertNotEqual( + captured["args"][7]["device_id"], + DEFAULT_GOLDEN_DEV["device_id"], + ) + + def test_wup_login_registration_failure_is_explicit(self): + """注册失败必须抛错终止，禁止静默回退旧链（不发任何登录请求）。""" + with patch("core.huya.app_login.register_device", + side_effect=DfpRegistrationError("注册链超时")) as m_reg, \ + patch("core.huya.app_login.requests.post") as m_post: + with self.assertRaises(HuyaAppLoginError): + wup_password_login_raw("300023887", "pw") + m_reg.assert_called_once() + m_post.assert_not_called() + + def test_login_cred_flow_registration_failure_is_explicit(self): + """login_cred_with_flow 注册失败同样包装为 HuyaAppLoginError 显式失败。""" + with patch("core.huya.app_login.register_device", + side_effect=DfpRegistrationError("注册链 HTTP 500")): + with self.assertRaisesRegex(HuyaAppLoginError, "注册失败"): + login_cred_with_flow("300023887", "pw") + def test_router_functions(self): mock_res = HuyaLoginResult( success=True, diff --git a/tests/test_huya_dfp_register.py b/tests/test_huya_dfp_register.py new file mode 100644 index 0000000..88cd92c --- /dev/null +++ b/tests/test_huya_dfp_register.py @@ -0,0 +1,152 @@ +"""虎牙新设备注册链 (core/huya/dfp_register) 测试。 + +覆盖: +* 随机 dfpReport body/cw 结构 (4226B / 4146B / 固定10B尾); +* dfpReport 响应解析 (t1/t2/t5) 及缺字段失败; +* register_device 全流程: 三步 POST 顺序与 content-type、fingerprint 注入、 + 注册结果三元组返回; +* 失败语义: 模板缺失/HTTP 异常 -> DfpRegistrationError (不静默、不发登录帧)。 +""" +import base64 +import json +import unittest +from pathlib import Path +from unittest.mock import patch + +import requests + +from core.huya.dfp_register import ( + CHAIN_FILE, + CW_LEN, + DfpRegistrationError, + _build_random_dfp_body, + _parse_response, + register_device, +) + +TAIL = bytes.fromhex("3600400c0b8c980ca80c") + + +def _fake_resp() -> bytes: + """合成 dfpReport 响应: t1(32hex) + t2(180B base64 action) + t5(40hex)。""" + return (b"\x16\x20" + b"a" * 32 + + b"\x26\xb4" + b"A" * 180 + + b"\x56\x28" + b"b" * 40) + + +_GOLDEN_FP = b"02df398797432eadefcc12767119ad5e80999389" + + +def _dummy_chain() -> dict: + # selectOperator 请求体包含金样本 fingerprint, 供替换注入测试 + return { + "getDfpConfig": (b"cfg-req", b"cfg-resp"), + "selectOperator": (b"sel-hdr" + _GOLDEN_FP + b"sel-tail", b"sel-resp"), + "dfpReport": (b"dfp-req", b"dfp-resp"), + } + + +class TestRandomDfpBody(unittest.TestCase): + + def test_body_shape(self): + body = _build_random_dfp_body() + self.assertEqual(len(body), 4226) + cw = body[-CW_LEN:] + self.assertEqual(len(cw), CW_LEN) # 4146 + self.assertTrue(cw.endswith(TAIL), "cw 尾部应为固定 10B") + + def test_body_is_randomized(self): + b1 = _build_random_dfp_body() + b2 = _build_random_dfp_body() + self.assertNotEqual(b1, b2, "每次注册应生成不同的随机 cw") + + +class TestParseResponse(unittest.TestCase): + + def test_parse_ok(self): + t1, t2, t5 = _parse_response(_fake_resp()) + self.assertEqual(t1, "a" * 32) + self.assertEqual(t2, "A" * 180) + self.assertEqual(t5, "b" * 40) + + def test_parse_golden_evidence(self): + """真实抓包模板响应必须可解析 (证据: evidence/dfp_chain_golden.json)。""" + if not CHAIN_FILE.exists(): + self.skipTest("缺失 golden 注册链模板") + data = json.loads(CHAIN_FILE.read_text(encoding="utf-8")) + resp = base64.b64decode(data["dfpReport"]["resp_b64"]) + t1, t2, t5 = _parse_response(resp) + self.assertEqual(len(t1), 32) + self.assertEqual(len(t2), 180) + self.assertEqual(len(t5), 40) + + def test_parse_missing_fields_raises(self): + with self.assertRaises(DfpRegistrationError): + _parse_response(b"\x16\x20" + b"a" * 32) # 缺 t2/t5 + with self.assertRaises(DfpRegistrationError): + _parse_response(b"\x26\xb4" + b"A" * 180) # 缺 t1/t5 + with self.assertRaises(DfpRegistrationError): + _parse_response(b"\x56\x28" + b"b" * 40) # 缺 t1/t2 + + +class TestRegisterDevice(unittest.TestCase): + + def test_flow_three_steps_and_returns_triple(self): + calls = [] + + def spy_post(body, content_type="application/octet-stream", + timeout=20, proxies=None): + calls.append(content_type) + return _fake_resp() + + with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ + patch("core.huya.dfp_register._post", side_effect=spy_post): + t1, t2, t5 = register_device(fingerprint=None) + self.assertEqual((t1, t2, t5), ("a" * 32, "A" * 180, "b" * 40)) + # getDfpConfig -> selectOperator -> dfpReport 的 content-type 序列 + self.assertEqual( + calls, + ["application/octet-stream", "application/x-wup", "application/octet-stream"], + ) + + def test_select_operator_injects_fingerprint(self): + captured = [] + + def spy_post(body, content_type="application/octet-stream", + timeout=20, proxies=None): + captured.append((body, content_type)) + return _fake_resp() + + new_fp = b"0" * 20 + b"f" * 20 + with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ + patch("core.huya.dfp_register._post", side_effect=spy_post): + register_device(fingerprint=new_fp.decode("ascii")) + sel_body, sel_ct = captured[1] + self.assertEqual(sel_ct, "application/x-wup") + self.assertIn(new_fp, sel_body, "selectOperator 应注入当前账号画像 fingerprint") + self.assertNotIn(_GOLDEN_FP, sel_body) + # dfpReport 请求体为随机 cw 结构 (零设备生成) + dfp_body, dfp_ct = captured[2] + self.assertEqual(len(dfp_body), 4226) + self.assertEqual(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 self.assertRaises(DfpRegistrationError): + register_device() + m_post.assert_not_called() + + def test_http_failure_raises(self): + """HTTP 层异常由 _post 转换为 DfpRegistrationError 后向上传播 (不静默)。""" + with patch("core.huya.dfp_register._load_chain", return_value=_dummy_chain()), \ + patch("core.huya.dfp_register.requests.post", + side_effect=requests.RequestException("connection reset")): + with self.assertRaises(DfpRegistrationError): + register_device() + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tools/huya_device_register.py b/tools/huya_device_register.py index 7353ffb..0bd3cf7 100644 --- a/tools/huya_device_register.py +++ b/tools/huya_device_register.py @@ -2,27 +2,28 @@ 链路 (servant=huyaudbwebui, 均为 TAF/WUP 信封 POST): getDfpConfig -> selectOperator -> dfpReport - -> 响应下发 t1(32hex 设备指纹) / t2(base64 action=safedeviceid) / t5(40hex session hash) + -> 响应下发 t1(32hex appkey回声) / t2(base64 action=safedeviceid) / t5(40hex device_id) -实测结论 (2026-08-26): -- 三接口均可纯 Python 直连重放 (HTTP 200)。 +实测结论 (2026-08-26/27): +- 三接口均可纯 Python 直连 (HTTP 200)。 - dfpReport 响应 t5 == WUP 登录帧的 device_id; 响应 t2 == 登录帧的 safedeviceid —— 注册链响应就是登录帧设备字段的签发源。用注册链新签发的 t2/t5 登录 -> cred ✅。 -- t1(32hex) 不是登录帧 hdid (实测作 hdid 登录 -> APP_SIGN_NOT_MATCH)。 -- 设备身份绑定在 dfpReport 请求的"加密体"(10B 魔数 + 加密采集数据)上: - selectOperator 里改 fingerprint / 机型号不影响 t1/t5 -> 不重新注册设备。 - 加密体生成算法未破解 (逆向笔记 待破解#2), 因此目前只能"重放真机加密体", - 得到的设备身份永远等于抓包那台真机 —— 无法纯代码铸造全新设备。 -- hdid 由 libhydeviceid.so 生成本地设备ID, 经 native setDeviceInfo 报告, - 不走 HTTP (抓包不可见), 纯代码无法铸造新 hdid -> 多账号必须复用金样本 hdid。 +- t1(32hex) 不是设备锚: 恒=appkey 变体回声; selectOperator 全改(sd+fp+机型)不影响它。 +- ★(2026-08-27 突破) 服务端不校验 dfpReport 加密体(cw)内容: 随机 4146B cw 照样 + 200 + 签发新 t2/t5。t5(device_id) = 服务端从 cw 解出的 40hex hdid, 随机 cw 时 + 为随机值但服务端宽容。-> 无需真机加密体, python 零设备注册链达成 (--gen)。 +- 唯一不可铸造: WUP 登录帧 32hex hdid (libhydeviceid.so 本地生成 + native + setDeviceInfo 上报, 不走 HTTP) -> 多账号必须复用金样本 hdid (APP_SIGN_NOT_MATCH 硬错)。 用途: - 登录前重放注册链拿"新鲜 action"作为 safedeviceid (比固定金样本更像真机), + 登录前注册链拿"新鲜 action" + [重放|随机] dfpReport 作为 safedeviceid, 配合每账号随机 fingerprint/device_id/机型 (soft 字段) 使用。 用法: - python tools/huya_device_register.py # 走完整注册链并打印 t1/t2/t5 - python tools/huya_device_register.py --login # 注册链 -> 用新action登录出cred + python tools/huya_device_register.py # 重放注册链并打印 t1/t2/t5 + python tools/huya_device_register.py --gen # ★生成器随机cw注册链 (零设备) + python tools/huya_device_register.py --gen-login # ★零设备注册链->登录 + python tools/huya_device_register.py --login # 重放注册链 -> 用新action登录出cred """ from __future__ import annotations @@ -102,6 +103,11 @@ def dfp_report(chain) -> tuple[str, str, str]: r = _post(WSAPI, reqb) r.raise_for_status() b = r.content + return parse_dfp_response(b) + + +def parse_dfp_response(b: bytes) -> tuple[str, str, str]: + """解析 dfpReport 响应 -> (t1, t2, t5).""" t1 = re.search(rb"\x16\x20([0-9a-f]{32})", b).group(1).decode() m2 = re.search(rb"\x26\xb4([A-Za-z0-9+/=]{180})", b) t2 = m2.group(1).decode("latin1") @@ -109,6 +115,58 @@ def dfp_report(chain) -> tuple[str, str, str]: return t1, t2, t5 +def gen_dfp_report(chain) -> tuple[str, str, str]: + """dfpReport 用生成器随机 cw (零设备) -> (t1, t2, t5). + + ★中新突破 (2026-08-27): 服务端不校验 cw 内容, 随机 4146B cw 照样 + 签发新 action(t2)/device_id(t5), 与重放真机加密体等价, 但无需真机! + """ + sys.path.insert(0, str(HERE)) + from dfp_gen import build_cw, build_body, make_triple_json + _hdid, _did, _appkey, jp = make_triple_json() + body = build_body(build_cw(jp)) + r = _post(WSAPI, body) + r.raise_for_status() + return parse_dfp_response(r.content) + + +def make_select_operator(chain, fingerprint_40hex: str | None = None, + safedeviceid: str | None = None) -> bytes: + """构造 selectOperator: 可替换 fingerprint 与 180B safedeviceid(action).""" + reqb, _ = chain["selectOperator"] + # 替换 180B safedeviceid (字段头 \x56\xb4 后接 180B base64) + if safedeviceid and len(safedeviceid) == 180: + idx = reqb.find(b"\x56\xb4") + if idx >= 0: + reqb = reqb[: idx + 2] + safedeviceid.encode() + reqb[idx + 2 + 180:] + if fingerprint_40hex and len(fingerprint_40hex) == 40: + idx = reqb.find(b"02df398797432eadefcc12767119ad5e80999389") + if idx >= 0: + reqb = reqb[:idx] + fingerprint_40hex.encode() + reqb[idx + 40:] + return reqb + + +def gen_fresh_identity() -> tuple[str, str, str]: + """生成器全链路注册 (零设备) -> (t2 action, t5 device_id, t1 appkey回声).""" + chain = _load_chain() + get_dfp_config(chain) + # 用全新 fingerprint + 新 sd (先用金样本走一遍拿新 action, 再自定义 selectOperator) + so = make_select_operator(chain, fingerprint_40hex=None, safedeviceid=None) + _post(WSAPI, so, "application/x-wup") + t1, t2, t5 = gen_dfp_report(chain) + return t1, t2, t5 + + +def gen_login(account: str, password: str, + fingerprint: str | None = None) -> str: + """零设备注册链 -> 登录 (金样本 hdid), 返回 cred hex / NEED_RISK / REJECTED.""" + t1, t2, t5 = gen_fresh_identity() + print(f"注册链: t1(appkey回声)={t1}") + print(f" t2(action)={t2[:16]}... len={len(t2)}") + print(f" t5(device_id)={t5}") + return login_with_registered(t2, t5, account, password, fingerprint=fingerprint) + + def fresh_action() -> str: """跑一遍注册链返回新签发的 action (登录帧 safedeviceid).""" chain = _load_chain() @@ -156,7 +214,17 @@ def login_with_registered(action: str, t5_device_id: str, if __name__ == "__main__": - if "--login" in sys.argv: + if "--gen-login" in sys.argv: + acct = sys.argv[sys.argv.index("--gen-login") + 1] + pwd = sys.argv[sys.argv.index("--gen-login") + 2] + out = gen_login(acct, pwd) + print("登录结果:", ("cred " + out[:16] + "...") if out.startswith(("0a", "0b")) else out) + elif "--gen" in sys.argv: + t1, t2, t5 = gen_fresh_identity() + print(f"生成器注册链: t1={t1}") + print(f" t2(action)={t2[:16]}... len={len(t2)}") + print(f" t5(device_id)={t5}") + elif "--login" in sys.argv: acct = sys.argv[sys.argv.index("--login") + 1] pwd = sys.argv[sys.argv.index("--login") + 2] act = fresh_action()