160 lines
5.5 KiB
Python
160 lines
5.5 KiB
Python
"""虎牙新设备注册链 (core/huya/dfp_register) 测试。
|
|
|
|
覆盖:
|
|
* 随机 dfpReport body/cw 结构 (4226B / 4146B / 固定10B尾);
|
|
* dfpReport 响应解析 (t1/t2/t5) 及缺字段失败;
|
|
* register_device 全流程: 三步 POST 顺序与 content-type、fingerprint 注入、
|
|
注册结果三元组返回;
|
|
* 失败语义: 模板缺失/HTTP 异常 -> DfpRegistrationError (不静默、不发登录帧)。
|
|
"""
|
|
|
|
import pytest
|
|
import base64
|
|
import json
|
|
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:
|
|
def test_body_shape(self):
|
|
body = _build_random_dfp_body()
|
|
assert len(body) == 4226
|
|
cw = body[-CW_LEN:]
|
|
assert len(cw) == CW_LEN # 4146
|
|
assert cw.endswith(TAIL)
|
|
|
|
def test_body_is_randomized(self):
|
|
b1 = _build_random_dfp_body()
|
|
b2 = _build_random_dfp_body()
|
|
assert b1 != b2, "每次注册应生成不同的随机 cw"
|
|
|
|
|
|
class TestParseResponse:
|
|
def test_parse_ok(self):
|
|
t1, t2, t5 = _parse_response(_fake_resp())
|
|
assert t1 == "a" * 32
|
|
assert t2 == "A" * 180
|
|
assert 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)
|
|
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
|
|
with pytest.raises(DfpRegistrationError):
|
|
_parse_response(b"\x26\xb4" + b"A" * 180) # 缺 t1/t5
|
|
with pytest.raises(DfpRegistrationError):
|
|
_parse_response(b"\x56\x28" + b"b" * 40) # 缺 t1/t2
|
|
|
|
|
|
class TestRegisterDevice:
|
|
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)
|
|
assert (t1, t2, t5) == ("a" * 32, "A" * 180, "b" * 40)
|
|
# getDfpConfig -> selectOperator -> dfpReport 的 content-type 序列
|
|
assert 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]
|
|
assert sel_ct == "application/x-wup"
|
|
assert new_fp in sel_body, "selectOperator 应注入当前账号画像 fingerprint"
|
|
assert _GOLDEN_FP not in sel_body
|
|
# dfpReport 请求体为随机 cw 结构 (零设备生成)
|
|
dfp_body, dfp_ct = captured[2]
|
|
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 (
|
|
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 pytest.raises(DfpRegistrationError):
|
|
register_device()
|