fix: harden real game login boundaries

cq2_sdk/src/cq2_sdk/client.py

@@ -15,6 +15,7 @@ creates a role implicitly.
 """
 from __future__ import annotations
 
+import asyncio
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -99,16 +100,21 @@ class LauncherFacade:
         timeout: float = 10.0,
     ) -> GameSession:
         """Launch the selected zone and complete CQ2 TCP world login."""
-        del server_name  # This facade currently targets one configured game.
         if self.session is None:
             raise RuntimeError("run_until_enter_game requires an authenticated BoxSession")
+        if server_name != "玖扬复古176":
+            raise ValueError(f"unsupported server_name: {server_name}")
+        if self.tcp is not None or self.game_session is not None:
+            raise RuntimeError("launcher facade already owns a retained TCP session")
+        if zone_index < 0:
+            raise ValueError("zone_index must be non-negative")
 
         servers = await self.box.get_today_open(game_id)
         if not servers:
             raise RuntimeError("no servers open today")
-        if zone_index < 0:
-            raise ValueError("zone_index must be non-negative")
-        target = servers[min(zone_index, len(servers) - 1)]
+        if zone_index >= len(servers):
+            raise IndexError("zone_index is outside today's server list")
+        target = servers[zone_index]
         await self.box.enter_game_client(game_id=game_id, serverid=target.serverid)
         server = await fetch_server_info(
             target.serverid,
@@ -117,7 +123,7 @@ class LauncherFacade:
         )
         tcp = TcpClient(server.host, server.proxy_port or server.port)
         try:
-            await tcp.connect()
+            await asyncio.wait_for(tcp.connect(), timeout=timeout)
             game_session = await GameLogin(tcp).login(
                 self.session,
                 server,

cq2_sdk/src/cq2_sdk/login/game.py

@@ -2,10 +2,11 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Awaitable
 from dataclasses import dataclass
 import json
 import random
-from typing import Callable
+from typing import Callable, TypeVar
 
 from cq2_sdk.login.models import BoxSession, ServerInfo
 from cq2_sdk.lua_bridge import Cq2Lua
@@ -15,6 +16,9 @@ from cq2_sdk.transport.framing import FrameHeader
 from cq2_sdk.transport.tcp import TcpClient
 
 
+_T = TypeVar("_T")
+
+
 @dataclass
 class GameSession:
     char_id: str
@@ -98,11 +102,30 @@ class GameLogin:
                 raise GameLoginError(
                     detail,
                     message_id=header.msg_id,
-                    error_code=header.recog or header.param1,
+                    error_code=(
+                        header.param1
+                        if header.msg_id == C.SC_527
+                        else header.recog
+                    ),
                 )
             if header.msg_id == expected:
                 return header, body
 
+    @staticmethod
+    async def _before_deadline(
+        operation: Callable[[], Awaitable[_T]],
+        *,
+        deadline: float,
+        context: str,
+    ) -> _T:
+        remaining = deadline - asyncio.get_running_loop().time()
+        if remaining <= 0:
+            raise TimeoutError(f"timed out during {context}")
+        try:
+            return await asyncio.wait_for(operation(), timeout=remaining)
+        except asyncio.TimeoutError as exc:
+            raise TimeoutError(f"timed out during {context}") from exc
+
     async def login(
         self,
         session: BoxSession,
@@ -117,17 +140,22 @@ class GameLogin:
         loop = asyncio.get_running_loop()
         deadline = loop.time() + timeout
         main_server_id = self._main_server_id(server)
+        self.tcp.set_sdata(0)
 
         token = encode6(session.token.encode("utf-8")).decode("ascii")
         channel = encode6(server.channel_id.encode("utf-8")).decode("ascii")
         body = f"{session.uid}/{token}/{channel}/1/0".encode("utf-8")
-        await self.tcp.send(
-            C.CS_2001,
-            900000,
-            self._stamp(),
-            main_server_id,
-            server.port,
-            body,
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_2001,
+                900000,
+                self._stamp(),
+                main_server_id,
+                server.port,
+                body,
+            ),
+            deadline=deadline,
+            context="CS2001 send",
         )
         response, response_body = await self._receive(C.SC_529, deadline=deadline)
         login_key, user_id = self._split(response_body, 2, "SC529 body")[:2]
@@ -138,13 +166,17 @@ class GameLogin:
             if new_login
             else str(main_server_id).encode("ascii")
         )
-        await self.tcp.send(
-            C.CS_104,
-            0,
-            0,
-            0,
-            0,
-            server_info_body,
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_104,
+                0,
+                0,
+                0,
+                0,
+                server_info_body,
+            ),
+            deadline=deadline,
+            context="CS104 send",
         )
         response, response_body = await self._receive(C.SC_530, deadline=deadline)
         role_port = server.port
@@ -159,7 +191,14 @@ class GameLogin:
                 raise GameLoginError("malformed SC530 role endpoint") from exc
         self.tcp.set_sdata(session_id)
         if not new_login:
-            await self.tcp.reconnect(server.host, server.proxy_port or role_port)
+            await self._before_deadline(
+                lambda: self.tcp.reconnect(
+                    server.host,
+                    server.proxy_port or role_port,
+                ),
+                deadline=deadline,
+                context="role-server reconnect",
+            )
 
         body = (
             f"{session.uid}/{session_id}/{server.server_id}/"
@@ -167,13 +206,17 @@ class GameLogin:
         )
         if new_login:
             body = f"{body}/{login_key}"
-        await self.tcp.send(
-            C.CS_100,
-            900000,
-            self._stamp(),
-            main_server_id,
-            role_port,
-            body.encode("utf-8"),
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_100,
+                900000,
+                self._stamp(),
+                main_server_id,
+                role_port,
+                body.encode("utf-8"),
+            ),
+            deadline=deadline,
+            context="CS100 send",
         )
         _, response_body = await self._receive(C.SC_520, deadline=deadline)
         try:
@@ -200,7 +243,11 @@ class GameLogin:
                 ensure_ascii=False,
                 separators=(",", ":"),
             ).encode("utf-8")
-            await self.tcp.send(C.CS_101, 0, 0, 0, 0, create_body)
+            await self._before_deadline(
+                lambda: self.tcp.send(C.CS_101, 0, 0, 0, 0, create_body),
+                deadline=deadline,
+                context="CS101 send",
+            )
             _, created_body = await self._receive(C.SC_521, deadline=deadline)
             role_name, role_id = self._split(
                 created_body, 2, "SC521 body"
@@ -225,13 +272,17 @@ class GameLogin:
             if not role_name or not role_id:
                 raise GameLoginError("malformed role in SC520")
 
-        await self.tcp.send(
-            C.CS_103,
-            0,
-            0,
-            0,
-            0,
-            f"{session.uid}/{role_name}".encode("utf-8"),
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_103,
+                0,
+                0,
+                0,
+                0,
+                f"{session.uid}/{role_name}".encode("utf-8"),
+            ),
+            deadline=deadline,
+            context="CS103 send",
         )
         _, response_body = await self._receive(C.SC_525, deadline=deadline)
         world_parts = self._split(response_body, 2, "SC525 body")
@@ -239,7 +290,14 @@ class GameLogin:
             world_port = int(world_parts[1])
         except ValueError as exc:
             raise GameLoginError("malformed SC525 world port") from exc
-        await self.tcp.reconnect(server.host, server.proxy_port or world_port)
+        await self._before_deadline(
+            lambda: self.tcp.reconnect(
+                server.host,
+                server.proxy_port or world_port,
+            ),
+            deadline=deadline,
+            context="world-server reconnect",
+        )
 
         body = (
             f"*#{session.uid}/{role_name}/{role_id}/{session_id}/"
@@ -247,13 +305,17 @@ class GameLogin:
             f"{server.certified}/{server.adult}/{server.is_box_login}/"
             f"{server.batch_messages}"
         ).encode("utf-8")
-        await self.tcp.send(
-            C.CS_20,
-            900000,
-            self._stamp(),
-            main_server_id,
-            world_port,
-            body,
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_20,
+                900000,
+                self._stamp(),
+                main_server_id,
+                world_port,
+                body,
+            ),
+            deadline=deadline,
+            context="CS20 send",
         )
         await self._receive(C.SC_658, deadline=deadline)
 
@@ -274,13 +336,17 @@ class GameLogin:
         screen = (server.screen_width & 0xFFFF) | (
             (server.screen_height & 0xFFFF) << 16
         )
-        await self.tcp.send(
-            C.CS_1018,
-            server.platform,
-            0,
-            screen,
-            server.device_code,
-            enter_body,
+        await self._before_deadline(
+            lambda: self.tcp.send(
+                C.CS_1018,
+                server.platform,
+                0,
+                screen,
+                server.device_code,
+                enter_body,
+            ),
+            deadline=deadline,
+            context="CS1018 send",
         )
         await self._receive(C.SC_2, deadline=deadline)
         return GameSession(

cq2_sdk/src/cq2_sdk/login/models.py

@@ -5,10 +5,12 @@ modules. Using dataclasses (not pydantic) because they're internal
 DTOs and don't need validation.
 """
 from __future__ import annotations
+from collections.abc import Mapping
 from dataclasses import dataclass, field
+from types import MappingProxyType
 
 
-@dataclass
+@dataclass(frozen=True)
 class ServerInfo:
     server_id: int
     name: str
@@ -16,6 +18,12 @@ class ServerInfo:
     port: int
     main_server_id: int | None = None
     proxy_port: int | None = None
+    state: int = 0
+    url_suffix: str = ""
+    tag: str | int = ""
+    gm_url_suffix: str = ""
+    explicit_ip: str = ""
+    beta: bool = False
     channel_id: str = "110585"
     game_id: int = 206923
     platform: int = 2
@@ -28,7 +36,14 @@ class ServerInfo:
     adult: int = 0
     is_box_login: int = 1
     batch_messages: int = 0
-    config_versions: dict[str, str] = field(default_factory=dict)
+    config_versions: Mapping[str, str] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(
+            self,
+            "config_versions",
+            MappingProxyType(dict(self.config_versions)),
+        )
 
 
 @dataclass

cq2_sdk/src/cq2_sdk/login/server_list.py

@@ -56,7 +56,11 @@ def _xxtea_decrypt(ciphertext: bytes) -> bytes:
 
     plaintext_length = values[-1]
     available = (len(values) - 1) * 4
-    if plaintext_length <= 0 or plaintext_length > available:
+    if (
+        plaintext_length <= 0
+        or plaintext_length > available
+        or plaintext_length < available - 3
+    ):
         raise ServerListError("invalid XXTEA plaintext length")
     return struct.pack(f"<{len(values) - 1}I", *values[:-1])[:plaintext_length]
 
@@ -127,6 +131,11 @@ def resolve_server(
     proxy_port = _integer(
         record.get("x", record.get("proxyport")), default=0
     ) or None
+    state = _integer(record.get("s", record.get("state")), default=0)
+    url_suffix = str(record.get("u", record.get("urlsuffix")) or "")
+    tag = record.get("t", record.get("tag", ""))
+    gm_url_suffix = str(record.get("g", record.get("gmurlsuffix")) or "")
+    beta = record.get("b", record.get("beta")) == "1"
 
     return ServerInfo(
         server_id=selected_id,
@@ -135,6 +144,12 @@ def resolve_server(
         host=host,
         port=port,
         proxy_port=proxy_port,
+        state=state,
+        url_suffix=url_suffix,
+        tag=tag,
+        gm_url_suffix=gm_url_suffix,
+        explicit_ip=explicit_host,
+        beta=beta,
         channel_id=str(channel_id),
         game_id=int(game_id),
     )
@@ -150,7 +165,7 @@ async def fetch_server_info(
     """Fetch and resolve one server from the channel-specific detailed list."""
     url = f"{SERVER_LIST_BASE}/serverlist/{channel_id}/serverlist.txt"
     owns_client = client is None
-    http = client or httpx.AsyncClient(timeout=15, verify=False)
+    http = client or httpx.AsyncClient(timeout=15)
     try:
         response = await http.get(url)
         response.raise_for_status()

cq2_sdk/tests/unit/test_client.py

@@ -1,4 +1,6 @@
 """Tests for Cq2Client + LauncherFacade using the REAL protocol (all mocked)."""
+import asyncio
+
 import httpx
 import pytest
 
@@ -175,6 +177,173 @@ async def test_run_until_enter_game_requires_authenticated_session():
         await box.close()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("server_name", "zone_index", "expected_today_calls"),
+    [("unsupported-game", 0, 0), ("玖扬复古176", 2, 1)],
+)
+async def test_invalid_launch_selection_has_no_enter_game_or_tcp_side_effects(
+    monkeypatch, server_name, zone_index, expected_today_calls
+):
+    calls = []
+
+    class Box:
+        async def get_today_open(self, game_id):
+            calls.append(("today", game_id))
+            return [
+                TodayOpenServer(
+                    server_name="One",
+                    serverid=1,
+                    open_time=1,
+                    section="00:00",
+                    state=1,
+                    main_serverid="1",
+                )
+            ]
+
+        async def enter_game_client(self, **kwargs):
+            calls.append(("enter", kwargs))
+
+    def forbidden_tcp(*args, **kwargs):
+        calls.append(("tcp", args, kwargs))
+        raise AssertionError("TCP setup must not run")
+
+    monkeypatch.setattr("cq2_sdk.client.TcpClient", forbidden_tcp)
+    facade = LauncherFacade(
+        Box(),
+        BoxSession(uid="u", token="t", sdkid="", signkey=""),
+    )
+
+    with pytest.raises((ValueError, IndexError)):
+        await facade.run_until_enter_game(
+            server_name=server_name,
+            zone_index=zone_index,
+        )
+
+    assert sum(call[0] == "today" for call in calls) == expected_today_calls
+    assert not any(call[0] in {"enter", "tcp"} for call in calls)
+
+
+@pytest.mark.asyncio
+async def test_initial_tcp_connect_is_bounded_by_login_timeout(monkeypatch):
+    server = ServerInfo(server_id=1, name="One", host="game.example", port=8041)
+    connect_started = asyncio.Event()
+    instances = []
+
+    async def fake_resolve(*args, **kwargs):
+        return server
+
+    class StalledTcp:
+        def __init__(self, host, port):
+            self.cancelled = False
+            self.closed = False
+            instances.append(self)
+
+        async def connect(self):
+            connect_started.set()
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled = True
+                raise
+
+        async def close(self):
+            self.closed = True
+
+    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
+    monkeypatch.setattr("cq2_sdk.client.TcpClient", StalledTcp)
+    box = make_box()
+    facade = LauncherFacade(
+        box,
+        BoxSession(uid="u", token="t", sdkid="", signkey=""),
+    )
+    loop = asyncio.get_running_loop()
+    began = loop.time()
+    try:
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(
+                facade.run_until_enter_game(timeout=0.02),
+                timeout=0.2,
+            )
+    finally:
+        await box.close()
+
+    assert connect_started.is_set()
+    assert loop.time() - began < 0.15
+    assert instances[0].cancelled is True
+    assert instances[0].closed is True
+
+
+@pytest.mark.asyncio
+async def test_second_launch_is_rejected_before_external_side_effects(monkeypatch):
+    events = []
+    server = ServerInfo(server_id=1, name="One", host="game.example", port=8041)
+    game_session = GameSession(
+        char_id="r1",
+        char_name="Hero",
+        session_id=149,
+        user_id="u1",
+        world_ready=True,
+    )
+
+    class Box:
+        async def get_today_open(self, game_id):
+            events.append("today")
+            return [
+                TodayOpenServer(
+                    server_name="One",
+                    serverid=1,
+                    open_time=1,
+                    section="00:00",
+                    state=1,
+                    main_serverid="1",
+                )
+            ]
+
+        async def enter_game_client(self, **kwargs):
+            events.append("enter")
+
+    async def fake_resolve(*args, **kwargs):
+        events.append("resolve")
+        return server
+
+    class FakeTcp:
+        def __init__(self, host, port):
+            events.append("tcp")
+
+        async def connect(self):
+            events.append("connect")
+
+        async def start_heartbeat(self):
+            events.append("heartbeat")
+
+        async def close(self):
+            events.append("close")
+
+    class FakeGameLogin:
+        def __init__(self, tcp):
+            pass
+
+        async def login(self, *args, **kwargs):
+            events.append("login")
+            return game_session
+
+    monkeypatch.setattr("cq2_sdk.client.fetch_server_info", fake_resolve)
+    monkeypatch.setattr("cq2_sdk.client.TcpClient", FakeTcp)
+    monkeypatch.setattr("cq2_sdk.client.GameLogin", FakeGameLogin)
+    facade = LauncherFacade(
+        Box(),
+        BoxSession(uid="u", token="t", sdkid="", signkey=""),
+    )
+
+    assert await facade.run_until_enter_game() is game_session
+    after_first_launch = list(events)
+    with pytest.raises(RuntimeError, match="already owns"):
+        await facade.run_until_enter_game()
+
+    assert events == after_first_launch
+
+
 @pytest.mark.asyncio
 async def test_run_until_enter_game_closes_tcp_when_ready_binding_fails(monkeypatch):
     events = []

cq2_sdk/tests/unit/test_game_login.py

@@ -51,11 +51,14 @@ class _ScriptedTcp:
     def __init__(self, responses):
         self.responses = list(responses)
         self.sent = []
+        self.sent_sdata = []
         self.reconnections = []
         self.sdata = 0
+        self.sdata_updates = []
 
     async def send(self, *args):
         self.sent.append(args)
+        self.sent_sdata.append(self.sdata)
 
     async def recv(self, timeout):
         if not self.responses:
@@ -65,6 +68,7 @@ class _ScriptedTcp:
 
     def set_sdata(self, value):
         self.sdata = value
+        self.sdata_updates.append(value)
 
     async def reconnect(self, host, port):
         self.reconnections.append((host, port))
@@ -89,6 +93,34 @@ def _server(**changes):
     return ServerInfo(**values)
 
 
+async def _start_role_server_for_world_gate(world_port):
+    async def role_handler(reader, writer):
+        responses = [
+            _response(C.SC_529, recog=1, body=b"login-key/user-42"),
+            _response(C.SC_530, recog=149, sequence=1),
+            _response(
+                C.SC_520,
+                body=(
+                    b'{"list":[{"chr":"Hero","UserID":"role-123",'
+                    b'"last":1}]}'
+                ),
+                sequence=2,
+            ),
+            _response(C.SC_525, body=f"ignored/{world_port}".encode(), sequence=3),
+        ]
+        try:
+            for response in responses:
+                await _read_frame(reader)
+                writer.write(response)
+                await writer.drain()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+    server = await asyncio.start_server(role_handler, "127.0.0.1", 0)
+    return server, server.sockets[0].getsockname()[1]
+
+
 @pytest.mark.asyncio
 async def test_new_login_existing_role_enters_world():
     role_frames = []
@@ -343,27 +375,160 @@ async def test_old_login_reconnects_from_sc530_before_role_lookup():
     assert tcp.sent[2][5] == b"account-1/151/7/110585/206923"
 
 
+@pytest.mark.asyncio
+async def test_fresh_login_resets_reused_transport_sdata_before_cs2001():
+    tcp = _ScriptedTcp([(_header(C.SC_503, recog=-6), b"")])
+    tcp.sdata = 149
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    with pytest.raises(GameLoginError):
+        await login.login(
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(),
+            timeout=0.05,
+        )
+
+    assert tcp.sent[0][0] == C.CS_2001
+    assert tcp.sent_sdata[0] == 0
+    assert tcp.sdata_updates[0] == 0
+
+
+@pytest.mark.asyncio
+async def test_stalled_send_is_cancelled_at_shared_login_deadline():
+    started = asyncio.Event()
+
+    class StalledSendTcp(_ScriptedTcp):
+        def __init__(self):
+            super().__init__([])
+            self.cancelled = False
+
+        async def send(self, *args):
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled = True
+                raise
+
+    tcp = StalledSendTcp()
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+    loop = asyncio.get_running_loop()
+    began = loop.time()
+
+    with pytest.raises(TimeoutError):
+        await asyncio.wait_for(
+            login.login(
+                BoxSession(uid="u", token="t", sdkid="", signkey=""),
+                _server(),
+                timeout=0.02,
+            ),
+            timeout=0.2,
+        )
+
+    assert started.is_set()
+    assert loop.time() - began < 0.15
+    assert tcp.cancelled is True
+
+
+@pytest.mark.asyncio
+async def test_external_cancellation_is_not_converted_to_timeout():
+    started = asyncio.Event()
+
+    class StalledSendTcp(_ScriptedTcp):
+        async def send(self, *args):
+            started.set()
+            await asyncio.Event().wait()
+
+    login_task = asyncio.create_task(
+        GameLogin(StalledSendTcp([]), randint=lambda _low, _high: 1).login(
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(),
+            timeout=1.0,
+        )
+    )
+    await asyncio.wait_for(started.wait(), timeout=0.1)
+    login_task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await login_task
+
+
+@pytest.mark.asyncio
+async def test_stalled_reconnect_is_cancelled_at_shared_login_deadline():
+    class StalledReconnectTcp(_ScriptedTcp):
+        def __init__(self):
+            super().__init__(
+                [
+                    (_header(C.SC_529, recog=0), b"unused/user-old"),
+                    (_header(C.SC_530), b"ignored/7777/151"),
+                ]
+            )
+            self.cancelled = False
+
+        async def reconnect(self, host, port):
+            self.reconnections.append((host, port))
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                self.cancelled = True
+                raise
+
+    tcp = StalledReconnectTcp()
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+    loop = asyncio.get_running_loop()
+    began = loop.time()
+
+    with pytest.raises(TimeoutError):
+        await asyncio.wait_for(
+            login.login(
+                BoxSession(uid="u", token="t", sdkid="", signkey=""),
+                _server(),
+                timeout=0.02,
+            ),
+            timeout=0.2,
+        )
+
+    assert tcp.reconnections == [("game.example", 7777)]
+    assert loop.time() - began < 0.15
+    assert tcp.cancelled is True
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("responses", "char_name", "message_id", "error_code"),
     [
-        ([(_header(C.SC_503, recog=-6), b"")], None, C.SC_503, -6),
+        (
+            [(_header(C.SC_503, recog=0, param1=101), b"")],
+            None,
+            C.SC_503,
+            0,
+        ),
         (
             [
                 (_header(C.SC_529, recog=1), b"key/user"),
                 (_header(C.SC_530, recog=149), b""),
-                (_header(C.SC_527, param1=3), b""),
+                (_header(C.SC_527, recog=202, param1=3), b""),
             ],
             None,
             C.SC_527,
             3,
         ),
+        (
+            [
+                (_header(C.SC_529, recog=1), b"key/user"),
+                (_header(C.SC_530, recog=149), b""),
+                (_header(C.SC_527, recog=202, param1=0), b""),
+            ],
+            None,
+            C.SC_527,
+            0,
+        ),
         (
             [
                 (_header(C.SC_529, recog=1), b"key/user"),
                 (_header(C.SC_530, recog=149), b""),
                 (_header(C.SC_520), b'{"list":[]}'),
-                (_header(C.SC_522, recog=-10), b""),
+                (_header(C.SC_522, recog=-10, param1=303), b""),
             ],
             "New",
             C.SC_522,
@@ -377,7 +542,7 @@ async def test_old_login_reconnects_from_sc530_before_role_lookup():
                     _header(C.SC_520),
                     b'{"list":[{"chr":"Hero","UserID":"r1","last":1}]}',
                 ),
-                (_header(C.SC_526, recog=99), b""),
+                (_header(C.SC_526, recog=99, param1=404), b""),
             ],
             None,
             C.SC_526,
@@ -441,3 +606,125 @@ async def test_unrelated_messages_are_ignored_until_shared_deadline():
             _server(),
             timeout=0.01,
         )
+
+
+@pytest.mark.asyncio
+async def test_world_entry_waits_for_each_protocol_gate():
+    allow_sc658 = asyncio.Event()
+    allow_sc2 = asyncio.Event()
+    received_cs20 = asyncio.Event()
+    probing_cs1018 = asyncio.Event()
+    received_cs1018 = asyncio.Event()
+    world_done = asyncio.Event()
+    world_frames = []
+
+    async def world_handler(reader, writer):
+        try:
+            world_frames.append(await _read_frame(reader))
+            received_cs20.set()
+            cs1018_task = asyncio.create_task(_read_frame(reader))
+            probing_cs1018.set()
+            await allow_sc658.wait()
+            writer.write(_response(C.SC_658))
+            await writer.drain()
+            world_frames.append(await cs1018_task)
+            received_cs1018.set()
+            await allow_sc2.wait()
+            writer.write(_response(C.SC_2, sequence=1))
+            await writer.drain()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            world_done.set()
+
+    world_server = await asyncio.start_server(world_handler, "127.0.0.1", 0)
+    world_port = world_server.sockets[0].getsockname()[1]
+    role_server, role_port = await _start_role_server_for_world_gate(world_port)
+    tcp = TcpClient("127.0.0.1", role_port)
+    await tcp.connect()
+    login_task = asyncio.create_task(
+        GameLogin(tcp, randint=lambda _low, _high: 1).login(
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(host="127.0.0.1", port=role_port),
+            timeout=1.0,
+        )
+    )
+    try:
+        await asyncio.wait_for(received_cs20.wait(), timeout=0.5)
+        await asyncio.wait_for(probing_cs1018.wait(), timeout=0.5)
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(received_cs1018.wait(), timeout=0.02)
+        assert [header.msg_id for header, _ in world_frames] == [C.CS_20]
+        assert login_task.done() is False
+
+        allow_sc658.set()
+        await asyncio.wait_for(received_cs1018.wait(), timeout=0.5)
+        assert [header.msg_id for header, _ in world_frames] == [C.CS_20, C.CS_1018]
+        assert login_task.done() is False
+
+        allow_sc2.set()
+        result = await asyncio.wait_for(login_task, timeout=0.5)
+        assert result.world_ready is True
+    finally:
+        allow_sc658.set()
+        allow_sc2.set()
+        if not login_task.done():
+            login_task.cancel()
+        await tcp.close()
+        await asyncio.wait_for(world_done.wait(), timeout=0.5)
+        role_server.close()
+        world_server.close()
+        await role_server.wait_closed()
+        await world_server.wait_closed()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("withheld_message", "expected_frames"),
+    [(C.SC_658, [C.CS_20]), (C.SC_2, [C.CS_20, C.CS_1018])],
+)
+async def test_world_entry_times_out_while_gate_is_withheld(
+    withheld_message, expected_frames
+):
+    world_frames = []
+    world_done = asyncio.Event()
+
+    async def world_handler(reader, writer):
+        try:
+            world_frames.append(await _read_frame(reader))
+            if withheld_message == C.SC_2:
+                writer.write(_response(C.SC_658))
+                await writer.drain()
+                world_frames.append(await _read_frame(reader))
+                await asyncio.sleep(0.2)
+            else:
+                try:
+                    world_frames.append(await _read_frame(reader))
+                except asyncio.IncompleteReadError:
+                    pass
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            world_done.set()
+
+    world_server = await asyncio.start_server(world_handler, "127.0.0.1", 0)
+    world_port = world_server.sockets[0].getsockname()[1]
+    role_server, role_port = await _start_role_server_for_world_gate(world_port)
+    tcp = TcpClient("127.0.0.1", role_port)
+    await tcp.connect()
+    try:
+        with pytest.raises(TimeoutError, match=f"SC {withheld_message}"):
+            await GameLogin(tcp, randint=lambda _low, _high: 1).login(
+                BoxSession(uid="u", token="t", sdkid="", signkey=""),
+                _server(host="127.0.0.1", port=role_port),
+                timeout=0.05,
+            )
+    finally:
+        await tcp.close()
+        await asyncio.wait_for(world_done.wait(), timeout=0.5)
+        role_server.close()
+        world_server.close()
+        await role_server.wait_closed()
+        await world_server.wait_closed()
+
+    assert [header.msg_id for header, _ in world_frames] == expected_frames

cq2_sdk/tests/unit/test_server_list.py

@@ -1,4 +1,5 @@
 import base64
+from dataclasses import FrozenInstanceError
 
 import pytest
 
@@ -18,6 +19,8 @@ ENCRYPTED_SERVER_LIST = (
     b"/x+A=="
 )
 
+INVALID_PADDED_LENGTH_SERVER_LIST = b"NG1lSm5QeWxWnBdj3dyvl1nAc+J5GmVq"
+
 
 def test_decode_and_resolve_compact_server_fields():
     data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)
@@ -29,7 +32,13 @@ def test_decode_and_resolve_compact_server_fields():
     assert server.name == "Seven"
     assert server.host == "s9-edge.example"
     assert server.port == 8041
+    assert server.state == 1
+    assert server.url_suffix == "v"
+    assert server.tag == 1
+    assert server.gm_url_suffix == ""
+    assert server.explicit_ip == ""
     assert server.proxy_port == 9000
+    assert server.beta is False
     assert server.channel_id == "110585"
     assert server.game_id == 206923
 
@@ -40,10 +49,50 @@ def test_resolve_server_prefers_explicit_ip():
     server = resolve_server(data, 8, channel_id="c", game_id=1)
 
     assert server.host == "1.2.3.4"
+    assert server.explicit_ip == "1.2.3.4"
     assert server.main_server_id == 8
     assert server.proxy_port is None
 
 
+def test_resolve_server_expands_long_form_aliases():
+    server = resolve_server(
+        {
+            "srvRoot": "edge.example",
+            "serverData": [
+                {
+                    "srvid": "17",
+                    "port": "8041",
+                    "state": "4",
+                    "urlsuffix": "stable-v2",
+                    "mainsrvid": "19",
+                    "tag": "featured",
+                    "srvname": "Seventeen",
+                    "gmurlsuffix": "gm-v2",
+                    "srvip": "10.0.0.17",
+                    "proxyport": "9001",
+                    "beta": "1",
+                }
+            ],
+        },
+        17,
+        channel_id="channel",
+        game_id=42,
+    )
+
+    assert server.server_id == 17
+    assert server.port == 8041
+    assert server.state == 4
+    assert server.url_suffix == "stable-v2"
+    assert server.main_server_id == 19
+    assert server.tag == "featured"
+    assert server.name == "Seventeen"
+    assert server.gm_url_suffix == "gm-v2"
+    assert server.explicit_ip == "10.0.0.17"
+    assert server.host == "10.0.0.17"
+    assert server.proxy_port == 9001
+    assert server.beta is True
+
+
 @pytest.mark.parametrize(
     ("payload", "message"),
     [
@@ -57,6 +106,11 @@ def test_decode_rejects_malformed_payload(payload, message):
         decode_detailed_server_list(payload)
 
 
+def test_decode_rejects_more_than_three_padding_bytes():
+    with pytest.raises(ServerListError, match="plaintext length"):
+        decode_detailed_server_list(INVALID_PADDED_LENGTH_SERVER_LIST)
+
+
 def test_resolve_server_rejects_unknown_server_id():
     data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)
 
@@ -90,3 +144,57 @@ async def test_fetch_server_info_uses_channel_specific_endpoint():
         "https://list-box.dhsf.xqhuyu.com/serverlist/110585/serverlist.txt"
     ]
     assert server.host == "s9-edge.example"
+
+
+@pytest.mark.asyncio
+async def test_fetch_server_info_does_not_disable_tls_verification(monkeypatch):
+    constructed = []
+
+    class Response:
+        content = ENCRYPTED_SERVER_LIST
+
+        def raise_for_status(self):
+            return None
+
+    class Client:
+        def __init__(self, **kwargs):
+            constructed.append(kwargs)
+
+        async def get(self, _url):
+            return Response()
+
+        async def aclose(self):
+            return None
+
+    monkeypatch.setattr("cq2_sdk.login.server_list.httpx.AsyncClient", Client)
+
+    await fetch_server_info(7, channel_id="110585", game_id=206923)
+
+    assert constructed[0].get("verify", True) is True
+
+
+def test_server_info_is_immutable_and_copies_config_versions():
+    source_versions = {"cfg_stditem": "hash-1"}
+    server = resolve_server(
+        {
+            "srvRoot": "edge.example",
+            "serverData": [{"d": 7, "p": 8041, "s": 1}],
+        },
+        7,
+        channel_id="110585",
+        game_id=206923,
+    )
+    server_with_config = type(server)(
+        server_id=server.server_id,
+        name=server.name,
+        host=server.host,
+        port=server.port,
+        config_versions=source_versions,
+    )
+
+    source_versions["cfg_stditem"] = "mutated"
+    with pytest.raises(FrozenInstanceError):
+        server_with_config.port = 9000
+    with pytest.raises(TypeError):
+        server_with_config.config_versions["new"] = "value"
+    assert dict(server_with_config.config_versions) == {"cfg_stditem": "hash-1"}