fix: enforce TCP framing boundary

cq2_sdk/src/cq2_sdk/login/game.py

@@ -1,120 +1,39 @@
-"""cq2 game TCP 3-step login (CS 2001→529→104→530→100→520 + create role 101/107).
+"""High-level CQ2 game login boundary.
 
-Per spec §1 / cq2-01 §2:
-  1. CS 2001 -> SC 529: send account creds, get loginKey
-  2. CS 104  -> SC 530: send loginKey (encoded), get SessionId
-  3. CS 100  -> SC 520: send account/sessionId, get role list
-  4. If role list empty: CS 101 -> SC 521 (create role), then re-fetch
-
-Live verification requires real game server + XXTEA frame encryption
-(spec §6.2 — algorithm not fully reverse-engineered; placeholder impl
-in transport/xxtea.py). This module is fully unit-tested with a scripted
-echo server that returns canned SC responses.
+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.
 """
 from __future__ import annotations
-import json
+
 from dataclasses import dataclass
 
-from cq2_sdk.transport.tcp import TcpClient
 from cq2_sdk.login.models import BoxSession, ServerInfo
-from cq2_sdk.lua_bridge import Cq2Lua, LuaRuntime
+from cq2_sdk.lua_bridge import Cq2Lua
+from cq2_sdk.transport.tcp import TcpClient
 
 
 @dataclass
 class GameSession:
+    """Reserved result shape for a future, live-validated game login."""
+
     char_id: int
     char_name: str
     session_id: int
 
 
 class GameLogin:
+    """Reject speculative game login until a matching live session is verified."""
+
     def __init__(self, tcp: TcpClient, lua: Cq2Lua | None = None) -> None:
         self.tcp = tcp
         self.lua = lua
 
-    async def _send_and_recv(
-        self,
-        msg_id: int,
-        recog: int,
-        param1: int,
-        param2: int,
-        param3: int,
-        body: bytes,
-        *,
-        expect: int,
-        timeout: float = 10.0,
-    ):
-        await self.tcp.send(msg_id, recog, param1, param2, param3, body)
-        hdr, resp_body = await self.tcp.recv(timeout=timeout)
-        if hdr.msg_id != expect:
-            raise RuntimeError(
-                f"Expected SC {expect}, got SC {hdr.msg_id} (recog={hdr.recog})"
-            )
-        return hdr, resp_body
-
     async def login(self, session: BoxSession, server: ServerInfo) -> GameSession:
-        """3-step login. Creates a role if the account has none."""
-        # Step 1: CS 2001 -> SC 529
-        # body: "account/password(Encode6BitBuf)/otpType/otpCode"
-        attach = f"{session.uid}/{session.token}/{server.server_id}/sms/1234"
-        hdr, body = await self._send_and_recv(
-            2001, recog=900000, param1=0, param2=0, param3=0,
-            body=attach.encode(), expect=529,
-        )
-        # SC 529 body: "loginKey/userID"
-        decoded = body.decode("utf-8", errors="replace")
-        parts = decoded.split("/", 1)
-        login_key = parts[0] if parts else ""
-        user_id = parts[1] if len(parts) > 1 else ""
-
-        # Step 2: CS 104 -> SC 530
-        # body: Encode6BitBuf(loginKey) in new login mode
-        if self.lua is not None:
-            enc = self.lua.encode6bit(login_key)
-            body_104 = enc.encode("utf-8") if isinstance(enc, str) else enc
-        else:
-            body_104 = login_key.encode("utf-8")  # fallback
-        hdr, _ = await self._send_and_recv(
-            104, 0, 0, 0, 0, body=body_104, expect=530,
-        )
-        session_id = hdr.recog
-
-        # Step 3: CS 100 -> SC 520 (role list)
-        attach_100 = f"{session.uid}/{login_key}/{server.server_id}/{session.sdkid}/1"
-        hdr, body = await self._send_and_recv(
-            100, 0, 0, 0, 0, body=attach_100.encode(), expect=520,
-        )
-        roles = json.loads(body).get("list", []) if body else []
-        if roles:
-            r = roles[0]
-            return GameSession(
-                char_id=int(r.get("chr", 0)),
-                char_name=str(r.get("name", "")),
-                session_id=int(session_id),
-            )
-
-        # No role yet: create one
-        await self._send_and_recv(
-            101, 0, 0, 0, 0,
-            body=json.dumps({
-                "account": session.uid,
-                "chr": "auto",
-                "job": 0,
-                "sex": 0,
-                "server_name": server.server_id,
-                "sdkid": session.sdkid,
-                "secsdk": "0",
-            }).encode(),
-            expect=521,
-        )
-        # Re-fetch role list
-        hdr, body = await self._send_and_recv(
-            100, 0, 0, 0, 0, body=attach_100.encode(), expect=520,
-        )
-        roles = json.loads(body).get("list", []) if body else []
-        r = roles[0] if roles else {}
-        return GameSession(
-            char_id=int(r.get("chr", 0)),
-            char_name=str(r.get("name", "")),
-            session_id=int(session_id),
+        """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."
         )

cq2_sdk/src/cq2_sdk/transport/framing.py

@@ -13,6 +13,8 @@ PLAIN_HEADER_FORMAT = "<9i"
 PLAIN_HEADER_SIZE = struct.calcsize(PLAIN_HEADER_FORMAT)
 WIRE_HEADER_SIZE = 48
 HEADER_FORMAT = PLAIN_HEADER_FORMAT
+# Deprecated compatibility alias: this is the decoded 36-byte plain format;
+# use PLAIN_HEADER_FORMAT instead of pairing it with HEADER_SIZE (48 bytes).
 HEADER_SIZE = WIRE_HEADER_SIZE
 
 

cq2_sdk/src/cq2_sdk/transport/tcp.py

@@ -29,7 +29,8 @@ class TcpClient:
         self._reader: Optional[asyncio.StreamReader] = None
         self._writer: Optional[asyncio.StreamWriter] = None
         self._buffer = bytearray()
-        self._msg_queue: asyncio.Queue[tuple[FrameHeader, bytes]] = asyncio.Queue()
+        self._msg_queue: asyncio.Queue[tuple[FrameHeader, bytes] | Exception] = asyncio.Queue()
+        self._recv_error: Exception | None = None
         self._recv_task: Optional[asyncio.Task] = None
         self._heartbeat_task: Optional[asyncio.Task] = None
         self._closed = False
@@ -76,7 +77,9 @@ class TcpClient:
                     await self._msg_queue.put((_hdr, body))
         except asyncio.CancelledError:
             return
-        except Exception:
+        except Exception as exc:
+            self._recv_error = exc
+            self._msg_queue.put_nowait(exc)
             return
 
     async def send(
@@ -104,7 +107,12 @@ class TcpClient:
         await self._writer.drain()
 
     async def recv(self, timeout: float = 30.0) -> tuple[FrameHeader, bytes]:
-        return await asyncio.wait_for(self._msg_queue.get(), timeout=timeout)
+        if self._recv_error is not None:
+            raise self._recv_error
+        result = await asyncio.wait_for(self._msg_queue.get(), timeout=timeout)
+        if isinstance(result, Exception):
+            raise result
+        return result
 
     async def start_heartbeat(self, interval: float = 10.0) -> None:
         if self._heartbeat_task and not self._heartbeat_task.done():

cq2_sdk/src/cq2_sdk/transport/xxtea.py

@@ -1,10 +1,8 @@
-"""XXTEA encryption wrapper for cq2 TCP frame encryption.
+"""Generic XXTEA helper retained for non-TCP resource/history experiments.
 
-NOTE: The exact algorithm and SessionId-derived key derivation is not fully
-reverse-engineered (see spec §6.2 / cq2-01 §2). On first live connection,
-capture frames and verify the cipher matches what the C++ netClient
-produces. This module ships a standard XXTEA wrapper as a placeholder;
-the production runtime path must be validated end-to-end before use.
+Verified CQ2 game TCP frames use ``Encode6BitBuf(header36) + body`` and do
+not use XXTEA.  This module must not be used as evidence for, or added to,
+the game TCP transport path.
 """
 from __future__ import annotations
 import xxtea  # type: ignore[import-untyped]

cq2_sdk/tests/unit/test_game_login.py

@@ -1,167 +1,15 @@
-import asyncio
 import pytest
-from cq2_sdk.transport.tcp import TcpClient
-from cq2_sdk.transport.framing import WIRE_HEADER_SIZE, decode_header, encode_frame
-from cq2_sdk.login.game import GameLogin, GameSession
-from cq2_sdk.login.models import BoxSession, ServerInfo
-
-
-SCRIPTED_RESPONSES = [
-    # CS 2001 -> SC 529
-    (529, 1, 0, 0, 0, b"loginKey-abc/user-1"),
-    # CS 104  -> SC 530
-    (530, 999, 0, 0, 0, b"ok"),
-    # CS 100  -> SC 520 (empty role list -> need to create)
-    (520, 0, 0, 0, 0, b'{"list":[]}'),
-    # CS 101  -> SC 521 (create role success)
-    (521, 0, 0, 0, 0, b"auto/12345"),
-    # CS 100  -> SC 520 (re-fetch, now has role)
-    (520, 0, 0, 0, 0, b'{"list":[{"chr":12345,"name":"auto"}]}'),
-]
-
-
-@pytest.fixture
-async def scripted_server():
-    state = {"i": 0, "server": None}
 
-    async def handler(reader, writer):
-        try:
-            while True:
-                try:
-                    hdr_raw = await reader.readexactly(WIRE_HEADER_SIZE)
-                except asyncio.IncompleteReadError:
-                    break
-                header = decode_header(hdr_raw)
-                body = await reader.readexactly(header.body_len) if header.body_len else b""
-                # Respond with next scripted SC
-                idx = state["i"]
-                if idx < len(SCRIPTED_RESPONSES):
-                    mid, r, p1o, p2o, p3o, b = SCRIPTED_RESPONSES[idx]
-                    state["i"] = idx + 1
-                    resp = encode_frame(mid, r, p1o, p2o, p3o, b)
-                    writer.write(resp)
-                    await writer.drain()
-                else:
-                    await asyncio.sleep(0.01)
-        finally:
-            writer.close()
-
-    server = await asyncio.start_server(handler, "127.0.0.1", 0)
-    state["server"] = server
-    port = server.sockets[0].getsockname()[1]
-    yield port
-    server.close()
-    await server.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_login_full_flow_creates_role(scripted_server):
-    tcp = TcpClient("127.0.0.1", scripted_server)
-    await tcp.connect()
-    try:
-        gl = GameLogin(tcp)
-        sess = await gl.login(
-            BoxSession(uid="u1", token="t1", sdkid="sdk1", signkey="sk1"),
-            ServerInfo(server_id="s1", name="n", host="h", port=1),
-        )
-        assert isinstance(sess, GameSession)
-        assert sess.session_id == 999  # recog from SC 530
-        assert sess.char_id == 12345  # from re-fetched role list
-        assert sess.char_name == "auto"
-    finally:
-        await tcp.close()
+from cq2_sdk.login.game import GameLogin
+from cq2_sdk.login.models import BoxSession, ServerInfo
 
 
 @pytest.mark.asyncio
-async def test_login_with_existing_role(scripted_server):
-    """Server returns a non-empty role list on first CS 100, skipping the create path."""
-    # Override the scripted server to skip creation
-    state = {"i": 0}
-    responses = [
-        (529, 1, 0, 0, 0, b"lk/uid"),
-        (530, 42, 0, 0, 0, b"ok"),
-        (520, 0, 0, 0, 0, b'{"list":[{"chr":999,"name":"existing"}]}'),
-    ]
+async def test_login_rejects_unverified_game_protocol():
+    login = GameLogin(tcp=object())  # type: ignore[arg-type]
 
-    async def handler(reader, writer):
-        try:
-            while True:
-                try:
-                    hdr_raw = await reader.readexactly(WIRE_HEADER_SIZE)
-                except asyncio.IncompleteReadError:
-                    break
-                header = decode_header(hdr_raw)
-                if header.body_len:
-                    await reader.readexactly(header.body_len)
-                idx = state["i"]
-                if idx < len(responses):
-                    mid, r, p1, p2, p3, b = responses[idx]
-                    state["i"] = idx + 1
-                    resp = encode_frame(mid, r, p1, p2, p3, b)
-                    writer.write(resp)
-                    await writer.drain()
-                else:
-                    await asyncio.sleep(0.01)
-        finally:
-            writer.close()
-
-    server = await asyncio.start_server(handler, "127.0.0.1", 0)
-    port = server.sockets[0].getsockname()[1]
-    try:
-        tcp = TcpClient("127.0.0.1", port)
-        await tcp.connect()
-        gl = GameLogin(tcp)
-        sess = await gl.login(
-            BoxSession(uid="u", token="t", sdkid="s", signkey="k"),
+    with pytest.raises(NotImplementedError, match="CS2001 body"):
+        await login.login(
+            BoxSession(uid="u1", token="t1", sdkid="sdk1", signkey="sk1"),
             ServerInfo(server_id="s1", name="n", host="h", port=1),
         )
-        assert sess.session_id == 42
-        assert sess.char_id == 999
-        assert sess.char_name == "existing"
-        await tcp.close()
-    finally:
-        server.close()
-        await server.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_login_wrong_response_raises():
-    """If server returns unexpected msg_id, login should raise RuntimeError."""
-    state = {"i": 0}
-
-    async def handler(reader, writer):
-        try:
-            while True:
-                try:
-                    hdr_raw = await reader.readexactly(WIRE_HEADER_SIZE)
-                except asyncio.IncompleteReadError:
-                    break
-                header = decode_header(hdr_raw)
-                if header.body_len:
-                    await reader.readexactly(header.body_len)
-                # Return wrong msg_id (9999 instead of 529)
-                if state["i"] == 0:
-                    resp = encode_frame(9999, 0, 0, 0, 0, b"")
-                    writer.write(resp)
-                    await writer.drain()
-                    state["i"] += 1
-                else:
-                    await asyncio.sleep(0.01)
-        finally:
-            writer.close()
-
-    server = await asyncio.start_server(handler, "127.0.0.1", 0)
-    port = server.sockets[0].getsockname()[1]
-    try:
-        tcp = TcpClient("127.0.0.1", port)
-        await tcp.connect()
-        gl = GameLogin(tcp)
-        with pytest.raises(RuntimeError, match="Expected SC 529"):
-            await gl.login(
-                BoxSession(uid="u", token="t", sdkid="s", signkey="k"),
-                ServerInfo(server_id="s1", name="n", host="h", port=1),
-            )
-        await tcp.close()
-    finally:
-        server.close()
-        await server.wait_closed()

cq2_sdk/tests/unit/test_message_registry.py

@@ -109,9 +109,9 @@ async def test_wait_for_sc_returns_matching():
         def __init__(self):
             self.queue = asyncio.Queue()
             # Queue some frames: 999 (drop), 55 (match), 100 (drop)
-            self.queue.put_nowait((FrameHeader(msg_id=999, recog=0, param1=0, param2=0, param3=0, body_len=0), b""))
-            self.queue.put_nowait((FrameHeader(msg_id=55, recog=0, param1=0, param2=0, param3=0, body_len=0), b"matched"))
-            self.queue.put_nowait((FrameHeader(msg_id=100, recog=0, param1=0, param2=0, param3=0, body_len=0), b""))
+            self.queue.put_nowait((FrameHeader(pre=0, msg_id=999, recog=0, param1=0, param2=0, param3=0, codec=0, body_len=0, sequence=0), b""))
+            self.queue.put_nowait((FrameHeader(pre=0, msg_id=55, recog=0, param1=0, param2=0, param3=0, codec=0, body_len=0, sequence=1), b"matched"))
+            self.queue.put_nowait((FrameHeader(pre=0, msg_id=100, recog=0, param1=0, param2=0, param3=0, codec=0, body_len=0, sequence=2), b""))
 
         async def recv(self, timeout=30.0):
             return await self.queue.get()

cq2_sdk/tests/unit/test_tcp_client.py

@@ -129,3 +129,23 @@ async def test_send_uses_and_increments_initial_sequence():
         await client.close()
         server.close()
         await server.wait_closed()
+
+
+@pytest.mark.asyncio
+async def test_recv_surfaces_unsupported_body_codec():
+    async def handler(reader, writer):
+        writer.write(encode_frame(529, 0, 0, 0, 0, b"compressed", codec=1))
+        await writer.drain()
+        writer.close()
+
+    server = await asyncio.start_server(handler, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    client = TcpClient("127.0.0.1", port)
+    await client.connect()
+    try:
+        with pytest.raises(ValueError, match="unsupported body codec: 1"):
+            await client.recv(timeout=1.0)
+    finally:
+        await client.close()
+        server.close()
+        await server.wait_closed()