feat: resolve cq2 tcp servers

cq2_sdk/src/cq2_sdk/login/__init__.py

@@ -15,6 +15,12 @@ from .box_996 import (
     USER_API_BASE,
 )
 from .game import GameLogin, GameSession
+from .server_list import (
+    ServerListError,
+    decode_detailed_server_list,
+    fetch_server_info,
+    resolve_server,
+)
 from .box_auth import (
     build_auth_header,
     sign_body,
@@ -40,6 +46,10 @@ __all__ = [
     "TodayOpenServer",
     "GameLogin",
     "GameSession",
+    "ServerListError",
+    "decode_detailed_server_list",
+    "fetch_server_info",
+    "resolve_server",
     "LOGIN_BASE",
     "BOX_API_BASE",
     "USER_API_BASE",

cq2_sdk/src/cq2_sdk/login/models.py

@@ -5,15 +5,30 @@ modules. Using dataclasses (not pydantic) because they're internal
 DTOs and don't need validation.
 """
 from __future__ import annotations
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 
 @dataclass
 class ServerInfo:
-    server_id: str
+    server_id: int
     name: str
     host: str
     port: int
+    main_server_id: int | None = None
+    proxy_port: int | None = None
+    channel_id: str = "110585"
+    game_id: int = 206923
+    platform: int = 2
+    package_name: str = ""
+    screen_width: int = 1136
+    screen_height: int = 639
+    device_code: int = 0
+    whitelist: int = 0
+    certified: int = 0
+    adult: int = 0
+    is_box_login: int = 1
+    batch_messages: int = 0
+    config_versions: dict[str, str] = field(default_factory=dict)
 
 
 @dataclass

cq2_sdk/src/cq2_sdk/login/server_list.py

@@ -0,0 +1,166 @@
+"""Decode the launcher's detailed list and resolve CQ2 TCP endpoints."""
+from __future__ import annotations
+
+import base64
+import binascii
+import json
+import struct
+from typing import Any
+
+import httpx
+
+from .models import ServerInfo
+
+
+SERVER_LIST_BASE = "https://list-box.dhsf.xqhuyu.com"
+SERVER_LIST_MARKER = b"4meJnPyl"
+SERVER_LIST_KEY = b"34UFXMgtkz".ljust(16, b"\0")
+_DELTA = 0x9E3779B9
+_MASK = 0xFFFFFFFF
+
+
+class ServerListError(ValueError):
+    """The detailed server list cannot be decoded or resolved."""
+
+
+def _xxtea_decrypt(ciphertext: bytes) -> bytes:
+    if len(ciphertext) < 8 or len(ciphertext) % 4:
+        raise ServerListError("invalid XXTEA ciphertext length")
+
+    values = list(struct.unpack(f"<{len(ciphertext) // 4}I", ciphertext))
+    key = struct.unpack("<4I", SERVER_LIST_KEY)
+    rounds = 6 + 52 // len(values)
+    total = (rounds * _DELTA) & _MASK
+    y = values[0]
+
+    while total:
+        e = (total >> 2) & 3
+        for pos in range(len(values) - 1, 0, -1):
+            z = values[pos - 1]
+            mix = (
+                (((z >> 5) ^ ((y << 2) & _MASK)) +
+                 ((y >> 3) ^ ((z << 4) & _MASK)))
+                ^ ((total ^ y) + (key[(pos & 3) ^ e] ^ z))
+            ) & _MASK
+            values[pos] = (values[pos] - mix) & _MASK
+            y = values[pos]
+        z = values[-1]
+        mix = (
+            (((z >> 5) ^ ((y << 2) & _MASK)) +
+             ((y >> 3) ^ ((z << 4) & _MASK)))
+            ^ ((total ^ y) + (key[e] ^ z))
+        ) & _MASK
+        values[0] = (values[0] - mix) & _MASK
+        y = values[0]
+        total = (total - _DELTA) & _MASK
+
+    plaintext_length = values[-1]
+    available = (len(values) - 1) * 4
+    if plaintext_length <= 0 or plaintext_length > available:
+        raise ServerListError("invalid XXTEA plaintext length")
+    return struct.pack(f"<{len(values) - 1}I", *values[:-1])[:plaintext_length]
+
+
+def decode_detailed_server_list(payload: bytes) -> dict[str, Any]:
+    """Decode one Base64 + marker + cocos XXTEA server-list response."""
+    try:
+        raw = base64.b64decode(payload.strip(), validate=True)
+    except (binascii.Error, ValueError) as exc:
+        raise ServerListError("invalid server-list Base64") from exc
+    if not raw.startswith(SERVER_LIST_MARKER):
+        raise ServerListError("invalid server-list marker")
+    plaintext = _xxtea_decrypt(raw[len(SERVER_LIST_MARKER):])
+    try:
+        result = json.loads(plaintext.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ServerListError("invalid server-list JSON") from exc
+    if not isinstance(result, dict):
+        raise ServerListError("server-list JSON root must be an object")
+    return result
+
+
+def _integer(value: Any, *, default: int = 0) -> int:
+    if value in (None, ""):
+        return default
+    try:
+        return int(value)
+    except (TypeError, ValueError) as exc:
+        raise ServerListError(f"invalid integer field: {value!r}") from exc
+
+
+def resolve_server(
+    data: dict[str, Any],
+    server_id: int,
+    *,
+    channel_id: str,
+    game_id: int,
+) -> ServerInfo:
+    """Expand a compact server record into the endpoint used by GameLogin."""
+    records = data.get("serverData")
+    if not isinstance(records, list):
+        raise ServerListError("server-list is missing serverData")
+
+    wanted = int(server_id)
+    record: dict[str, Any] | None = None
+    for candidate in records:
+        if not isinstance(candidate, dict):
+            continue
+        candidate_id = _integer(candidate.get("d", candidate.get("srvid")))
+        if candidate_id == wanted:
+            record = candidate
+            break
+    if record is None:
+        raise ServerListError(f"server {wanted} is not in the detailed list")
+
+    selected_id = _integer(record.get("d", record.get("srvid")))
+    main_id = _integer(
+        record.get("m", record.get("mainsrvid")), default=selected_id
+    ) or selected_id
+    port = _integer(record.get("p", record.get("port")))
+    if port <= 0:
+        raise ServerListError(f"server {wanted} has no valid login port")
+    explicit_host = str(record.get("i", record.get("srvip")) or "")
+    root = str(data.get("srvRoot") or "")
+    host = explicit_host or (f"s{main_id}-{root}" if root else "")
+    if not host:
+        raise ServerListError(f"server {wanted} has no resolvable host")
+    proxy_port = _integer(
+        record.get("x", record.get("proxyport")), default=0
+    ) or None
+
+    return ServerInfo(
+        server_id=selected_id,
+        main_server_id=main_id,
+        name=str(record.get("n", record.get("srvname")) or selected_id),
+        host=host,
+        port=port,
+        proxy_port=proxy_port,
+        channel_id=str(channel_id),
+        game_id=int(game_id),
+    )
+
+
+async def fetch_server_info(
+    server_id: int,
+    *,
+    channel_id: str,
+    game_id: int,
+    client: httpx.AsyncClient | None = None,
+) -> ServerInfo:
+    """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)
+    try:
+        response = await http.get(url)
+        response.raise_for_status()
+        data = decode_detailed_server_list(response.content)
+        return resolve_server(
+            data,
+            server_id,
+            channel_id=channel_id,
+            game_id=game_id,
+        )
+    finally:
+        if owns_client:
+            await http.aclose()

cq2_sdk/tests/unit/test_server_list.py

@@ -0,0 +1,92 @@
+import base64
+
+import pytest
+
+from cq2_sdk.login.server_list import (
+    ServerListError,
+    decode_detailed_server_list,
+    fetch_server_info,
+    resolve_server,
+)
+
+
+ENCRYPTED_SERVER_LIST = (
+    b"NG1lSm5QeWzEk/B+5xXXfjnMHIpGB/PPbL/y7tEPyFCzWvBJq2fmPD5PiYHcKj/FZ"
+    b"+tehXBa0XEMs9YMS3E1N7HCzgBWTVsHbF5MEs9KDXnW4j1XaHjjnjRVMP2z94vFm"
+    b"LOtIDVedS9/y8FZfCpM1CigHp3uJMWqZDWQMLiTVL9RZOrKik67VPrH3MBdtPhLRb"
+    b"t00X1Uyd+JuagpL6xRX4nTFP3VdO9TWOvsw/ZqOK0C54GFugWGMAjsydz5A3A/mf"
+    b"/x+A=="
+)
+
+
+def test_decode_and_resolve_compact_server_fields():
+    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)
+
+    assert data["srvRoot"] == "edge.example"
+    server = resolve_server(data, 7, channel_id="110585", game_id=206923)
+    assert server.server_id == 7
+    assert server.main_server_id == 9
+    assert server.name == "Seven"
+    assert server.host == "s9-edge.example"
+    assert server.port == 8041
+    assert server.proxy_port == 9000
+    assert server.channel_id == "110585"
+    assert server.game_id == 206923
+
+
+def test_resolve_server_prefers_explicit_ip():
+    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)
+
+    server = resolve_server(data, 8, channel_id="c", game_id=1)
+
+    assert server.host == "1.2.3.4"
+    assert server.main_server_id == 8
+    assert server.proxy_port is None
+
+
+@pytest.mark.parametrize(
+    ("payload", "message"),
+    [
+        (b"not-base64!", "Base64"),
+        (base64.b64encode(b"wrong-sign"), "marker"),
+        (base64.b64encode(b"4meJnPyl" + b"short"), "XXTEA"),
+    ],
+)
+def test_decode_rejects_malformed_payload(payload, message):
+    with pytest.raises(ServerListError, match=message):
+        decode_detailed_server_list(payload)
+
+
+def test_resolve_server_rejects_unknown_server_id():
+    data = decode_detailed_server_list(ENCRYPTED_SERVER_LIST)
+
+    with pytest.raises(ServerListError, match="server 999"):
+        resolve_server(data, 999, channel_id="110585", game_id=206923)
+
+
+@pytest.mark.asyncio
+async def test_fetch_server_info_uses_channel_specific_endpoint():
+    seen = []
+
+    class Response:
+        content = ENCRYPTED_SERVER_LIST
+
+        def raise_for_status(self):
+            return None
+
+    class Client:
+        async def get(self, url):
+            seen.append(url)
+            return Response()
+
+    server = await fetch_server_info(
+        7,
+        channel_id="110585",
+        game_id=206923,
+        client=Client(),
+    )
+
+    assert seen == [
+        "https://list-box.dhsf.xqhuyu.com/serverlist/110585/serverlist.txt"
+    ]
+    assert server.host == "s9-edge.example"