真实浏览器抓包与匿名设备态协议生成器

2026-09-01 19:17:36 +08:00
parent 48c37bb8d7
commit 8447ca82ed
3 changed files with 901 additions and 0 deletions
@@ -0,0 +1,132 @@
+"""core/huya/anon_device.py 的协议装配与解析测试（mock 网络，不打真实请求）。"""
+
+from unittest.mock import Mock
+
+import pytest
+import requests
+
+from core.huya.anon_device import (
+    ANON_LOGIN_URL,
+    AnonymousDeviceError,
+    fetch_anonymous_device_state,
+)
+
+
+def _fake_response(status=200, headers=None, text="", json_body=None):
+    resp = Mock(spec=requests.Response)
+    resp.status_code = status
+    resp.headers = headers or {}
+    resp.text = text
+    resp.json = Mock(return_value=json_body) if json_body is not None else Mock(
+        side_effect=ValueError("bad json")
+    )
+    return resp
+
+
+def _fake_anon_body(uid="1471255709907", biztoken="A" * 344):
+    return {
+        "uri": 10014,
+        "returnCode": 0,
+        "message": "success",
+        "data": {"uid": uid, "biztoken": biztoken},
+    }
+
+
+def test_fetches_four_fields_from_middle_and_anon(monkeypatch):
+    session = Mock(spec=requests.Session)
+    deviceid = "w_1147246591246680065"
+    session.get.return_value = _fake_response(
+        headers={"Set-Cookie": f"udb_deviceid={deviceid}; Max-Age=315360000; Domain=huya.com"}
+    )
+    session.post.return_value = _fake_response(json_body=_fake_anon_body())
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    state = fetch_anonymous_device_state(session=session)
+
+    assert state["udb_deviceid"] == deviceid
+    assert state["udb_anouid"] == "1471255709907"
+    assert state["udb_anobiztoken"] == "A" * 344
+    assert len(state["udb_guiddata"]) == 32
+
+    # middle 请求装配：URL 含 8 位随机数与 32hex guiddata
+    get_url = session.get.call_args.args[0]
+    parts = get_url.split("/")
+    assert len(parts[-1]) == 32
+    assert len(parts[-3]) == 8
+    assert parts[-3].isdigit()
+    assert state["udb_guiddata"] == parts[-1]
+
+    # anonymousLogin 装配：必须带 middle Referer；body 含随机 context/csid
+    post_kwargs = session.post.call_args.kwargs
+    assert session.post.call_args.args[0] == ANON_LOGIN_URL
+    assert post_kwargs["headers"]["Referer"] == get_url
+    import json as _json
+
+    payload = _json.loads(post_kwargs["data"])
+    assert payload["context"].startswith("WB-")
+    assert payload["sdid"].startswith("csid_")
+
+
+def test_values_are_random_across_calls(monkeypatch):
+    session = Mock(spec=requests.Session)
+    session.get.return_value = _fake_response(
+        headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
+    )
+    session.post.return_value = _fake_response(json_body=_fake_anon_body(uid="1"))
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    import json as _json
+
+    a = fetch_anonymous_device_state(session=session)
+    b = fetch_anonymous_device_state(session=session)
+
+    # 客户端随机部分：guiddata(middle URL) 与 anonymousLogin 的 context/csid
+    assert a["udb_guiddata"] != b["udb_guiddata"]
+    post_a = _json.loads(session.post.call_args_list[0].kwargs["data"])
+    post_b = _json.loads(session.post.call_args_list[1].kwargs["data"])
+    assert post_a["context"] != post_b["context"]
+    assert post_a["sdid"] != post_b["sdid"]
+
+
+def test_bad_middle_status_raises(monkeypatch):
+    session = Mock(spec=requests.Session)
+    session.get.return_value = _fake_response(status=403)
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    with pytest.raises(AnonymousDeviceError):
+        fetch_anonymous_device_state(session=session)
+
+
+def test_missing_deviceid_in_set_cookie_raises(monkeypatch):
+    session = Mock(spec=requests.Session)
+    session.get.return_value = _fake_response(headers={"Set-Cookie": "PHPSESSID=x"})
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    with pytest.raises(AnonymousDeviceError, match="udb_deviceid"):
+        fetch_anonymous_device_state(session=session)
+
+
+def test_anon_error_return_code_raises(monkeypatch):
+    session = Mock(spec=requests.Session)
+    session.get.return_value = _fake_response(
+        headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
+    )
+    session.post.return_value = _fake_response(
+        json_body={"uri": 10014, "returnCode": 108, "message": "blocked"}
+    )
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    with pytest.raises(AnonymousDeviceError, match="returnCode=108"):
+        fetch_anonymous_device_state(session=session)
+
+
+def test_non_json_anon_body_raises(monkeypatch):
+    session = Mock(spec=requests.Session)
+    session.get.return_value = _fake_response(
+        headers={"Set-Cookie": "udb_deviceid=w_1147246591246680065; Max-Age=1"}
+    )
+    session.post.return_value = _fake_response(text="error!")
+    monkeypatch.setattr("core.huya.anon_device._new_session", lambda **kw: session)
+
+    with pytest.raises(AnonymousDeviceError, match="非 JSON"):
+        fetch_anonymous_device_state(session=session)