虎牙网页Cookie来源驱动合并与WSS传输门禁

tests/test_huya_app_login.py

@@ -12,7 +12,11 @@ from sqlalchemy.orm import sessionmaker
 from core.huya import (
     HuyaAppLoginError,
 )
-from core.huya.app_login import login_cred_with_flow, wup_password_login_raw
+from core.huya.app_login import (
+    HuyaAppPasswordLogin,
+    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_fingerprint import account_state_dir, reset_account_state
 from core.huya.device_profile import (
@@ -39,6 +43,110 @@ from web.backend.schemas import (
 )
 
 
+def test_app_login_merges_web_biztoken_and_device_state():
+    session = MagicMock()
+    session.cookies = []
+    cookie = HuyaAppPasswordLogin._merge_web_cookie(
+        session,
+        "web-token",
+        "sdid-value",
+        "hdid-value",
+        {"guid": "a" * 32, "udb_guiddata": "b" * 32},
+    )
+    assert "udb_biztoken=web-token" in cookie
+    assert "sdid=sdid-value" in cookie
+    assert "hdid=hdid-value" in cookie
+    assert "guid=" + "a" * 32 in cookie
+    assert "udb_guiddata=" + "b" * 32 in cookie
+    assert "udb_deviceid=" not in cookie
+
+
+def test_app_login_cookie_injects_app_credentials_and_drops_transient_id():
+    session = MagicMock()
+    session.cookies = {"web_qrlogin_confirm_id": "one-time"}
+    cookie = HuyaAppPasswordLogin._merge_web_cookie(
+        session,
+        "b" * 344,
+        "s" * 193,
+        "h" * 40,
+        {"guid32": "a" * 64, "device_id": "d" * 40},
+        cred=b"c" * 114,
+        uid=1199666914671,
+        username="hy_300023887",
+        include_transient=False,
+    )
+    assert "udb_cred=" in cookie
+    assert "udb_uid=1199666914671" in cookie
+    assert "yyuid=1199666914671" in cookie
+    assert "udb_passport=hy_300023887" in cookie
+    assert "web_qrlogin_confirm_id" not in cookie
+
+
+def test_app_login_cookie_does_not_fabricate_browser_state():
+    session = MagicMock()
+    session.cookies = {}
+    cookie = HuyaAppPasswordLogin._merge_web_cookie(
+        session,
+        "b" * 344,
+        "s" * 193,
+        "h" * 40,
+        {"guid32": "a" * 64, "device_id": "d" * 40},
+        cred=b"c" * 114,
+        uid=1199666914671,
+        username="hy_300023887",
+        include_transient=False,
+    )
+    values = dict(item.split("=", 1) for item in cookie.split("; "))
+    assert set(values) == {
+        "udb_cred", "udb_uid", "yyuid", "udb_version", "udb_biztoken",
+        "udb_origin", "udb_status", "sdid", "hdid", "udb_passport", "username",
+    }
+    assert not {"_qimei_uuid42", "_qimei_fingerprint", "_qimei_h38", "udb_anobiztoken"} & set(values)
+
+
+def test_app_login_cookie_overwrites_auth_but_keeps_distinct_explicit_device_values():
+    session = MagicMock()
+    session.cookies = {
+        "udb_uid": "old",
+        "udb_biztoken": "old-token",
+        "guid": "old-guid",
+        "udb_guiddata": "old-device-guid",
+        "__yamid_new": "persisted",
+    }
+    cookie = HuyaAppPasswordLogin._merge_web_cookie(
+        session,
+        "new-token",
+        "s" * 193,
+        "h" * 40,
+        {"guid": "a" * 32, "udb_guiddata": "b" * 32},
+        cred=b"c" * 114,
+        uid=1199666914671,
+        username="19507362959",
+        include_transient=False,
+    )
+    values = dict(item.split("=", 1) for item in cookie.split("; "))
+    assert values["udb_uid"] == "1199666914671"
+    assert values["udb_biztoken"] == "new-token"
+    assert values["guid"] == "a" * 32
+    assert values["udb_guiddata"] == "b" * 32
+    assert values["__yamid_new"] == "persisted"
+    assert "udb_passport" not in values
+
+
+def test_app_login_cookie_contains_web_session_shape():
+    session = MagicMock()
+    session.cookies = {"udb_uid": "123"}
+    cookie = HuyaAppPasswordLogin._merge_web_cookie(
+        session, "web-token", "s" * 193, "h" * 40,
+        {"guid": "a" * 32, "udb_guiddata": "b" * 32},
+    )
+    keys = {item.split("=", 1)[0] for item in cookie.split("; ")}
+    expected = {
+        "udb_biztoken", "sdid", "hdid", "guid", "udb_guiddata",
+    }
+    assert expected <= keys
+
+
 class TestHuyaAppLogin:
     def setup_method(self):
         self.engine = create_engine("sqlite://")

tests/test_huya_elite_protocol.py

@@ -1,3 +1,6 @@
+import asyncio
+from unittest.mock import AsyncMock
+
 from core.huya.activity_structs import ActPrizeDetailItem, GetActPrizeDetailResp
 from core.huya.http_client import SHOP_BIZ_UA, HuyaHttpClient
 from core.huya.shop_structs import (
@@ -8,6 +11,8 @@ from core.huya.shop_structs import (
     ShopCodeRsp,
 )
 from core.huya.taf_protocol import TafInputStream, TafOutputStream, TafType
+from core.huya.wss_client import TAIL_BYTES, HuyaWssClient, WssCommand, WssMessage
+from web.backend.services.huya_runner_core import HuyaBatchRunnerCore
 
 
 def _round_trip(value):
@@ -87,3 +92,39 @@ def test_shop_preflight_response_contracts():
     status.message = ""
     decoded_status = _round_trip(status)
     assert decoded_status.code == 200
+
+
+def test_wss_requires_web_device_cookie_fields():
+        assert not HuyaBatchRunnerCore._has_web_wss_cookie(
+            "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1"
+        )
+        assert HuyaBatchRunnerCore._has_web_wss_cookie(
+            "udb_cred=x; udb_biztoken=x; udb_uid=1; yyuid=1; sdid=s; guid=g; "
+            "udb_guiddata=gd; udb_deviceid=w_1234567890123456789; game_did=did; "
+            "_qimei_uuid42=q; udb_anobiztoken=ab; __yamid_new=y"
+        )
+
+
+def test_activity_auth_frame_matches_latest_capture_layout():
+    async def build_frame():
+        client = HuyaWssClient()
+        client.ws = type("FakeWs", (), {"send": AsyncMock()})()
+        await client.send_auth(1199664135026, "udb_cred=" + "x" * 300, sequence=0x1D000109)
+        return client.ws.send.call_args.args[0]
+
+    message = WssMessage.decode(asyncio.run(build_frame()))
+    assert message.command == WssCommand.AUTH
+    assert message.sequence == 0x1D000109
+    assert message.body.endswith(TAIL_BYTES)
+
+    stream = TafInputStream(message.body[: -len(TAIL_BYTES)])
+    assert stream.read_int64(0) == 1199664135026
+    assert stream.read_string(1) == "webh5&0.0.1&websocket&&diypc_52775"
+    assert stream.read_head() == (2, TafType.STRING4)
+    cookie_len = int.from_bytes(stream.buf.read(4), "big")
+    assert cookie_len > 255
+    stream.buf.read(cookie_len)
+    assert stream.read_int32(3) == 0
+    assert stream.read_int8(4) == 1
+    assert stream.read_string(5) == "HUYA&ZH&2052"
+    assert stream.read_string(6) == ""

tests/test_huya_runner_transport.py

@@ -0,0 +1,28 @@
+"""虎牙精英宝典传输选择的联调前置检查。"""
+
+from web.backend.services.huya_runner_core import HuyaBatchRunnerCore
+
+
+def test_minimal_app_cookie_reports_only_missing_web_guid():
+    cookie = (
+        "udb_cred=cred; udb_biztoken=biz; udb_uid=1199659425268; "
+        "yyuid=1199659425268; sdid=sdid"
+    )
+
+    assert HuyaBatchRunnerCore._web_wss_cookie_missing(cookie) == (
+        "guid", "udb_guiddata", "udb_deviceid", "game_did",
+        "_qimei_uuid42", "udb_anobiztoken", "__yamid_new",
+    )
+    assert not HuyaBatchRunnerCore._has_web_wss_cookie(cookie)
+
+
+def test_complete_web_cookie_enables_wss_without_logging_values():
+    cookie = (
+        "udb_cred=cred; udb_biztoken=biz; udb_uid=1199659425268; "
+        "yyuid=1199659425268; sdid=sdid; guid=0123456789abcdef0123456789abcdef; "
+        "udb_guiddata=guiddata; udb_deviceid=w_1234567890123456789; game_did=game; "
+        "_qimei_uuid42=qimei; udb_anobiztoken=anobiz; __yamid_new=yamid"
+    )
+
+    assert HuyaBatchRunnerCore._web_wss_cookie_missing(cookie) == ()
+    assert HuyaBatchRunnerCore._has_web_wss_cookie(cookie)