feat: implement cq2 game login state machine

cq2_sdk/src/cq2_sdk/__init__.py

@@ -61,9 +61,12 @@ _LAZY = {
     # Login
     "Box996Login": ("cq2_sdk.login", "Box996Login"),
     "GameLogin": ("cq2_sdk.login", "GameLogin"),
+    "GameLoginError": ("cq2_sdk.login", "GameLoginError"),
     "BoxSession": ("cq2_sdk.login", "BoxSession"),
     "ServerInfo": ("cq2_sdk.login", "ServerInfo"),
     "GameSession": ("cq2_sdk.login", "GameSession"),
+    "ServerListError": ("cq2_sdk.login", "ServerListError"),
+    "fetch_server_info": ("cq2_sdk.login", "fetch_server_info"),
     "GeetestSolver": ("cq2_sdk.login", "GeetestSolver"),
     "GeetestResult": ("cq2_sdk.login", "GeetestResult"),
     "ManualGeetestSolver": ("cq2_sdk.login", "ManualGeetestSolver"),

cq2_sdk/src/cq2_sdk/login/__init__.py

@@ -14,7 +14,7 @@ from .box_996 import (
     BOX_API_BASE,
     USER_API_BASE,
 )
-from .game import GameLogin, GameSession
+from .game import GameLogin, GameLoginError, GameSession
 from .server_list import (
     ServerListError,
     decode_detailed_server_list,
@@ -45,6 +45,7 @@ __all__ = [
     "CaptchaRequiredError",
     "TodayOpenServer",
     "GameLogin",
+    "GameLoginError",
     "GameSession",
     "ServerListError",
     "decode_detailed_server_list",

cq2_sdk/src/cq2_sdk/login/game.py

@@ -1,39 +1,292 @@
-"""High-level CQ2 game login boundary.
-
-The TCP frame layout is verified, but a matching live session has not yet
-validated the CS2001 body, production initial sequence, body codecs, or the
-subsequent game state machine.  This module intentionally refuses to turn the
-old scripted CS/SC sequence into a purported live game login.
-"""
+"""High-level CQ2 role and world-server login state machine."""
 from __future__ import annotations
 
+import asyncio
 from dataclasses import dataclass
+import json
+import random
+from typing import Callable
 
 from cq2_sdk.login.models import BoxSession, ServerInfo
 from cq2_sdk.lua_bridge import Cq2Lua
+from cq2_sdk.messages import constants as C
+from cq2_sdk.transport.encode6 import encode6
+from cq2_sdk.transport.framing import FrameHeader
 from cq2_sdk.transport.tcp import TcpClient
 
 
 @dataclass
 class GameSession:
-    """Reserved result shape for a future, live-validated game login."""
-
-    char_id: int
+    char_id: str
     char_name: str
     session_id: int
+    user_id: str
+    world_ready: bool = False
+
+
+class GameLoginError(RuntimeError):
+    """A server response or malformed payload stopped the login workflow."""
+
+    def __init__(
+        self,
+        detail: str,
+        *,
+        message_id: int = 0,
+        error_code: int = 0,
+    ) -> None:
+        super().__init__(detail)
+        self.message_id = message_id
+        self.error_code = error_code
+        self.detail = detail
 
 
 class GameLogin:
-    """Reject speculative game login until a matching live session is verified."""
+    """Execute the client-observed CS2001 through SC2 login workflow."""
+
+    _ERROR_MESSAGES = {
+        C.SC_503: "token validation failed",
+        C.SC_527: "role lookup failed",
+        C.SC_522: "role creation failed",
+        C.SC_526: "world server lookup failed",
+    }
 
-    def __init__(self, tcp: TcpClient, lua: Cq2Lua | None = None) -> None:
+    def __init__(
+        self,
+        tcp: TcpClient,
+        lua: Cq2Lua | None = None,
+        *,
+        randint: Callable[[int, int], int] | None = None,
+    ) -> None:
         self.tcp = tcp
         self.lua = lua
+        self._randint = randint or random.randint
+
+    def _stamp(self) -> int:
+        return self._randint(1, 1_000_000_000)
 
-    async def login(self, session: BoxSession, server: ServerInfo) -> GameSession:
-        """Refuse the unverified CS2001-to-role workflow."""
-        del session, server
-        raise NotImplementedError(
-            "TCP wire framing is implemented, but game login requires a validated "
-            "CS2001 body, initial sequence, body codec, and state machine."
+    @staticmethod
+    def _main_server_id(server: ServerInfo) -> int:
+        return server.main_server_id or server.server_id
+
+    @staticmethod
+    def _split(body: bytes, count: int, context: str) -> list[str]:
+        try:
+            parts = body.decode("utf-8").split("/")
+        except UnicodeDecodeError as exc:
+            raise GameLoginError(f"{context} is not UTF-8") from exc
+        if len(parts) < count or any(part == "" for part in parts[:count]):
+            raise GameLoginError(f"malformed {context}")
+        return parts
+
+    async def _receive(
+        self,
+        expected: int,
+        *,
+        deadline: float,
+    ) -> tuple[FrameHeader, bytes]:
+        loop = asyncio.get_running_loop()
+        while True:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                raise TimeoutError(f"timed out waiting for SC {expected}")
+            try:
+                header, body = await self.tcp.recv(timeout=remaining)
+            except asyncio.TimeoutError as exc:
+                raise TimeoutError(f"timed out waiting for SC {expected}") from exc
+            detail = self._ERROR_MESSAGES.get(header.msg_id)
+            if detail is not None:
+                raise GameLoginError(
+                    detail,
+                    message_id=header.msg_id,
+                    error_code=header.recog or header.param1,
+                )
+            if header.msg_id == expected:
+                return header, body
+
+    async def login(
+        self,
+        session: BoxSession,
+        server: ServerInfo,
+        *,
+        char_name: str | None = None,
+        job: int = 0,
+        sex: int = 0,
+        timeout: float = 10.0,
+    ) -> GameSession:
+        """Authenticate, select or explicitly create a role, and enter world."""
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout
+        main_server_id = self._main_server_id(server)
+
+        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,
+        )
+        response, response_body = await self._receive(C.SC_529, deadline=deadline)
+        login_key, user_id = self._split(response_body, 2, "SC529 body")[:2]
+        new_login = response.recog == 1
+
+        server_info_body = (
+            encode6(login_key.encode("utf-8"))
+            if new_login
+            else str(main_server_id).encode("ascii")
+        )
+        await self.tcp.send(
+            C.CS_104,
+            0,
+            0,
+            0,
+            0,
+            server_info_body,
+        )
+        response, response_body = await self._receive(C.SC_530, deadline=deadline)
+        role_port = server.port
+        if new_login:
+            session_id = response.recog
+        else:
+            role_parts = self._split(response_body, 3, "SC530 body")
+            try:
+                role_port = int(role_parts[1])
+                session_id = int(role_parts[2])
+            except ValueError as exc:
+                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)
+
+        body = (
+            f"{session.uid}/{session_id}/{server.server_id}/"
+            f"{server.channel_id}/{server.game_id}"
+        )
+        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"),
+        )
+        _, response_body = await self._receive(C.SC_520, deadline=deadline)
+        try:
+            role_data = json.loads(response_body.decode("utf-8"))
+            roles = role_data.get("list")
+        except (UnicodeDecodeError, json.JSONDecodeError, AttributeError) as exc:
+            raise GameLoginError("malformed SC520 role JSON") from exc
+        if not isinstance(roles, list):
+            raise GameLoginError("malformed SC520 role JSON")
+
+        if not roles:
+            if not char_name:
+                raise GameLoginError("server has no role; char_name is required")
+            create_body = json.dumps(
+                {
+                    "account": session.uid,
+                    "chr": char_name,
+                    "job": job,
+                    "sex": sex,
+                    "server_name": server.server_id,
+                    "sdkid": server.channel_id,
+                    "secsdk": "0",
+                },
+                ensure_ascii=False,
+                separators=(",", ":"),
+            ).encode("utf-8")
+            await self.tcp.send(C.CS_101, 0, 0, 0, 0, create_body)
+            _, created_body = await self._receive(C.SC_521, deadline=deadline)
+            role_name, role_id = self._split(
+                created_body, 2, "SC521 body"
+            )[:2]
+        else:
+            usable = [
+                role
+                for role in roles
+                if isinstance(role, dict)
+                and not any(
+                    role.get(key) == 1
+                    for key in ("deny_ip", "deny_account", "deny_uid")
+                )
+            ]
+            if not usable:
+                raise GameLoginError("all roles are banned")
+            role = next(
+                (item for item in usable if item.get("last") == 1), usable[0]
+            )
+            role_name = str(role.get("chr") or "")
+            role_id = str(role.get("UserID") or "")
+            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"),
+        )
+        _, response_body = await self._receive(C.SC_525, deadline=deadline)
+        world_parts = self._split(response_body, 2, "SC525 body")
+        try:
+            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)
+
+        body = (
+            f"*#{session.uid}/{role_name}/{role_id}/{session_id}/"
+            f"{server.platform}/0/{server.channel_id}/{server.whitelist}/"
+            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._receive(C.SC_658, deadline=deadline)
+
+        enter_data = dict(server.config_versions)
+        enter_data.update(
+            {
+                "gameid": server.game_id,
+                "channelId": server.channel_id,
+                "packageName": server.package_name,
+                "tripartite": 0,
+            }
+        )
+        enter_body = json.dumps(
+            enter_data,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        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._receive(C.SC_2, deadline=deadline)
+        return GameSession(
+            char_id=role_id,
+            char_name=role_name,
+            session_id=session_id,
+            user_id=user_id,
+            world_ready=True,
         )

cq2_sdk/src/cq2_sdk/messages/constants.py

@@ -9,11 +9,13 @@ CS_20 = 20
 CS_100 = 100
 CS_101 = 101
 CS_103 = 103
+CS_104 = 104
 CS_107 = 107
 CS_1001 = 1001
 CS_1006 = 1006
 CS_1010 = 1010
 CS_1011 = 1011
+CS_1018 = 1018
 CS_2001 = 2001
 CS_3011 = 3011
 CS_3012 = 3012
@@ -51,7 +53,10 @@ SC_200 = 200
 SC_503 = 503
 SC_520 = 520
 SC_521 = 521
+SC_522 = 522
 SC_525 = 525
+SC_526 = 526
+SC_527 = 527
 SC_529 = 529
 SC_530 = 530
 SC_610 = 610

cq2_sdk/src/cq2_sdk/messages/registry.py

@@ -18,10 +18,13 @@ if TYPE_CHECKING:
 
 # Human-readable aliases — wire protocol name -> constant suffix (after "CS_" / "SC_")
 _CS_ALIASES: dict[str, str] = {
+    "LOGIN_SERVER": "20",
     "LOGIN": "2001",
     "LOGIN_KEY": "104",
-    "SELECT_ROLE": "100",
+    "ROLE_INFO": "100",
     "CREATE_ROLE": "101",
+    "SELECT_ROLE": "103",
+    "ENTER_WORLD": "1018",
     "RANDOM_NAME": "107",
     "USE_ITEM": "1006",
     "PICKUP": "1001",
@@ -48,6 +51,7 @@ _CS_ALIASES: dict[str, str] = {
 
 _SC_ALIASES: dict[str, str] = {
     "WORLD_INFO": "2",
+    "HELLO_TIPS": "658",
     "ACTOR_ENTER": "10",
     "ACTOR_LEAVE": "30",
     "HP_UPDATE": "31",

cq2_sdk/src/cq2_sdk/utils/encoding.py

@@ -1,86 +1,14 @@
-"""Encode6BitBuf — Python reference implementation.
-
-Reverse-engineered from the cq2 client Lua. The algorithm (per spec §6.1
-and cq2-01 §3.2):
-
-  1. Take 3 input bytes at a time.
-  2. Pack into 4 6-bit values:
-       v0 = b0[7:2]
-       v1 = b0[1:0] | b1[7:4]
-       v2 = b1[3:0] | b2[7:6]
-       v3 = b2[5:0]
-  3. Add 0x3B to each 6-bit value (maps into printable ASCII range).
-  4. Handle tail: 1 or 2 remaining bytes get 2 or 3 6-bit values.
-
-The Python reference is used only as a cross-check against the Lua
-implementation. Runtime path is via `Cq2Lua.encode6bit` (see bindings).
-"""
+"""Public Encode6BitBuf helpers backed by the verified client codec."""
 from __future__ import annotations
 
-_6BIT_OFFSET = 0x3B
+from cq2_sdk.transport.encode6 import decode6, encode6
 
 
 def encode6bit(buf: bytes) -> bytes:
-    """Encode bytes to 6-bit-per-char ASCII, matching the client Lua."""
-    if not buf:
-        return b""
-    out = bytearray()
-    n = len(buf)
-    i = 0
-    while i + 3 <= n:
-        b0, b1, b2 = buf[i], buf[i + 1], buf[i + 2]
-        v0 = (b0 >> 2) & 0x3F
-        v1 = ((b0 & 0x03) << 4) | ((b1 >> 4) & 0x0F)
-        v2 = ((b1 & 0x0F) << 2) | ((b2 >> 6) & 0x03)
-        v3 = b2 & 0x3F
-        out.append(v0 + _6BIT_OFFSET)
-        out.append(v1 + _6BIT_OFFSET)
-        out.append(v2 + _6BIT_OFFSET)
-        out.append(v3 + _6BIT_OFFSET)
-        i += 3
-    rem = n - i
-    if rem == 1:
-        b0 = buf[i]
-        v0 = (b0 >> 2) & 0x3F
-        v1 = (b0 & 0x03) << 4
-        out.append(v0 + _6BIT_OFFSET)
-        out.append(v1 + _6BIT_OFFSET)
-    elif rem == 2:
-        b0, b1 = buf[i], buf[i + 1]
-        v0 = (b0 >> 2) & 0x3F
-        v1 = ((b0 & 0x03) << 4) | ((b1 >> 4) & 0x0F)
-        v2 = (b1 & 0x0F) << 2
-        out.append(v0 + _6BIT_OFFSET)
-        out.append(v1 + _6BIT_OFFSET)
-        out.append(v2 + _6BIT_OFFSET)
-    return bytes(out)
+    """Encode bytes with the client's substitution-table 6-bit codec."""
+    return encode6(buf)
 
 
 def decode6bit(buf: bytes) -> bytes:
-    """Decode 6-bit-per-char ASCII back to bytes, matching the client Lua."""
-    if not buf:
-        return b""
-    out = bytearray()
-    n = len(buf)
-    i = 0
-    while i + 4 <= n:
-        v0 = buf[i] - _6BIT_OFFSET
-        v1 = buf[i + 1] - _6BIT_OFFSET
-        v2 = buf[i + 2] - _6BIT_OFFSET
-        v3 = buf[i + 3] - _6BIT_OFFSET
-        out.append(((v0 & 0x3F) << 2) | ((v1 >> 4) & 0x03))
-        out.append(((v1 & 0x0F) << 4) | ((v2 >> 2) & 0x0F))
-        out.append(((v2 & 0x03) << 6) | (v3 & 0x3F))
-        i += 4
-    rem = n - i
-    if rem == 2:
-        v0 = buf[i] - _6BIT_OFFSET
-        v1 = buf[i + 1] - _6BIT_OFFSET
-        out.append(((v0 & 0x3F) << 2) | ((v1 >> 4) & 0x03))
-    elif rem == 3:
-        v0 = buf[i] - _6BIT_OFFSET
-        v1 = buf[i + 1] - _6BIT_OFFSET
-        v2 = buf[i + 2] - _6BIT_OFFSET
-        out.append(((v0 & 0x3F) << 2) | ((v1 >> 4) & 0x03))
-        out.append(((v1 & 0x0F) << 4) | ((v2 >> 2) & 0x0F))
-    return bytes(out)
+    """Decode bytes produced by :func:`encode6bit`."""
+    return decode6(buf)

cq2_sdk/tests/unit/test_encoding.py

@@ -48,10 +48,5 @@ def test_roundtrip_various_lengths(inp):
 
 
 def test_encode_known_vector_abc():
-    # Verify the bit packing manually for "abc" (0x61 0x62 0x63):
-    #   b0=0x61=01100001, b1=0x62=01100010, b2=0x63=01100011
-    #   v0 = 0x61 >> 2         = 0x18 -> 0x18 + 0x3B = 0x53 ('S')
-    #   v1 = (0x61&3)<<4 | 0x62>>4 = 0x16 -> 0x16 + 0x3B = 0x51 ('Q')
-    #   v2 = (0x62&0xF)<<2 | 0x63>>6 = 0x09 -> 0x09 + 0x3B = 0x44 ('D')
-    #   v3 = 0x63 & 0x3F = 0x23 -> 0x23 + 0x3B = 0x5E ('^')
-    assert encode6bit(b"abc") == b"SQD^"
+    # Literal derived from the extracted substitution table and ARM64 codec.
+    assert encode6bit(b"abc") == b"xK_H"

cq2_sdk/tests/unit/test_game_login.py

@@ -1,15 +1,443 @@
+import asyncio
+import json
+
 import pytest
 
-from cq2_sdk.login.game import GameLogin
+from cq2_sdk.login.game import GameLogin, GameLoginError
 from cq2_sdk.login.models import BoxSession, ServerInfo
+from cq2_sdk.messages import constants as C
+from cq2_sdk.transport.framing import (
+    WIRE_HEADER_SIZE,
+    FrameHeader,
+    decode_header,
+    encode_frame,
+)
+from cq2_sdk.transport.tcp import TcpClient
+
+
+async def _read_frame(reader):
+    header = decode_header(await reader.readexactly(WIRE_HEADER_SIZE))
+    body = await reader.readexactly(header.body_len) if header.body_len else b""
+    return header, body
+
+
+def _response(msg_id, *, recog=0, param1=0, param2=0, param3=0, body=b"", sequence=0):
+    return encode_frame(
+        msg_id,
+        recog,
+        param1,
+        param2,
+        param3,
+        body,
+        sequence=sequence,
+    )
+
+
+def _header(msg_id, *, recog=0, param1=0, param2=0, param3=0):
+    return FrameHeader(
+        param3=param3,
+        msg_id=msg_id,
+        recog=recog,
+        param1=param1,
+        param2=param2,
+        sdata=0,
+        codec=0,
+        body_len=0,
+        sequence=0,
+    )
+
+
+class _ScriptedTcp:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.sent = []
+        self.reconnections = []
+        self.sdata = 0
+
+    async def send(self, *args):
+        self.sent.append(args)
+
+    async def recv(self, timeout):
+        if not self.responses:
+            await asyncio.sleep(timeout)
+            raise asyncio.TimeoutError
+        return self.responses.pop(0)
+
+    def set_sdata(self, value):
+        self.sdata = value
+
+    async def reconnect(self, host, port):
+        self.reconnections.append((host, port))
+
+
+def _server(**changes):
+    values = {
+        "server_id": 7,
+        "main_server_id": 9,
+        "name": "Seven",
+        "host": "game.example",
+        "port": 8041,
+        "channel_id": "110585",
+        "game_id": 206923,
+        "platform": 2,
+        "package_name": "pkg.name",
+        "screen_width": 1136,
+        "screen_height": 639,
+        "device_code": 98395,
+    }
+    values.update(changes)
+    return ServerInfo(**values)
+
+
+@pytest.mark.asyncio
+async def test_new_login_existing_role_enters_world():
+    role_frames = []
+    world_frames = []
+    world_done = asyncio.Event()
+
+    async def world_handler(reader, writer):
+        world_frames.append(await _read_frame(reader))
+        writer.write(
+            _response(
+                C.SC_658,
+                body=b'{"UserID":"role-123","Msg":"hello","ver":"v1"}',
+            )
+        )
+        await writer.drain()
+        world_frames.append(await _read_frame(reader))
+        writer.write(_response(C.SC_2, body=b'{"MapID":3}', sequence=1))
+        await writer.drain()
+        world_done.set()
+        writer.close()
+        await writer.wait_closed()
+
+    world_server = await asyncio.start_server(world_handler, "127.0.0.1", 0)
+    world_port = world_server.sockets[0].getsockname()[1]
+
+    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=json.dumps(
+                    {
+                        "list": [
+                            {
+                                "chr": "Hero",
+                                "UserID": "role-123",
+                                "last": 1,
+                                "deny_ip": 0,
+                                "deny_account": 0,
+                                "deny_uid": 0,
+                            }
+                        ]
+                    },
+                    separators=(",", ":"),
+                ).encode(),
+                sequence=2,
+            ),
+            _response(C.SC_525, body=f"ignored/{world_port}".encode(), sequence=3),
+        ]
+        for response in responses:
+            role_frames.append(await _read_frame(reader))
+            writer.write(response)
+            await writer.drain()
+        writer.close()
+        await writer.wait_closed()
+
+    role_server = await asyncio.start_server(role_handler, "127.0.0.1", 0)
+    role_port = role_server.sockets[0].getsockname()[1]
+    tcp = TcpClient("127.0.0.1", role_port)
+    await tcp.connect()
+    stamps = iter([111, 222, 333])
+    login = GameLogin(tcp, randint=lambda _low, _high: next(stamps))
+    server = ServerInfo(
+        server_id=7,
+        main_server_id=9,
+        name="Seven",
+        host="127.0.0.1",
+        port=role_port,
+        channel_id="110585",
+        game_id=206923,
+        platform=2,
+        package_name="pkg.name",
+        screen_width=1136,
+        screen_height=639,
+        device_code=98395,
+        config_versions={"cfg_stditem": "hash"},
+    )
+    try:
+        result = await login.login(
+            BoxSession(uid="account-1", token="token-xyz", sdkid="", signkey=""),
+            server,
+            timeout=2.0,
+        )
+        await asyncio.wait_for(world_done.wait(), timeout=1.0)
+    finally:
+        await tcp.close()
+        role_server.close()
+        world_server.close()
+        await role_server.wait_closed()
+        await world_server.wait_closed()
+
+    assert result.char_id == "role-123"
+    assert result.char_name == "Hero"
+    assert result.session_id == 149
+    assert result.user_id == "user-42"
+    assert result.world_ready is True
+
+    (cs2001, body2001), (cs104, body104), (cs100, body100), (cs103, body103) = role_frames
+    assert (cs2001.msg_id, cs2001.recog, cs2001.param1) == (C.CS_2001, 900000, 111)
+    assert (cs2001.param2, cs2001.param3, cs2001.sdata, cs2001.sequence) == (
+        9,
+        role_port,
+        0,
+        0,
+    )
+    assert body2001 == b"account-1/pGgZXJ?^LvhA/PaDP<wt?/1/0"
+    assert (cs104.msg_id, cs104.sdata, cs104.sequence) == (C.CS_104, 0, 1)
+    assert body104 == b"VwgIVZ?^sc>g"
+    assert (cs100.msg_id, cs100.recog, cs100.param1) == (C.CS_100, 900000, 222)
+    assert (cs100.param2, cs100.param3, cs100.sdata, cs100.sequence) == (
+        9,
+        role_port,
+        149,
+        2,
+    )
+    assert body100 == b"account-1/149/7/110585/206923/login-key"
+    assert (cs103.msg_id, cs103.sdata, cs103.sequence) == (C.CS_103, 149, 3)
+    assert body103 == b"account-1/Hero"
+
+    (cs20, body20), (cs1018, body1018) = world_frames
+    assert (cs20.msg_id, cs20.recog, cs20.param1) == (C.CS_20, 900000, 333)
+    assert (cs20.param2, cs20.param3, cs20.sdata, cs20.sequence) == (
+        9,
+        world_port,
+        149,
+        0,
+    )
+    assert body20 == b"*#account-1/Hero/role-123/149/2/0/110585/0/0/0/1/0"
+    assert (cs1018.msg_id, cs1018.recog, cs1018.param1) == (C.CS_1018, 2, 0)
+    assert (cs1018.param2, cs1018.param3, cs1018.sdata, cs1018.sequence) == (
+        41_878_640,
+        98_395,
+        149,
+        1,
+    )
+    assert json.loads(body1018) == {
+        "cfg_stditem": "hash",
+        "gameid": 206923,
+        "channelId": "110585",
+        "packageName": "pkg.name",
+        "tripartite": 0,
+    }
+
+
+@pytest.mark.asyncio
+async def test_empty_role_list_requires_explicit_name_without_sending_cs101():
+    tcp = _ScriptedTcp(
+        [
+            (_header(C.SC_529, recog=1), b"login-key/user-42"),
+            (_header(C.SC_530, recog=149), b""),
+            (_header(C.SC_520), b'{"list":[]}'),
+        ]
+    )
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    with pytest.raises(GameLoginError, match="char_name is required"):
+        await login.login(
+            BoxSession(uid="account-1", token="token-xyz", sdkid="", signkey=""),
+            _server(),
+            timeout=1.0,
+        )
+
+    assert [sent[0] for sent in tcp.sent] == [C.CS_2001, C.CS_104, C.CS_100]
+
+
+@pytest.mark.asyncio
+async def test_empty_role_list_creates_named_role_then_enters_world():
+    tcp = _ScriptedTcp(
+        [
+            (_header(C.SC_529, recog=1), b"login-key/user-42"),
+            (_header(C.SC_530, recog=149), b""),
+            (_header(C.SC_520), b'{"list":[]}'),
+            (_header(C.SC_521), "新手/role-new".encode()),
+            (_header(C.SC_525), b"ignored/8043"),
+            (_header(C.SC_658), b'{"UserID":"role-new"}'),
+            (_header(C.SC_2), b"{}"),
+        ]
+    )
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    result = await login.login(
+        BoxSession(uid="account-1", token="token-xyz", sdkid="", signkey=""),
+        _server(),
+        char_name="新手",
+        job=1,
+        sex=0,
+        timeout=1.0,
+    )
+
+    assert result.char_id == "role-new"
+    assert result.char_name == "新手"
+    assert [sent[0] for sent in tcp.sent] == [
+        C.CS_2001,
+        C.CS_104,
+        C.CS_100,
+        C.CS_101,
+        C.CS_103,
+        C.CS_20,
+        C.CS_1018,
+    ]
+    create_body = tcp.sent[3][5]
+    assert json.loads(create_body) == {
+        "account": "account-1",
+        "chr": "新手",
+        "job": 1,
+        "sex": 0,
+        "server_name": 7,
+        "sdkid": "110585",
+        "secsdk": "0",
+    }
+
+
+@pytest.mark.asyncio
+async def test_old_login_reconnects_from_sc530_before_role_lookup():
+    roles = {
+        "list": [
+            {
+                "chr": "Hero",
+                "UserID": "role-old",
+                "last": 1,
+                "deny_ip": 0,
+                "deny_account": 0,
+                "deny_uid": 0,
+            }
+        ]
+    }
+    tcp = _ScriptedTcp(
+        [
+            (_header(C.SC_529, recog=0), b"unused/user-old"),
+            (_header(C.SC_530), b"ignored/7777/151"),
+            (_header(C.SC_520), json.dumps(roles).encode()),
+            (_header(C.SC_525), b"ignored/8888"),
+            (_header(C.SC_658), b"{}"),
+            (_header(C.SC_2), b"{}"),
+        ]
+    )
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    result = await login.login(
+        BoxSession(uid="account-1", token="token-xyz", sdkid="", signkey=""),
+        _server(),
+        timeout=1.0,
+    )
+
+    assert result.session_id == 151
+    assert tcp.reconnections == [("game.example", 7777), ("game.example", 8888)]
+    assert tcp.sent[1][0] == C.CS_104
+    assert tcp.sent[1][5] == b"9"
+    assert tcp.sent[2][0] == C.CS_100
+    assert tcp.sent[2][4] == 7777
+    assert tcp.sent[2][5] == b"account-1/151/7/110585/206923"
+
+
+@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_529, recog=1), b"key/user"),
+                (_header(C.SC_530, recog=149), b""),
+                (_header(C.SC_527, 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_520), b'{"list":[]}'),
+                (_header(C.SC_522, recog=-10), b""),
+            ],
+            "New",
+            C.SC_522,
+            -10,
+        ),
+        (
+            [
+                (_header(C.SC_529, recog=1), b"key/user"),
+                (_header(C.SC_530, recog=149), b""),
+                (
+                    _header(C.SC_520),
+                    b'{"list":[{"chr":"Hero","UserID":"r1","last":1}]}',
+                ),
+                (_header(C.SC_526, recog=99), b""),
+            ],
+            None,
+            C.SC_526,
+            99,
+        ),
+    ],
+)
+async def test_known_login_errors_preserve_message_and_error_code(
+    responses, char_name, message_id, error_code
+):
+    tcp = _ScriptedTcp(responses)
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    with pytest.raises(GameLoginError) as caught:
+        await login.login(
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(),
+            char_name=char_name,
+            timeout=0.05,
+        )
+
+    assert caught.value.message_id == message_id
+    assert caught.value.error_code == error_code
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("responses", "message"),
+    [
+        ([(_header(C.SC_529, recog=1), b"missing-separator")], "SC529"),
+        (
+            [
+                (_header(C.SC_529, recog=1), b"key/user"),
+                (_header(C.SC_530, recog=149), b""),
+                (_header(C.SC_520), b"{"),
+            ],
+            "SC520",
+        ),
+    ],
+)
+async def test_malformed_login_responses_are_protocol_errors(responses, message):
+    tcp = _ScriptedTcp(responses)
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
+
+    with pytest.raises(GameLoginError, match=message):
+        await login.login(
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(),
+            timeout=0.05,
+        )
 
 
 @pytest.mark.asyncio
-async def test_login_rejects_unverified_game_protocol():
-    login = GameLogin(tcp=object())  # type: ignore[arg-type]
+async def test_unrelated_messages_are_ignored_until_shared_deadline():
+    tcp = _ScriptedTcp([(_header(9999), b"unrelated")])
+    login = GameLogin(tcp, randint=lambda _low, _high: 1)
 
-    with pytest.raises(NotImplementedError, match="CS2001 body"):
+    with pytest.raises(TimeoutError, match="SC 529"):
         await login.login(
-            BoxSession(uid="u1", token="t1", sdkid="sdk1", signkey="sk1"),
-            ServerInfo(server_id="s1", name="n", host="h", port=1),
+            BoxSession(uid="u", token="t", sdkid="", signkey=""),
+            _server(),
+            timeout=0.01,
         )

cq2_sdk/tests/unit/test_message_registry.py

@@ -37,6 +37,19 @@ def test_registry_falls_back_to_hardcoded():
     assert reg.sc("BAG_FULL") == 874
 
 
+def test_registry_login_aliases_match_verified_state_machine():
+    reg = MessageRegistry()
+
+    assert reg.cs("LOGIN_SERVER") == 20
+    assert reg.cs("ROLE_INFO") == 100
+    assert reg.cs("CREATE_ROLE") == 101
+    assert reg.cs("SELECT_ROLE") == 103
+    assert reg.cs("LOGIN_KEY") == 104
+    assert reg.cs("ENTER_WORLD") == 1018
+    assert reg.sc("HELLO_TIPS") == 658
+    assert reg.sc("WORLD_INFO") == 2
+
+
 def test_registry_lua_overrides_hardcoded():
     """If Lua returns a value, it takes precedence; otherwise fallback is used."""
     class PartialLua:

cq2_sdk/tests/unit/test_public_api.py

@@ -14,7 +14,15 @@ def test_public_api_exports():
         find_server, select_zone, launch_game, wait_for_game_ready, fetch_srvlist,
         ServerGroup, Zone, GameLaunchInfo,
     )
-    from cq2_sdk.login import Box996Login, GameLogin, BoxSession, GeetestSolver
+    from cq2_sdk.login import (
+        Box996Login,
+        GameLogin,
+        GameLoginError,
+        BoxSession,
+        GeetestSolver,
+        ServerListError,
+        fetch_server_info,
+    )
     from cq2_sdk.login.geetest import ManualGeetestSolver, MockGeetestSolver, GeetestResult
     from cq2_sdk.transport import TcpClient, HttpClient
     from cq2_sdk.lua_bridge import LuaRuntime, Cq2Lua, LuaLoader
@@ -30,3 +38,6 @@ def test_public_api_exports():
     assert Cq2Client is not None
     assert FlowResult is not None
     assert Task is not None
+    assert GameLoginError is not None
+    assert ServerListError is not None
+    assert fetch_server_info is not None