diff --git a/README.md b/README.md index 6e01dac..b18b938 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,7 @@ from opengsq.protocols import ( StrongholdCrusader, TeamSpeak3, TrackmaniaNations, + TrackmaniaSunrise, Toxikk, UDK, Unreal2, diff --git a/docs/tests/protocols/index.rst b/docs/tests/protocols/index.rst index d51656c..3e5d3cb 100644 --- a/docs/tests/protocols/index.rst +++ b/docs/tests/protocols/index.rst @@ -42,6 +42,7 @@ Protocols Tests test_teamspeak3/index test_toxikk/index test_trackmania_nations/index + test_trackmania_sunrise/index test_unreal2/index test_ut3/index test_vcmp/index diff --git a/docs/tests/protocols/test_trackmania_nations/test_get_info.rst b/docs/tests/protocols/test_trackmania_nations/test_get_info.rst index 8568411..11f0749 100644 --- a/docs/tests/protocols/test_trackmania_nations/test_get_info.rst +++ b/docs/tests/protocols/test_trackmania_nations/test_get_info.rst @@ -6,26 +6,159 @@ Here are the results for the test method. .. code-block:: json { - "name": "Kawabonga", - "map": "B02-Race", - "players": 1, - "max_players": 6, - "game_mode": "Team", - "password_protected": true, + "name": "TMNF Dedicated", + "map": "D15-Endurance", + "players": 0, + "max_players": 32, + "game_mode": "TimeAttack", + "password_protected": false, "version": null, "environment": "Stadium", - "comment": "PC-ce9b0c", - "server_login": "", - "pc_guid": "PC-ce9b0c", - "time_limit": 0, + "comment": "Trackmania Nations Forever Dedicated Server", + "server_login": "tmnf-test", + "pc_guid": "tmnf-test", + "time_limit": 180000, "nb_laps": 0, - "spectator_slots": 0, + "spectator_slots": 32, "build_number": 0, - "private_server": true, + "private_server": false, "ladder_server": false, "status_flags": 0, "challenge_crc": 0, "public_ip": "", - "local_ip": "", - "raw_data": "9b0000008303681ac2009b0000000a0700000006000000d53d41008b5c00000b2d1d641dac2e090900000050432d636539623063050000002353525623500204000106000600094402074b617761626f6e6761075001075374616469756d0100002c6c0001ffffffff940602e09304000178030001080000004230322d526163657a710000b9020079020374040b000040070000005374616469756d110000" + "local_ip": "10.10.100.212", + "spectators": 0, + "max_spectators": 32, + "spectator_password_protected": false, + "ladder_mode": 0, + "game_mode_id": 1, + "points_limit": 0, + "pack_mask": "Stadium", + "nb_challenges": 60, + "challenges": [ + { + "name": "D15-Endurance", + "gold_time": 498250, + "copper_price": 2521, + "environment": "Stadium" + }, + { + "name": "B11-Race", + "gold_time": 33700, + "copper_price": 1990, + "environment": "Stadium" + }, + { + "name": "C07-Race", + "gold_time": 41980, + "copper_price": 1564, + "environment": "Stadium" + }, + { + "name": "D11-Acrobatic", + "gold_time": 15630, + "copper_price": 876, + "environment": "Stadium" + }, + { + "name": "C06-Speed", + "gold_time": 59030, + "copper_price": 1966, + "environment": "Stadium" + }, + { + "name": "D02-Race", + "gold_time": 56860, + "copper_price": 1713, + "environment": "Stadium" + }, + { + "name": "B03-Race", + "gold_time": 28790, + "copper_price": 1313, + "environment": "Stadium" + }, + { + "name": "B01-Race", + "gold_time": 27800, + "copper_price": 1961, + "environment": "Stadium" + }, + { + "name": "D01-endurance", + "gold_time": 163040, + "copper_price": 2227, + "environment": "Stadium" + }, + { + "name": "C14-Endurance", + "gold_time": 121280, + "copper_price": 1831, + "environment": "Stadium" + }, + { + "name": "C09-Race", + "gold_time": 53350, + "copper_price": 1630, + "environment": "Stadium" + }, + { + "name": "D07-Race", + "gold_time": 64550, + "copper_price": 2059, + "environment": "Stadium" + }, + { + "name": "C15-Speed", + "gold_time": 53920, + "copper_price": 2010, + "environment": "Stadium" + }, + { + "name": "D10-Race", + "gold_time": 59610, + "copper_price": 1533, + "environment": "Stadium" + }, + { + "name": "C02-Race", + "gold_time": 44880, + "copper_price": 2010, + "environment": "Stadium" + }, + { + "name": "A08-Endurance", + "gold_time": 68380, + "copper_price": 1563, + "environment": "Stadium" + }, + { + "name": "D13-Race", + "gold_time": 78930, + "copper_price": 2272, + "environment": "Stadium" + }, + { + "name": "C01-Race", + "gold_time": 31200, + "copper_price": 939, + "environment": "Stadium" + }, + { + "name": "B12-Race", + "gold_time": 47500, + "copper_price": 1233, + "environment": "Stadium" + }, + { + "name": "A03-Race", + "gold_time": 19840, + "copper_price": 952, + "environment": "Stadium" + } + ], + "player_list": [], + "server_address": "10.10.100.212", + "server_port": 2350, + "raw_data": "2dd4640a0a2e0909000000746d6e662d746573740500000023535256230000000000200020000e000000544d4e4620446564696361746564070000005374616469756d000000002b000000547261636b6d616e6961204e6174696f6e7320466f726576657220446564696361746564205365727665720120bf02003c140000000d0000004431352d456e647572616e63654a9a0700d90900080000004231312d52616365a4830000c60700080000004330372d52616365fca300001c06000d0000004431312d4163726f62617469630e3d00006c0300090000004330362d537065656496e60000ae0700080000004430322d526163651cde0000b10600080000004230332d5261636576700000210500080000004230312d52616365986c0000a907000d0000004430312d656e647572616e6365e07c0200b308000d0000004331342d456e647572616e6365c0d90100270700080000004330392d5261636566d000005e0600080000004430372d5261636526fc00000b0800090000004331352d5370656564a0d20000da0700080000004431302d52616365dae80000fd0500080000004330322d5261636550af0000da07000d0000004130382d456e647572616e63651c0b01001b0600080000004431332d5261636552340100e00800080000004330312d52616365e0790000ab0300080000004231322d526163658cb90000d10400080000004130332d52616365804d0000b80300010000000300000000000040070000005374616469756d" } diff --git a/docs/tests/protocols/test_trackmania_sunrise/index.rst b/docs/tests/protocols/test_trackmania_sunrise/index.rst new file mode 100644 index 0000000..641e30e --- /dev/null +++ b/docs/tests/protocols/test_trackmania_sunrise/index.rst @@ -0,0 +1,8 @@ +.. _test_trackmania_sunrise: + +test_trackmania_sunrise +======================= + +.. toctree:: + test_get_info + test_get_session diff --git a/docs/tests/protocols/test_trackmania_sunrise/test_get_info.rst b/docs/tests/protocols/test_trackmania_sunrise/test_get_info.rst new file mode 100644 index 0000000..194a18e --- /dev/null +++ b/docs/tests/protocols/test_trackmania_sunrise/test_get_info.rst @@ -0,0 +1,198 @@ +test_get_info +============= + +Here are the results for the test method. + +.. code-block:: json + + { + "name": "Sunrise LAN Server", + "map": "GoodMorning", + "players": 0, + "max_players": 32, + "game_mode": "TimeAttack", + "game_id": "TmSunrise", + "environment": "Island", + "mood": "Sunrise", + "game_tag": 7, + "protocol_version": 4, + "password_protected": false, + "spectator_password_protected": false, + "spectators": 0, + "max_spectators": 32, + "ladder_mode": 0, + "ladder_server": false, + "comment": "TrackMania Sunrise Extreme LAN", + "server_login": "DESKTOP-JBKL5J0", + "server_address": "10.10.101.4", + "server_port": 2350, + "game_mode_id": 1, + "time_limit": 300000, + "nb_laps": 0, + "points_limit": 0, + "nb_challenges": 54, + "challenges": [ + { + "name": "GoodMorning", + "gold_time": 56800, + "copper_price": 1544, + "decoration_index": 14, + "environment": "Island", + "mood": "Sunrise" + }, + { + "name": "High Tide", + "gold_time": 16000, + "copper_price": 700, + "decoration_index": 15, + "environment": "Island", + "mood": "Sunset" + }, + { + "name": "Midnight", + "gold_time": 65660, + "copper_price": 1606, + "decoration_index": 13, + "environment": "Island", + "mood": "Night" + }, + { + "name": "HighStreet", + "gold_time": 117660, + "copper_price": 1500, + "decoration_index": 3, + "environment": "Bay", + "mood": "Night" + }, + { + "name": "Suburbs", + "gold_time": 64360, + "copper_price": 1554, + "decoration_index": 2, + "environment": "Bay", + "mood": "Day" + }, + { + "name": "GrandPrix30", + "gold_time": 1280000, + "copper_price": 1549, + "decoration_index": 10, + "environment": "Coast", + "mood": "Sunset" + }, + { + "name": "QuietRide", + "gold_time": 27970, + "copper_price": 913, + "decoration_index": 9, + "environment": "Coast", + "mood": "Sunrise" + }, + { + "name": "HomeRun", + "gold_time": 71340, + "copper_price": 1452, + "decoration_index": 10, + "environment": "Coast", + "mood": "Sunset" + }, + { + "name": "BuildingRider", + "gold_time": 110840, + "copper_price": 1662, + "decoration_index": 2, + "environment": "Bay", + "mood": "Day" + }, + { + "name": "XRace09", + "gold_time": 84560, + "copper_price": 3190, + "decoration_index": 12, + "environment": "Island", + "mood": "Day" + }, + { + "name": "Bouncy Alley", + "gold_time": 12000, + "copper_price": 1048, + "decoration_index": 2, + "environment": "Bay", + "mood": "Day" + }, + { + "name": "XRace01", + "gold_time": 47490, + "copper_price": 1968, + "decoration_index": 8, + "environment": "Coast", + "mood": "Night" + }, + { + "name": "FollowTheLeader", + "gold_time": 47500, + "copper_price": 1660, + "decoration_index": 2, + "environment": "Bay", + "mood": "Day" + }, + { + "name": "XRace08", + "gold_time": 74350, + "copper_price": 3203, + "decoration_index": 2, + "environment": "Bay", + "mood": "Day" + }, + { + "name": "SkidOrDie", + "gold_time": 50690, + "copper_price": 1274, + "decoration_index": 14, + "environment": "Island", + "mood": "Sunrise" + }, + { + "name": "BeautifulDay", + "gold_time": 188250, + "copper_price": 1406, + "decoration_index": 12, + "environment": "Island", + "mood": "Day" + }, + { + "name": "Five Rows", + "gold_time": 31000, + "copper_price": 706, + "decoration_index": 9, + "environment": "Coast", + "mood": "Sunrise" + }, + { + "name": "JumpOnBrakes", + "gold_time": 47800, + "copper_price": 1398, + "decoration_index": 10, + "environment": "Coast", + "mood": "Sunset" + }, + { + "name": "NightFlight", + "gold_time": 39150, + "copper_price": 1451, + "decoration_index": 13, + "environment": "Island", + "mood": "Night" + }, + { + "name": "CarPark", + "gold_time": 37110, + "copper_price": 1233, + "decoration_index": 12, + "environment": "Island", + "mood": "Day" + } + ], + "player_list": [], + "raw_data": "0704650a0a2e090f0000004445534b544f502d4a424b4c354a300500000023535256230000000000200020001200000053756e72697365204c414e20536572766572000000001e000000547261636b4d616e69612053756e726973652045787472656d65204c414e01e093040036140000000b000000476f6f644d6f726e696e670e000000e0dd000008060000090000004869676820546964650f000000803e0000bc020000080000004d69646e696768740d0000007c000100460600000a00000048696768537472656574030000009ccb0100dc05000007000000537562757262730200000068fb0000120600000b0000004772616e645072697833300a000000008813000d0600000900000051756965745269646509000000426d00009103000007000000486f6d6552756e0a000000ac160100ac0500000d0000004275696c64696e67526964657202000000f8b001007e06000007000000585261636530390c000000504a0100760c00000c000000426f756e637920416c6c657902000000e02e00001804000007000000585261636530310800000082b90000b00700000f000000466f6c6c6f775468654c6561646572020000008cb900007c0600000700000058526163653038020000006e220100830c000009000000536b69644f724469650e00000002c60000fa0400000c00000042656175746966756c4461790c0000005adf02007e050000090000004669766520526f77730900000018790000c20200000c0000004a756d704f6e4272616b65730a000000b8ba0000760500000b0000004e69676874466c696768740d000000ee980000ab050000070000004361725061726b0c000000f6900000d1040000" + } diff --git a/docs/tests/protocols/test_trackmania_sunrise/test_get_session.rst b/docs/tests/protocols/test_trackmania_sunrise/test_get_session.rst new file mode 100644 index 0000000..10beb85 --- /dev/null +++ b/docs/tests/protocols/test_trackmania_sunrise/test_get_session.rst @@ -0,0 +1,18 @@ +test_get_session +================ + +Here are the results for the test method. + +.. code-block:: json + + { + "game_id": "TmSunrise", + "game": "TrackMania Sunrise", + "version": "1.043", + "host_name": "DESKTOP-JBKL5J0", + "application": "GameNet", + "server_address": "10.10.101.4", + "server_port": 2350, + "secondary_address": "10.10.101.4", + "secondary_port": 2350 + } diff --git a/opengsq/protocols/__init__.py b/opengsq/protocols/__init__.py index a222ef1..b64a669 100644 --- a/opengsq/protocols/__init__.py +++ b/opengsq/protocols/__init__.py @@ -39,6 +39,7 @@ from opengsq.protocols.stronghold_crusader import StrongholdCrusader from opengsq.protocols.teamspeak3 import TeamSpeak3 from opengsq.protocols.trackmania_nations import TrackmaniaNations +from opengsq.protocols.trackmania_sunrise import TrackmaniaSunrise from opengsq.protocols.toxikk import Toxikk from opengsq.protocols.udk import UDK from opengsq.protocols.unreal2 import Unreal2 @@ -90,6 +91,7 @@ "StrongholdCrusader", "TeamSpeak3", "TrackmaniaNations", + "TrackmaniaSunrise", "Toxikk", "UDK", "Unreal2", diff --git a/opengsq/protocols/trackmania_nations.py b/opengsq/protocols/trackmania_nations.py index ceffa6a..44fb62f 100644 --- a/opengsq/protocols/trackmania_nations.py +++ b/opengsq/protocols/trackmania_nations.py @@ -1,866 +1,549 @@ -from __future__ import annotations +""" +TrackMania Nations/United Forever (TmForever) native server query. -import asyncio -import struct -from typing import Optional, Dict, Any, List, Tuple -from dataclasses import dataclass -from opengsq.protocol_base import ProtocolBase -from opengsq.exceptions import InvalidPacketException -from opengsq.responses.trackmania_nations import ServerInfo +The query uses the game's own TCP protocol on the game port (default 2350), the +same way the game client fetches the details shown in its server browser. +Reverse-engineered from TrackmaniaServer.exe (2011-02-21). + +All integers are little-endian. +TCP framing:: -@dataclass -class TrackmaniaPayloadData: - """Strukturierte Daten aus dem Trackmania Payload""" + u32 length | message[length] - server_name: Optional[str] = None - srv_type: Optional[str] = None - environment: Optional[str] = None - maps: List[str] = None - players: Optional[int] = None - max_players: Optional[int] = None - game_mode: Optional[str] = None - comment: Optional[str] = None - raw_strings: List[str] = None +Message (CNetNod):: - def __post_init__(self): - if self.maps is None: - self.maps = [] - if self.raw_strings is None: - self.raw_strings = [] + u8 flags 0x80 | 0x01 compressed | 0x02 checksum | 0x04/0x08 sequenced + u8 type 0x03 = CNetFormConnectionAdmin (class 0x12010000) + u16 sequence only if flags & 0x0C + u32 checksum only if flags & 0x02: sum of the four u32 words of + HMAC-MD5(_CHECKSUM_KEY, message with the checksum zeroed) + payload if flags & 0x01: u32 uncompressed size + LZO1X stream + +CNetFormConnectionAdmin payload (the checksum is mandatory):: + + u32 version must be 7, otherwise the server answers "Please upgrade" + u32 subtype 8: switch the connection into query mode (no payload) + 7: u32 request_id (client -> server) + 6: u32 request_id | u32 size | u8[size] (server -> client) + +The subtype 6 data is the server info (CTrackManiaNetworkServerInfo), see +TrackmaniaNations.parse_server_info. str is u32 length + bytes, wstr the same +with UTF-8 (prefixed by a BOM if it contains non-ASCII characters). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import secrets +import struct +from typing import List, Optional, Tuple + +from opengsq.exceptions import InvalidPacketException, ServerNotFoundException +from opengsq.protocol_base import ProtocolBase +from opengsq.responses.trackmania_nations import Challenge, Player, ServerInfo class TrackmaniaNations(ProtocolBase): """ Trackmania Nations Protocol Implementation - Basiert auf MCP/Ghidra Reverse-Engineering - - MCP-Erkenntnisse: - - Servername bei Position 0x27 mit 4-Byte Längen-Präfix (Little Endian) - - #SRV# Marker mit 5-Byte Länge und Typ-Indikator - - Drei Haupt-Typen: SRV#f (Float), SRV#s (String), SRV#p (Packet) - - Strings verwenden 4-Byte Längen-Präfixe """ @property def full_name(self) -> str: - return "Trackmania Nations Protocol (MCP-Enhanced)" + return "Trackmania Nations Protocol" - # Standard Trackmania Nations port DEFAULT_PORT = 2350 - # TCP packets (verifiziert) - _PACKET_1 = bytes.fromhex("0e000000820399f895580700000008000000") - _PACKET_2 = bytes.fromhex("1200000082033bd464400700000007000000d53d4100") + GAME_MODES = { + 1: "TimeAttack", + 3: "Rounds", + 6: "Team", + 7: "Laps", + 8: "Stunts", + 9: "Cup", + } + + _FLAG_COMPRESSED = 0x01 + _FLAG_CHECKSUM = 0x02 + _FLAG_SEQUENCE = 0x0C + + _MESSAGE_CONNECTION_ADMIN = 0x03 + _CONNECTION_ADMIN_VERSION = 7 + _SUBTYPE_REFUSED = 1 + _SUBTYPE_INFO = 6 + _SUBTYPE_INFO_REQUEST = 7 + _SUBTYPE_QUERY_MODE = 8 + + # Request id the server puts into the reply when it has no game info yet. + _NO_INFO = 0xFFFFFFFF + + _CHECKSUM_KEY = struct.pack("<4I", 0x80D79DB8, 0xBA216B72, 0x15439598, 0xE1EC1CFA) + _MAX_MESSAGE_SIZE = 0x100000 + + # First byte of the server info: high bits 001 = valid, low bits = game. + _GAME_TAG_TRACKMANIA = 0x0D def __init__(self, host: str, port: int = DEFAULT_PORT, timeout: float = 5.0): super().__init__(host, port, timeout) async def get_info(self) -> ServerInfo: """ - Retrieves server information by sending the two TCP packets in sequence. + Retrieves the server information. :return: A ServerInfo object containing server information - :raises InvalidPacketException: If the response doesn't contain #SRV# marker + :raises ServerNotFoundException: If no TCP connection can be established + :raises InvalidPacketException: If the server does not answer like a TrackMania server """ - # Connect via TCP + request_id = secrets.randbelow(0x7FFFFFFF) + 1 + try: reader, writer = await asyncio.wait_for( asyncio.open_connection(self._host, self._port), timeout=self._timeout ) except (OSError, asyncio.TimeoutError) as e: - raise InvalidPacketException( + raise ServerNotFoundException( f"Failed to connect to {self._host}:{self._port}: {e}" - ) + ) from e try: - # Send first packet - writer.write(self._PACKET_1) + writer.write( + self.build_connection_admin(self._SUBTYPE_QUERY_MODE) + + self.build_connection_admin(self._SUBTYPE_INFO_REQUEST, request_id) + ) await writer.drain() - # Wait 200ms as specified - await asyncio.sleep(0.2) + data = await asyncio.wait_for( + self._receive_info(reader, request_id), timeout=self._timeout + ) + except asyncio.TimeoutError as e: + raise InvalidPacketException("Timeout while waiting for server info") from e + except asyncio.IncompleteReadError as e: + raise InvalidPacketException("Connection closed by the server") from e + except OSError as e: + raise InvalidPacketException(f"Connection error: {e}") from e + finally: + writer.close() + try: + await writer.wait_closed() + except OSError: + pass - # Send second packet - writer.write(self._PACKET_2) - await writer.drain() + return self.parse_server_info(data) + + async def _receive_info( + self, reader: asyncio.StreamReader, request_id: int + ) -> bytes: + while True: + length = struct.unpack(" Optional[bytes]: + reader = _Reader(payload) + version = reader.u32() + subtype = reader.u32() - # Parse using MCP-based parser - payload_data = self.parse_server_payload(response_data) - - # Convert to ServerInfo format - # Die Namens-Logik in parse_server_payload hat bereits die richtigen Namen zugeordnet - return ServerInfo( - name=payload_data.server_name - or "Unknown", # Echter Server-Name (korrigiert in parse_server_payload) - map=payload_data.maps[0] if payload_data.maps else "Unknown", - players=payload_data.players or 0, - max_players=payload_data.max_players or 0, - game_mode=payload_data.game_mode or "Unknown", - password_protected=payload_data.srv_type == "p", - version=None, - environment=payload_data.environment, - comment=payload_data.comment, # PC-UID oder andere Info - server_login="", - pc_guid=payload_data.comment - if payload_data.comment and payload_data.comment.startswith("PC-") - else None, # PC-UID - time_limit=0, - nb_laps=0, - spectator_slots=0, - build_number=0, - private_server=payload_data.srv_type == "p", - ladder_server=payload_data.srv_type == "s", - status_flags=0, - challenge_crc=0, - public_ip="", - local_ip="", - raw_data=response_data.hex(), + if version != self._CONNECTION_ADMIN_VERSION: + raise InvalidPacketException( + f"Unsupported ConnectionAdmin version: {version}" ) - except asyncio.TimeoutError: - raise InvalidPacketException("Timeout while waiting for server response") - finally: - writer.close() - await writer.wait_closed() + if subtype == self._SUBTYPE_REFUSED: + raise InvalidPacketException("The server refused the query") - def parse_server_payload(self, data: bytes) -> TrackmaniaPayloadData: - """ - Parst einen Server-Payload basierend auf MCP-Erkenntnissen. + if subtype != self._SUBTYPE_INFO: + return None - Args: - data: Die Rohdaten des Payloads + reply_id = reader.u32() + data = reader.take(reader.u32()) - Returns: - TrackmaniaPayloadData mit extrahierten Informationen - """ - result = TrackmaniaPayloadData() - - # 1. String bei 0x27 extrahieren (kann PC-UID oder Server-Name sein, abhängig vom SRV-Typ) - string_at_0x27 = None - if len(data) >= 0x2B: # 0x27 + 4 bytes für Länge - string_at_0x27, _ = self._deserialize_string(data, 0x27) - # Temporär speichern - wird später basierend auf SRV-Typ zugeordnet - result.server_name = string_at_0x27 - - # 2. #SRV# Marker und Typ finden - srv_pos = data.find(b"#SRV#") - if srv_pos != -1 and srv_pos + 5 < len(data): - # Typ-Byte nach #SRV# - srv_type_byte = data[srv_pos + 5] - if srv_type_byte == 0x00: - result.srv_type = "null" - elif chr(srv_type_byte).lower() in ["f", "s", "p"]: - result.srv_type = chr(srv_type_byte).lower() - else: - result.srv_type = f"unknown_{srv_type_byte:02x}" - - # 3. MCP-basierte Challenge/Map-Namen Extraktion (mit SRV-Typ) - challenge_name = self._extract_challenge_name(data, srv_pos, result.srv_type) - if challenge_name: - result.maps.append(challenge_name) - - # 4. Weitere Strings extrahieren für Environment, etc. - strings = self._extract_all_strings(data) - result.raw_strings = strings - - # 5. Spezifische Daten extrahieren - for string in strings: - # Fallback für Maps wenn MCP-Extraktion nichts fand - if not result.maps and self._is_valid_challenge_name(string): - result.maps.append(string) - # Environment - elif string.lower() in ["stadium", "island", "bay", "coast"]: - result.environment = string.title() - - # 6. Spielerzahlen extrahieren (basierend auf Typ) - player_data = self._extract_player_counts(data, srv_pos, result.srv_type) - if player_data: - result.players, result.max_players = player_data - - # 7. Game-Mode via MCP/Ghidra-Marker extrahieren (robust, ohne String-Heuristik) - mode_name = self._extract_game_mode(data) - if mode_name: - result.game_mode = mode_name - - # 8. MCP-basierte korrekte Zuordnung basierend auf SRV-Typ - if result.srv_type == "p": - # Private Server: 0x27 = PC-UID, echter Name in ASCII-Strings - result.comment = string_at_0x27 # PC-UID - - # Finde echten Server-Namen aus ASCII-Strings - potential_names = [ - s - for s in strings - if len(s) >= 4 - and not s.startswith("PC-") - and s != result.environment - and not self._is_valid_challenge_name(s) - and not s.startswith("#") - and "lanparty" not in s.lower() - and "obstacle" not in s.lower() - ] # Filter korrupte Namen - - if potential_names: - potential_names.sort(key=len, reverse=True) - result.server_name = potential_names[0] # Längster = echter Server-Name - - elif result.srv_type == "null" or result.srv_type is None: - # Default/Null Server: Finde echten Server-Namen in ASCII-Strings - # 0x27 könnte PC-UID oder Server-Name sein - prüfe Pattern - - # Finde potentielle Server-Namen (alphabetische Namen bevorzugt) - potential_names = [ - s - for s in strings - if 3 <= len(s) <= 15 # Kurze, prägnante Namen - and not s.startswith("PC-") - and not s.startswith("#") - and s != result.environment - and not self._is_valid_challenge_name(s) - and not any( - kw in s.lower() for kw in ["stadium", "lanparty", "obstacle"] - ) - and s.isalpha() - ] # Nur alphabetische Namen (wie "Bruno") - - if potential_names: - # Priorisiere kürzeste alphabetische Namen - potential_names.sort(key=len) - real_server_name = potential_names[0] - - # Wenn 0x27 String länger/anders ist, ist es wahrscheinlich PC-UID - if string_at_0x27 and string_at_0x27 != real_server_name: - result.server_name = real_server_name - result.comment = string_at_0x27 # PC-UID/Login - else: - result.server_name = string_at_0x27 or real_server_name - result.comment = None - else: - # Fallback: 0x27 als Server-Name - result.server_name = string_at_0x27 - result.comment = None - - else: - # Andere Server-Typen: Fallback zur alten Logik - potential_names = [ - s - for s in strings - if len(s) >= 3 - and s != result.environment - and not self._is_valid_challenge_name(s) - and not s.startswith("#") - ] - - if potential_names: - potential_names.sort(key=len, reverse=True) - longest = potential_names[0] - if len(longest) > len(string_at_0x27 or ""): - result.server_name = longest - result.comment = string_at_0x27 - else: - result.comment = longest + if reply_id == self._NO_INFO: + raise InvalidPacketException("The server has no game info available yet") - return result + return data if reply_id == request_id else None - def _deserialize_string( - self, data: bytes, offset: int - ) -> Tuple[Optional[str], int]: - """ - Deserialisiert einen String mit 4-Byte Längen-Präfix (Little Endian). + @classmethod + def build_connection_admin( + cls, subtype: int, request_id: Optional[int] = None + ) -> bytes: + """Builds a framed CNetFormConnectionAdmin message.""" + payload = struct.pack(" len(data): - return None, 0 + return cls.build_message(cls._MESSAGE_CONNECTION_ADMIN, payload) - # Länge lesen (4 Bytes, Little Endian) - length = struct.unpack(" bytes: + """Builds a framed, checksummed and uncompressed message.""" + message = bytearray([0x80 | cls._FLAG_CHECKSUM, message_type]) + message += bytes(4) + payload + message[2:6] = struct.pack(" 100 or offset + 4 + length > len(data): - return None, 0 + return struct.pack(" Tuple[int, bytes]: + """ + Decodes a message without its length prefix. - def _extract_all_strings(self, data: bytes) -> List[str]: + :return: Message type and (decompressed) payload """ - Extrahiert alle lesbaren ASCII-Strings aus den Daten. + if len(message) < 2 or message[0] & 0xF0 != 0x80: + raise InvalidPacketException("Invalid message header") - Args: - data: Die Rohdaten + flags, message_type = message[0], message[1] + position = 2 - Returns: - Liste der gefundenen Strings - """ - strings = [] - current_string = bytearray() + if flags & cls._FLAG_SEQUENCE: + position += 2 - for byte in data: - if 32 <= byte <= 126: # Druckbare ASCII-Zeichen - current_string.append(byte) - else: - if len(current_string) >= 3: # Mindestens 3 Zeichen - try: - string = current_string.decode("ascii") - strings.append(string) - except Exception: - pass - current_string = bytearray() - - # Letzten String nicht vergessen - if len(current_string) >= 3: - try: - string = current_string.decode("ascii") - strings.append(string) - except Exception: - pass + if flags & cls._FLAG_CHECKSUM: + if len(message) < position + 4: + raise InvalidPacketException("Truncated message") - return strings + checksum = struct.unpack_from(" bool: - """ - Prüft ob ein String ein Map-Name ist (striktere Typen, keine korrupten Suffixe). - """ - import re - - allowed = r"(race|acrobatic|speed|endurance|platform|puzzle)" - if re.match(rf"^[A-E]\d{{2}}-{allowed}$", string, re.IGNORECASE): - return True - if re.match(rf"^\d+-{allowed}$", string, re.IGNORECASE): - return True - return False - - def _extract_player_counts( - self, data: bytes, srv_pos: int, srv_type: str - ) -> Optional[Tuple[int, int]]: - """ - Extrahiert Spielerzahlen basierend auf dem SRV-Typ. + if checksum != cls._checksum(message, position): + raise InvalidPacketException("Message checksum mismatch") - MCP-Erkenntnisse zeigen verschiedene Offsets für verschiedene Typen. - """ - if srv_pos == -1: - return None + position += 4 - # Verschiedene Offset-Patterns basierend auf Typ (MCP-korrigiert) - if srv_type == "null": - # Für Null-Byte: Offsets +7 und +9 - offsets = [(7, 9)] - elif srv_type == "p": - # Für Private Server: MCP-Analyse zeigt +10/+11 für aktive Spieler, +9/+11 fallback - offsets = [ - # Beobachtung 172.29.100.29: plausibles Paar 1/6 bei SRV+29/SRV+50 - (29, 50), - (10, 11), - (9, 11), - (7, 11), - # zusätzliche pragmatische Kandidaten, beobachtet auf manchen 'p'-Servern - (12, 14), - (7, 9), - (41, 45), - ] # Reihenfolge: etabliert, dann heuristisch - else: - # Für andere Typen: Teste mehrere Patterns - offsets = [(7, 9), (9, 11), (7, 11), (41, 45), (12, 14), (15, 17)] - - # Teste die Offset-Patterns (MCP-korrigiert für aktuelle Spieler) - for current_offset, max_offset in offsets: - if srv_pos + max_offset < len(data): - current_players = data[srv_pos + current_offset] - max_players = data[srv_pos + max_offset] - - # Plausibilitätsprüfung - if 0 <= current_players <= max_players <= 200 and max_players > 0: - return current_players, max_players - - # Letzter Fallback nur für 'p'-Server: heuristische Suche in kleinem Fenster - # Motiv: Es gibt Varianten, bei denen die Felder deutlich verschoben sind. - if srv_type == "p": - window_start = max(0, srv_pos) - window_end = min(len(data), srv_pos + 96) - best_pair = None - best_score = 1e9 - common_max_values = {6, 8, 10, 12, 14, 16, 20, 24, 32, 48, 64} - for max_idx in range(srv_pos + 16, window_end): - max_val = data[max_idx] - if not (1 <= max_val <= 64): - continue - # Suche current in der Nähe, bevorzugt vorher - search_from = max(window_start, max_idx - 40) - for cur_idx in range(search_from, max_idx): - cur_val = data[cur_idx] - if 0 <= cur_val <= max_val: - # Scoring: kleinere max-Werte bevorzugen (realistische Slot-Zahlen), Nähe der Felder - score = (0 if max_val in common_max_values else 10) + ( - max_idx - cur_idx - ) - if score < best_score: - best_score = score - best_pair = (cur_val, max_val) - if best_pair is not None: - return best_pair - - return None - - def _extract_game_mode(self, data: bytes) -> Optional[str]: - """ - Extrahiert den Spielmodus aus dem Payload. + if not flags & cls._FLAG_COMPRESSED: + return message_type, bytes(message[position:]) - Strategien (in dieser Reihenfolge): - 1) #SRV#-Offset-Erkennung: Für bestimmte Varianten (z. B. 'p') liegt die Mode-ID an einem festen Offset - 2) Marker-basierte Erkennung: Suche nach 0xFF 0xFF 0xFF 0xFF und nutze Byte an +7 als Modus-ID - 3) Stadium-Pattern-Fallback: Auswertung der Bytes nach dem 'Stadium' String - 4) Letzter Fallback: Keine Heuristik über Mapnamen (vermeidet Fehlzuordnung wie 'A01-Race') - """ - # 1) Marker-basierte Erkennung - mode_id = self._extract_game_mode_id_by_marker(data) - if mode_id is not None: - name = self._map_game_mode_id_to_name(mode_id) - if name: - return name - - # 2) Stadium-Pattern-Fallback - mode_id = self._extract_game_mode_id_by_stadium_pattern(data) - if mode_id is not None: - name = self._map_game_mode_id_to_name(mode_id) - if name: - return name - - return None - - def _extract_game_mode_id_by_marker(self, data: bytes) -> Optional[int]: - """ - Sucht nach dem 0xFFFFFFFF Marker und liest das Spielmodus-Byte bei +7. - Laut Analyse liefert dieses Byte Werte wie 0x09 (Cup), 0x07 (Rounds), 0x06 (Team), 0x00 (Time Attack). - """ - marker = b"\xff\xff\xff\xff" - idx = data.find(marker) - if idx != -1 and idx + 8 <= len(data): - try: - # Kandidaten-Offsets testen (+6, +7, +5), nur plausible IDs akzeptieren - candidates = [idx + 7, idx + 6, idx + 5] - valid_ids = {0, 1, 2, 3, 4, 5, 6, 7, 9} - for off in candidates: - if 0 <= off < len(data): - val = data[off] - if val in valid_ids: - return val - except Exception: - return None - return None - - def _extract_game_mode_id_by_stadium_pattern(self, data: bytes) -> Optional[int]: - """ - Fallback-Erkennung über Byte-Muster relativ zum 'Stadium'-String. - Bekanntes Mapping: - - 0x01 0x20 => TimeAttack (ID 0) - - 0x03 0x1e => Tournament (ID 3) - - 0x06 0x32 => Team (ID 6) - - 0x07 0x03 => Rounds (ID 7) - - 0x09 xx => Cup (ID 9) - """ - # Suche 'Stadium' NACH dem '#SRV#'-Marker, um den richtigen Kontext zu erwischen - anchor = b"Stadium" - srv_pos = data.find(b"#SRV#") - if srv_pos == -1: - return None - pos = data.find(anchor, srv_pos) - if pos == -1: - return None - pattern_start = pos + len(anchor) - # Wir benötigen mindestens ein kleines Fenster nach dem Anchor - window_end = min(len(data), pattern_start + 32) - if pattern_start >= window_end: - return None + if len(message) < position + 4: + raise InvalidPacketException("Truncated message") - # 1) Klassische Position b5/b6 (kompatibel zu früherer Implementierung) - if len(data) > pattern_start + 6: - b5 = data[pattern_start + 5] - b6 = data[pattern_start + 6] - if b5 == 0x01 and b6 == 0x20: - return 0 # TimeAttack - if b5 == 0x03 and b6 == 0x1E: - return 3 # Tournament - if b5 == 0x06 and b6 == 0x32: - return 6 # Team - if b5 == 0x07 and b6 == 0x03: - return 7 # Rounds - if b5 == 0x09: - return 9 # Cup - - # 2) Flexibles Scannen im kleinen Fenster: suche bekannte Paare in beliebiger Ausrichtung - window = data[pattern_start:window_end] - # Paare, die als direkt aufeinanderfolgende Bytes auftreten sollten - pair_to_mode = { - (0x01, 0x20): 0, # TimeAttack - (0x03, 0x1E): 3, # Tournament - (0x06, 0x32): 6, # Team - (0x07, 0x03): 7, # Rounds - } - for i in range(0, len(window) - 1): - a, b = window[i], window[i + 1] - if (a, b) in pair_to_mode: - return pair_to_mode[(a, b)] - # Cup kann als Einzelwert im Fenster auftreten - if 0x09 in window: - return 9 - - # 3) Schwache Heuristik: Einzel-ID im Fenster (z. B. 0x07 für Rounds) bevorzugt, wenn eindeutig - for candidate in (7, 6, 3, 0): - if candidate in window: - return candidate - - return None - - def _map_game_mode_id_to_name(self, mode_id: int) -> Optional[str]: - """ - Mappt erkannte Modus-IDs auf sprechende Namen. - Bevorzugt bekannte TMNF-Bezeichnungen. - """ - mapping = { - 0: "TimeAttack", - 3: "Tournament", # In manchen Quellen auch 'Tournament'; hier konservativ auf Laps mappen - 6: "Team", - 7: "Rounds", - 9: "Cup", - } - # Weitere bekannte IDs aus Dokus (falls auftauchen) - extra_aliases = { - 1: "TimeAttack", - 2: "Team", - 4: "Stunts", - 5: "Cup", - } - return mapping.get(mode_id) or extra_aliases.get(mode_id) - - def _extract_challenge_name( - self, data: bytes, srv_pos: int, srv_type: str = None - ) -> Optional[str]: - """ - Extrahiert den Challenge/Map-Namen basierend auf MCP-Analyse. + size = struct.unpack_from(" cls._MAX_MESSAGE_SIZE: + raise InvalidPacketException(f"Invalid uncompressed size: {size}") - Args: - data: Die Rohdaten - srv_pos: Position des #SRV# Markers - srv_type: Typ des Servers ('null', 'p', etc.) + return message_type, _lzo1x_decompress(message[position + 4 :], size) - Returns: - Challenge-Name oder None - """ - if srv_pos == -1: - return None + @classmethod + def _checksum(cls, message: bytes, position: int) -> int: + data = bytearray(message) + data[position : position + 4] = bytes(4) + digest = hmac.new(cls._CHECKSUM_KEY, bytes(data), hashlib.md5).digest() - # Zuerst Prefix-Varianten versuchen (1/2/4 Bytes), unabhängig vom SRV-Typ - name = self._extract_challenge_with_prefix(data, srv_pos) - if name: - return name - # Fallback: direkte ASCII-Strings - return self._extract_challenge_without_prefix(data, srv_pos) + return sum(struct.unpack("<4I", digest)) & 0xFFFFFFFF - def _extract_challenge_with_prefix( - self, data: bytes, srv_pos: int - ) -> Optional[str]: + @classmethod + def parse_server_info(cls, data: bytes) -> ServerInfo: """ - Extrahiert Challenge-Namen mit Längenpräfix (1/2/4 Byte; LE für 2/4). + Parses the server info sent in reply to an info request. + + Layout, one block per class of the serialisation chain:: + + CNetMasterHost + u8 game tag: (tag & 0xE0) == 0x20, (tag & 0x1F) == 0x0D + u8[4] IP address, reversed byte order + u16 port + str host login + CGameNetServerInfo + str "#SRV#" + "p" player password / "s" spectator password / "f" both + CGameCtnNetServerInfo (only for a "#SRV#" login) + str unused, always empty + u8 player count, max players, spectator count, max spectators, + ladder mode + wstr server name + str pack mask + u32 n wstr player names[n], i32 ladder rankings[n] + wstr comment + CTrackManiaNetworkServerInfo + u8 game mode (GAME_MODES) + u32 time limit (ms) / points limit / number of laps + u8 number of challenges in the playlist + u32 n challenges[n]: wstr name, u32 gold time, u16 copper price, + u8 environment index; current challenge first + u32 n environment ids[n] (Nadeo lookback strings) """ - for prefix_size in (1, 2, 4): - candidate = self._scan_challenge_with_prefix_size( - data, srv_pos, prefix_size + reader = _Reader(data) + game_tag = reader.u8() + + if game_tag & 0xE0 != 0x20 or game_tag & 0x1F != cls._GAME_TAG_TRACKMANIA: + raise InvalidPacketException( + f"Not a TrackMania server (game tag 0x{game_tag:02x})" ) - if candidate: - return candidate - return None - def _scan_challenge_with_prefix_size( - self, data: bytes, srv_pos: int, prefix_size: int - ) -> Optional[str]: - """ - Durchsucht den Bereich nach #SRV# nach einem length-prefixed String mit gegebener Präfixgröße. - """ - search_start = srv_pos + 32 - search_end = min(len(data), srv_pos + 220) - if search_start >= search_end: - return None + address = ".".join(str(octet) for octet in reversed(reader.take(4))) + port = reader.u16() + server_login = reader.string() + player_login = reader.string() + + info = ServerInfo( + name=server_login, + map="", + players=0, + max_players=0, + game_mode="Unknown", + server_login=server_login, + pc_guid=server_login, + server_address=address, + server_port=port, + local_ip=address, + raw_data=data.hex(), + ) - step = 1 - for offset in range(search_start, search_end - (prefix_size + 4), step): - try: - if offset + prefix_size >= len(data): - break + if not player_login.startswith("#SRV#"): + return info + + password_flag = player_login[5:6] + info.password_protected = info.private_server = password_flag in ("p", "f") + info.spectator_password_protected = password_flag in ("s", "f") + + reader.string() + info.players = reader.u8() + info.max_players = reader.u8() + info.spectators = reader.u8() + info.max_spectators = info.spectator_slots = reader.u8() + info.ladder_mode = reader.u8() + info.ladder_server = info.ladder_mode != 0 + info.name = reader.wstring() + info.pack_mask = reader.string() + + names = [reader.wstring() for _ in range(reader.count())] + info.player_list = [Player(name, reader.i32()) for name in names] + info.comment = reader.wstring() + + if reader.remaining == 0: + return info + + info.game_mode_id = reader.u8() + info.game_mode = cls.GAME_MODES.get( + info.game_mode_id, f"Unknown ({info.game_mode_id})" + ) + limit = reader.u32() + + if info.game_mode_id in (1, 8): + info.time_limit = limit + elif info.game_mode_id == 7: + info.nb_laps = limit + elif info.game_mode_id in (3, 6, 9): + info.points_limit = limit + + info.nb_challenges = reader.u8() + challenges = [ + (reader.wstring(), reader.u32(), reader.u16(), reader.u8()) + for _ in range(reader.count()) + ] + ids = _IdReader(reader) + environments = [ids.read() for _ in range(reader.count())] + + info.challenges = [ + Challenge( + name, + gold_time, + copper_price, + environments[index] if index < len(environments) else "Unknown", + ) + for name, gold_time, copper_price, index in challenges + ] - if prefix_size == 1: - length = data[offset] - elif prefix_size == 2: - length = struct.unpack(" len(data): - continue - segment = data[start:end] - try: - raw_text = segment.decode("ascii", errors="ignore") - except Exception: - continue +class _Reader: + """Bounds-checked little-endian reader for Nadeo archives.""" - # Nur druckbare Zeichen behalten - cleaned = "".join(ch for ch in raw_text if 32 <= ord(ch) <= 126) - if not cleaned: - continue + def __init__(self, data: bytes): + self._data = data + self._position = 0 - # Strikte Map-Erkennung als Substring - strict = self._find_strict_challenge_in_text(cleaned) - if strict: - return strict - except Exception: - continue - return None + @property + def remaining(self) -> int: + return len(self._data) - self._position - def _extract_challenge_without_prefix( - self, data: bytes, srv_pos: int - ) -> Optional[str]: - """ - Extrahiert Challenge-Namen ohne Längenpräfix (für Private Server). - """ - # Suche nach direkten ASCII-Strings ab SRV-Position - search_start = srv_pos + 10 - search_data = data[search_start:] + def take(self, count: int) -> bytes: + if count < 0 or count > self.remaining: + raise InvalidPacketException("Truncated server info") - current_string = bytearray() - found_strings = [] + data = self._data[self._position : self._position + count] + self._position += count - for i, byte in enumerate(search_data): - if 32 <= byte <= 126: # Druckbare ASCII-Zeichen - current_string.append(byte) - else: - if len(current_string) >= 5: # Mindestens 5 Zeichen für Challenge-Namen - try: - string = current_string.decode("ascii") - if self._is_valid_challenge_name(string): - return string - found_strings.append(string) - except Exception: - pass - current_string = bytearray() - - # Letzten String nicht vergessen - if len(current_string) >= 5: - try: - string = current_string.decode("ascii") - if self._is_valid_challenge_name(string): - return string - found_strings.append(string) - except Exception: - pass + return data - # Fallback: Erste gültige Challenge aus gefundenen Strings - for string in found_strings: - if self._is_valid_challenge_name(string): - return string + def u8(self) -> int: + return self.take(1)[0] - return None + def u16(self) -> int: + return struct.unpack(" bool: - """ - Prüft ob ein String ein gültiger Challenge/Map-Name ist. + def u32(self) -> int: + return struct.unpack(" int: + return struct.unpack(" int: + # Every list element takes at least one byte. + count = self.u32() - Returns: - True wenn gültiger Challenge-Name - """ - import re - - # Nur druckbare ASCII-Zeichen zulassen - if any(ord(c) < 32 or ord(c) > 126 for c in name): - return False - - # Zu kurz oder zu lang - if len(name) < 3 or len(name) > 50: - return False - - # Standard TrackMania Challenge Pattern: A01-Race, C02-Acrobatic - # Aber nur vollständige bekannte Challenge-Namen (KEINE korrupten wie "C04-Raceh") - standard_pattern = re.match(r"^[A-E]\d{2}-([A-Za-z]{4,})$", name) - if standard_pattern: - challenge_type = standard_pattern.group(1).lower() - # Nur bekannte Challenge-Typen aus MCP-Analyse - known_types = [ - "race", - "acrobatic", - "speed", - "endurance", - "platform", - "puzzle", - ] - # WICHTIG: "raceh" ist NICHT in known_types, also wird C04-Raceh abgelehnt! - if challenge_type in known_types: - return True - - # Verkürzte Namen: 5-Endurance, 1-Speed (nur bekannte Typen) - short_pattern = re.match(r"^\d+-([A-Za-z]{4,})$", name) - if short_pattern: - challenge_type = short_pattern.group(1).lower() - # Nur bekannte Challenge-Typen - known_types = [ - "race", - "acrobatic", - "speed", - "endurance", - "platform", - "puzzle", - ] - if challenge_type in known_types: - return True - - # Challenge/Race Keywords (aus MCP-Strings) - challenge_keywords = [ - "race", - "speed", - "endurance", - "acrobatic", - "challenge", - "track", - "circuit", - "course", - "stage", - ] + if count > self.remaining: + raise InvalidPacketException(f"Invalid list length: {count}") + + return count + + def string(self) -> str: + return self.take(self.u32()).decode("utf-8", errors="replace") + + def wstring(self) -> str: + return self.take(self.u32()).decode("utf-8-sig", errors="replace") + + +class _IdReader: + """Reads Nadeo identifiers ("lookback strings") sharing one string table.""" - name_lower = name.lower() - if any(keyword in name_lower for keyword in challenge_keywords): - # Aber nicht wenn es offensichtlich ein Server-Name oder anderer String ist - # Und mindestens ein Wort muss vollständig sein (nicht nur Teil eines Wortes) - # WICHTIG: Blockiere korrupte Namen wie "raceh" (race + unbekanntes Ende) - if ( - not any( - exclude in name_lower - for exclude in ["server", "player", "time", "score"] + def __init__(self, reader: _Reader): + self._reader = reader + self._version: Optional[int] = None + self._strings: List[str] = [] + + def read(self) -> str: + if self._version is None: + self._version = self._reader.u32() + + if self._version not in (2, 3): + raise InvalidPacketException(f"Unsupported id version: {self._version}") + + value = self._reader.u32() + + if value == 0xFFFFFFFF: + return "" + + if value & 0xC0000000 not in (0x40000000, 0x80000000): + # Numeric collection id + return str(value) + + index = value & 0x0FFFFFFF + + if self._version == 2 or index == 0: + string = self._reader.string() + self._strings.append(string) + return string + + if index > len(self._strings): + raise InvalidPacketException(f"Invalid id reference: {index}") + + return self._strings[index - 1] + + +def _lzo1x_decompress(source: bytes, size: int) -> bytes: + """Decompresses an LZO1X stream into exactly size bytes.""" + output = bytearray() + position = 0 + + def copy_literals(count: int): + nonlocal position + + if position + count > len(source) or len(output) + count > size: + raise InvalidPacketException("Corrupted LZO stream") + + output.extend(source[position : position + count]) + position += count + + def copy_match(distance_position: int, count: int): + if distance_position < 0 or len(output) + count > size: + raise InvalidPacketException("Corrupted LZO stream") + + for i in range(count): + output.append(output[distance_position + i]) + + def read_length(base: int) -> int: + nonlocal position + length = 0 + + while source[position] == 0: + length += 255 + position += 1 + + length += base + source[position] + position += 1 + + return length + + try: + # 0: next code < 16 is a literal run, 1-3: a 2 byte match follows + # the trailing literals, 4: a 3 byte match follows a literal run. + state = 0 + + if source[0] > 17: + position = 1 + count = source[0] - 17 + copy_literals(count) + state = count if count < 4 else 4 + + while True: + code = source[position] + position += 1 + + if code < 16: + if state == 0: + copy_literals((code or read_length(15)) + 3) + state = 4 + continue + + distance = (code >> 2) + (source[position] << 2) + position += 1 + + if state == 4: + copy_match(len(output) - 0x801 - distance, 3) + else: + copy_match(len(output) - 1 - distance, 2) + elif code >= 64: + distance = ((code >> 2) & 7) + (source[position] << 3) + position += 1 + copy_match(len(output) - 1 - distance, (code >> 5) + 1) + elif code >= 32: + count = (code & 31 or read_length(31)) + 2 + distance = (source[position] | source[position + 1] << 8) >> 2 + position += 2 + copy_match(len(output) - 1 - distance, count) + else: + count = (code & 7 or read_length(7)) + 2 + distance = ((code & 8) << 11) + ( + (source[position] | source[position + 1] << 8) >> 2 ) - and len(name) >= 4 # Mindestlänge - and not name_lower.endswith("p") # Nicht unvollständig wie "RaceP" - and not name_lower.endswith("h") # Nicht unvollständig wie "Raceh" - and not re.match( - r"^[A-E]\d{2}-.*[ph]$", name, re.IGNORECASE - ) # Nicht Standard-Pattern mit 'p'/'h' am Ende - and (" " in name or len(name) >= 5) - ): # Entweder Leerzeichen oder mindestens 5 Zeichen - return True - - # GBX-Pattern (aus MCP: .TrackMania.gbx) - if ".gbx" in name_lower or "trackmania" in name_lower: - return True - - return False - - def _find_strict_challenge_in_text(self, text: str) -> Optional[str]: - """ - Sucht in einem Text nach einem strikt passenden Challenge-Namen - (z. B. A01-Race, C06-Speed, etc.) und gibt den ersten Treffer zurück. - """ - import re + position += 2 - pattern = re.compile( - r"([A-E]\d{2}-(?:Race|Acrobatic|Speed|Endurance|Platform|Puzzle))", - re.IGNORECASE, - ) - m = pattern.search(text) - return m.group(0) if m else None + if distance == 0: + break - def debug_payload(self, data: bytes) -> Dict[str, Any]: - """ - Debug-Funktion zur Analyse eines Payloads. + copy_match(len(output) - distance - 0x4000, count) - Args: - data: Die Rohdaten + state = source[position - 2] & 3 - Returns: - Dictionary mit Debug-Informationen - """ - debug_info = { - "length": len(data), - "hex_dump": data[:100].hex() if len(data) > 100 else data.hex(), - "server_name_offset": 0x27, - "srv_marker_pos": -1, - "srv_type": None, - "strings": [], - "potential_player_offsets": {}, - } - - # Servername bei 0x27 - if len(data) >= 0x2B: - server_name, bytes_read = self._deserialize_string(data, 0x27) - debug_info["server_name"] = server_name - debug_info["server_name_bytes_read"] = bytes_read - - # SRV Marker - srv_pos = data.find(b"#SRV#") - if srv_pos != -1: - debug_info["srv_marker_pos"] = srv_pos - if srv_pos + 5 < len(data): - srv_type_byte = data[srv_pos + 5] - debug_info["srv_type_byte"] = f"0x{srv_type_byte:02x}" - if srv_type_byte == 0x00: - debug_info["srv_type"] = "null" - elif chr(srv_type_byte) in ["f", "s", "p"]: - debug_info["srv_type"] = chr(srv_type_byte) - - # Alle Strings - debug_info["strings"] = self._extract_all_strings(data) - - # MCP-basierte Challenge-Namen Extraktion - challenge_name = self._extract_challenge_name(data, srv_pos) - debug_info["mcp_challenge_name"] = challenge_name - debug_info["challenge_extraction_method"] = ( - "MCP-based" if challenge_name else "fallback" - ) + if state: + copy_literals(state) + except IndexError as e: + raise InvalidPacketException("Truncated LZO stream") from e + + if len(output) != size: + raise InvalidPacketException("LZO size mismatch") - # Potentielle Spielerzahl-Offsets - if srv_pos != -1: - test_offsets = [(7, 9), (41, 45), (12, 14), (15, 17)] - for curr_off, max_off in test_offsets: - if srv_pos + max_off < len(data): - curr = data[srv_pos + curr_off] - max_val = data[srv_pos + max_off] - debug_info["potential_player_offsets"][ - f"+{curr_off}/+{max_off}" - ] = f"{curr}/{max_val}" - - return debug_info + return bytes(output) diff --git a/opengsq/protocols/trackmania_sunrise.py b/opengsq/protocols/trackmania_sunrise.py new file mode 100644 index 0000000..a54b8e4 --- /dev/null +++ b/opengsq/protocols/trackmania_sunrise.py @@ -0,0 +1,625 @@ +""" +TrackMania Original/Sunrise/Nations ESWC native server query. + +The three games share one dedicated server, TrackManiaServer.exe, started with +/game=Original, /game=Sunrise or /game=Nations, and one network protocol: the +predecessor of the TmForever protocol (see trackmania_nations.py). +Reverse-engineered from TrackManiaServer.exe of TrackMania Sunrise eXtreme. + +All integers are little-endian. Framing, message header and LZO1X compression +are the same as for TmForever:: + + u32 length (TCP only) | u8 flags | u8 type | [u16 sequence] | [u32 checksum] | payload + +The checksum uses another key and adds the first key word instead of the +fourth digest word:: + + w = HMAC-MD5(_CHECKSUM_KEY, message with the checksum zeroed) as four u32 + checksum = w[0] + w[1] + w[2] + key[0] + +The server does not require it for the messages used here, but it sets it on +every reply, which the query uses to recognise a TrackMania server. + +TCP query on the game port (default 2350), CNetFormConnectionAdmin (type 3):: + + u32 version 4 (Original, Sunrise) or 5 (Nations ESWC). A lower version + is refused with subtype 1 (the reply carries the server + version), a higher one is ignored. + u32 subtype 8: switch the connection into query mode (no payload) + 7: u32 request_id (client -> server) + 6: u32 request_id | u32 size | u8[size] (server -> client) + +UDP LAN discovery on the game port, a server only answers for its own game id:: + + CNetFormQuerrySessions (type 0): str game_id | str client | addr | u32 nonce + CNetFormEnumSessions (type 1): u32 nonce | str application | str game_id + | str version | str host_name | addr | addr + +addr is the IPv4 address in reversed byte order followed by the u16 port. +str is u32 length + bytes, wstr the same with UTF-8 (prefixed by a BOM if it +contains non-ASCII characters). +""" + +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import secrets +import struct +from typing import Optional, Tuple + +from opengsq.exceptions import InvalidPacketException, ServerNotFoundException +from opengsq.protocol_base import ProtocolBase +from opengsq.protocol_socket import UdpClient +from opengsq.protocols.trackmania_nations import _lzo1x_decompress, _Reader +from opengsq.responses.trackmania_sunrise import ( + Challenge, + Player, + ServerInfo, + SessionInfo, +) + + +class TrackmaniaSunrise(ProtocolBase): + """ + TrackMania Original/Sunrise/Nations ESWC Protocol Implementation + """ + + @property + def full_name(self) -> str: + return "TrackMania Sunrise Protocol" + + DEFAULT_PORT = 2350 + + # Game ids the server compares LAN session queries against + GAMES = { + "TmOriginal": "TrackMania Original", + "TmSunrise": "TrackMania Sunrise", + "TmNationsESWC": "TrackMania Nations ESWC", + } + + GAME_MODES = { + 1: "TimeAttack", + 3: "Rounds", + 6: "Team", + 7: "Laps", + 8: "Stunts", + } + + # Environments and decorations (moods) the server puts into its decoration + # table, per game (0x005ea740). The decoration index of a challenge is + # 2 + its position in that table, see decoration(). + DECORATIONS = { + "TmOriginal": ( + ("Alpine", "Speed", "Rally"), + ( + "32x32Sunset", + "32x32Sunrise", + "Simple", + "30x30Sunrise", + "30x30", + "30x30Sunset", + "20x60Sunrise", + "20x60", + "20x60Sunset", + "10x150Sunrise", + "10x150", + "10x150Sunset", + ), + ), + "TmSunrise": ( + ("Bay", "Coast", "Island"), + ("Sunrise", "Day", "Sunset", "Night"), + ), + "TmNationsESWC": (("Stadium",), ("Day",)), + } + + _FLAG_COMPRESSED = 0x01 + _FLAG_CHECKSUM = 0x02 + _FLAG_SEQUENCE = 0x0C + + _MESSAGE_QUERY_SESSIONS = 0x00 + _MESSAGE_ENUM_SESSIONS = 0x01 + _MESSAGE_CONNECTION_ADMIN = 0x03 + + # ConnectionAdmin version of Original/Sunrise and of Nations ESWC + _VERSION_SUNRISE = 4 + _VERSION_NATIONS_ESWC = 5 + + _SUBTYPE_REFUSED = 1 + _SUBTYPE_INFO = 6 + _SUBTYPE_INFO_REQUEST = 7 + _SUBTYPE_QUERY_MODE = 8 + + # Request id the server puts into the reply when it has no game info yet. + _NO_INFO = 0xFFFFFFFF + + _CHECKSUM_KEY = bytes.fromhex("08c481303a1226abaf1d6ae4fb65fbc9") + _MAX_MESSAGE_SIZE = 0x100000 + + # First byte of the server info: high bits 000 = valid, low bits = game. + # The game id is known for Nations ESWC only, Original and Sunrise share 0x07. + _GAME_TAGS = {0x07: "", 0x09: "TmNationsESWC"} + + _CLIENT_NAME = "opengsq" + + def __init__(self, host: str, port: int = DEFAULT_PORT, timeout: float = 5.0): + super().__init__(host, port, timeout) + + async def get_info(self, game_id: Optional[str] = None) -> ServerInfo: + """ + Retrieves the server information via TCP. + + Original and Sunrise servers cannot be told apart by this query + (game_id is empty for them), use get_session() for that. + + :param game_id: Game of the server (TmOriginal or TmSunrise), known from + get_session(). Needed to resolve the environments of the challenges of + an Original or Sunrise server. + :return: A ServerInfo object containing server information + :raises ServerNotFoundException: If no TCP connection can be established + :raises InvalidPacketException: If the server does not answer like a TrackMania server + """ + version = self._VERSION_SUNRISE + + try: + data = await self._query_info(version) + except _RefusedException as e: + # Nations ESWC refuses the Original/Sunrise version and tells its own + if e.server_version != self._VERSION_NATIONS_ESWC: + raise InvalidPacketException( + f"The server refused the query (version {e.server_version})" + ) from e + + version = e.server_version + data = await self._query_info(version) + + info = self.parse_server_info(data, game_id) + info.protocol_version = version + + return info + + async def get_session(self, game_id: Optional[str] = None) -> SessionInfo: + """ + Retrieves the LAN session announcement of the server via UDP. + + The server only answers for its own game, so this identifies whether + it runs TrackMania Original, Sunrise or Nations ESWC. + + :param game_id: Game id to ask for (TmOriginal, TmSunrise or TmNationsESWC), + all of them if omitted + :return: A SessionInfo object + :raises ServerNotFoundException: If no matching server answers + """ + game_ids = [game_id] if game_id else list(self.GAMES) + nonce = secrets.randbits(32) + + with UdpClient() as udp_client: + udp_client.settimeout(self._timeout) + await udp_client.connect((self._host, self._port)) + + for query_game_id in game_ids: + udp_client.send(self.build_session_query(query_game_id, nonce)) + + try: + return await asyncio.wait_for( + self._receive_session(udp_client, nonce), timeout=self._timeout + ) + except asyncio.TimeoutError as e: + raise ServerNotFoundException( + f"No TrackMania session announced by {self._host}:{self._port}" + ) from e + + async def _receive_session(self, udp_client: UdpClient, nonce: int) -> SessionInfo: + while True: + try: + return self.parse_session_reply(await udp_client.recv(), nonce) + except InvalidPacketException: + continue + + async def _query_info(self, version: int) -> bytes: + request_id = secrets.randbelow(0x7FFFFFFF) + 1 + + try: + reader, writer = await asyncio.wait_for( + asyncio.open_connection(self._host, self._port), timeout=self._timeout + ) + except (OSError, asyncio.TimeoutError) as e: + raise ServerNotFoundException( + f"Failed to connect to {self._host}:{self._port}: {e}" + ) from e + + try: + writer.write( + self.build_connection_admin(self._SUBTYPE_QUERY_MODE, version=version) + + self.build_connection_admin( + self._SUBTYPE_INFO_REQUEST, request_id, version=version + ) + ) + await writer.drain() + + return await asyncio.wait_for( + self._receive_info(reader, request_id), timeout=self._timeout + ) + except asyncio.TimeoutError as e: + raise InvalidPacketException("Timeout while waiting for server info") from e + except asyncio.IncompleteReadError as e: + raise InvalidPacketException("Connection closed by the server") from e + except OSError as e: + raise InvalidPacketException(f"Connection error: {e}") from e + finally: + writer.close() + try: + await writer.wait_closed() + except OSError: + pass + + async def _receive_info( + self, reader: asyncio.StreamReader, request_id: int + ) -> bytes: + while True: + length = struct.unpack(" Optional[bytes]: + reader = _Reader(payload) + version = reader.u32() + subtype = reader.u32() + + if subtype == self._SUBTYPE_REFUSED: + raise _RefusedException(version) + + if version not in (self._VERSION_SUNRISE, self._VERSION_NATIONS_ESWC): + raise InvalidPacketException( + f"Unsupported ConnectionAdmin version: {version}" + ) + + if subtype != self._SUBTYPE_INFO: + return None + + reply_id = reader.u32() + data = reader.take(reader.u32()) + + if reply_id == self._NO_INFO: + raise InvalidPacketException("The server has no game info available yet") + + return data if reply_id == request_id else None + + @classmethod + def build_connection_admin( + cls, + subtype: int, + request_id: Optional[int] = None, + version: int = _VERSION_SUNRISE, + ) -> bytes: + """Builds a framed CNetFormConnectionAdmin message.""" + payload = struct.pack(" bytes: + """Builds a CNetFormQuerrySessions datagram for LAN discovery.""" + payload = ( + _pack_string(game_id) + + _pack_string(client_name) + + bytes(6) # Address, not used by the server + + struct.pack(" bytes: + """Builds a checksummed and uncompressed message without length prefix.""" + message = bytearray([0x80 | cls._FLAG_CHECKSUM, message_type]) + message += bytes(4) + payload + message[2:6] = struct.pack(" Tuple[int, bytes]: + """ + Decodes a message without its length prefix. + + The checksum is mandatory: every message of a TrackMania server carries one. + + :return: Message type and (decompressed) payload + """ + if len(message) < 2 or message[0] & 0xF0 != 0x80: + raise InvalidPacketException("Invalid message header") + + flags, message_type = message[0], message[1] + position = 2 + + if flags & cls._FLAG_SEQUENCE: + position += 2 + + if not flags & cls._FLAG_CHECKSUM: + raise InvalidPacketException("Message without checksum") + + if len(message) < position + 4: + raise InvalidPacketException("Truncated message") + + checksum = struct.unpack_from(" cls._MAX_MESSAGE_SIZE: + raise InvalidPacketException(f"Invalid uncompressed size: {size}") + + return message_type, _lzo1x_decompress(message[position + 4 :], size) + + @classmethod + def _checksum(cls, message: bytes, position: int) -> int: + data = bytearray(message) + data[position : position + 4] = bytes(4) + digest = hmac.new(cls._CHECKSUM_KEY, bytes(data), hashlib.md5).digest() + words = struct.unpack("<4I", digest) + key = struct.unpack_from(" SessionInfo: + """ + Parses a CNetFormEnumSessions datagram sent in reply to a session query. + + :param nonce: Nonce of the query, the reply must echo it if given + :raises InvalidPacketException: If the datagram is no matching reply + """ + message_type, payload = cls.decode_message(datagram) + + if message_type != cls._MESSAGE_ENUM_SESSIONS: + raise InvalidPacketException(f"Unexpected message type: {message_type}") + + reader = _Reader(payload) + reply_nonce = reader.u32() + + if nonce is not None and reply_nonce != nonce: + raise InvalidPacketException("Reply to another session query") + + application = reader.string() + game_id = reader.string() + version = reader.string() + host_name = reader.string() + address, port = _read_address(reader) + secondary_address, secondary_port = _read_address(reader) + + return SessionInfo( + game_id=game_id, + game=cls.GAMES.get(game_id, game_id), + version=version, + host_name=host_name, + application=application, + server_address=address, + server_port=port, + secondary_address=secondary_address, + secondary_port=secondary_port, + ) + + @classmethod + def decoration( + cls, game_id: str, decoration_index: int + ) -> Optional[Tuple[str, str]]: + """ + Resolves the decoration index of a challenge. + + The server builds its decoration table from DECORATIONS: every environment + with each decoration plus one entry without decoration (an unassigned id), + sorted case-insensitively by environment, then by decoration name + ("Unassigned" for the missing one). The index is 2 + the table position. + + :return: Environment and decoration (mood), the decoration is empty for + the entry without one; None if the index is unknown + """ + if game_id not in cls.DECORATIONS: + return None + + environments, decorations = cls.DECORATIONS[game_id] + table = sorted( + ( + (environment, decoration) + for environment in environments + for decoration in (*decorations, "") + ), + key=lambda entry: (entry[0].lower(), (entry[1] or "Unassigned").lower()), + ) + position = decoration_index - 2 + + return table[position] if 0 <= position < len(table) else None + + @classmethod + def parse_server_info( + cls, data: bytes, game_id: Optional[str] = None + ) -> ServerInfo: + """ + Parses the server info sent in reply to an info request. + + :param game_id: Game of the server (TmOriginal or TmSunrise) if known, to + resolve the environments. Nations ESWC is recognised by its game tag. + + Layout, one block per class of the serialisation chain:: + + CNetMasterHost + u8 game tag: (tag & 0xE0) == 0, (tag & 0x1F) == 0x07 or 0x09 + u8[4] IP address, reversed byte order + u16 port + str host login + CGameNetServerInfo + str "#SRV#" + "p" player password / "s" spectator password / "f" both + game server info (only for a "#SRV#" login) + wstr unused, always empty + u8 player count, max players, spectator count, max spectators, + ladder mode + wstr server name + u32 n wstr player names[n], i32 ladder rankings[n] + wstr comment + CTrackManiaNetworkServerInfo + u8 game mode (GAME_MODES) + u32 time limit (ms) / points limit / number of laps + u8 number of challenges in the playlist + u32 n challenges[n]: wstr name, u32 decoration index, + u32 gold time, u32 copper price; current challenge first + + The decoration index is resolved to environment and mood, see decoration(). + """ + reader = _Reader(data) + game_tag = reader.u8() + + if game_tag & 0xE0 != 0 or game_tag not in cls._GAME_TAGS: + raise InvalidPacketException( + f"Not a TrackMania Sunrise server (game tag 0x{game_tag:02x})" + ) + + address = ".".join(str(octet) for octet in reversed(reader.take(4))) + port = reader.u16() + server_login = reader.string() + player_login = reader.string() + + # Original and Sunrise share the game tag, Nations ESWC has its own + if cls._GAME_TAGS[game_tag] or game_id not in ("TmOriginal", "TmSunrise"): + game_id = cls._GAME_TAGS[game_tag] + + info = ServerInfo( + name=server_login, + map="", + players=0, + max_players=0, + game_mode="Unknown", + game_id=game_id, + game_tag=game_tag, + server_login=server_login, + server_address=address, + server_port=port, + raw_data=data.hex(), + ) + + if not player_login.startswith("#SRV#"): + return info + + password_flag = player_login[5:6] + info.password_protected = password_flag in ("p", "f") + info.spectator_password_protected = password_flag in ("s", "f") + + reader.wstring() + info.players = reader.u8() + info.max_players = reader.u8() + info.spectators = reader.u8() + info.max_spectators = reader.u8() + info.ladder_mode = reader.u8() + info.ladder_server = info.ladder_mode != 0 + info.name = reader.wstring() + + names = [reader.wstring() for _ in range(reader.count())] + info.player_list = [Player(name, reader.i32()) for name in names] + info.comment = reader.wstring() + + if reader.remaining == 0: + return info + + info.game_mode_id = reader.u8() + info.game_mode = cls.GAME_MODES.get( + info.game_mode_id, f"Unknown ({info.game_mode_id})" + ) + limit = reader.u32() + + if info.game_mode_id in (1, 8): + info.time_limit = limit + elif info.game_mode_id == 7: + info.nb_laps = limit + elif info.game_mode_id in (3, 6): + info.points_limit = limit + + info.nb_challenges = reader.u8() + info.challenges = [ + Challenge( + name=reader.wstring(), + decoration_index=reader.u32(), + gold_time=reader.u32(), + copper_price=reader.u32(), + ) + for _ in range(reader.count()) + ] + + for challenge in info.challenges: + decoration = cls.decoration(game_id, challenge.decoration_index) + + if decoration: + challenge.environment, challenge.mood = decoration + + if info.challenges: + info.map = info.challenges[0].name + info.environment = info.challenges[0].environment + info.mood = info.challenges[0].mood + + return info + + +class _RefusedException(InvalidPacketException): + """The server refused the ConnectionAdmin version (subtype 1).""" + + def __init__(self, server_version: int): + super().__init__(f"The server refused the query (version {server_version})") + self.server_version = server_version + + +def _pack_string(text: str) -> bytes: + data = text.encode("utf-8") + return struct.pack(" Tuple[str, int]: + address = ".".join(str(octet) for octet in reversed(reader.take(4))) + return address, reader.u16() + + +if __name__ == "__main__": + import json + import sys + + async def main_async(): + host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1" + tm = TrackmaniaSunrise(host, TrackmaniaSunrise.DEFAULT_PORT, 5.0) + session = await tm.get_session() + print(json.dumps(session.to_dict(), indent=4, ensure_ascii=False)) + info = await tm.get_info(session.game_id) + print(json.dumps(info.to_dict(), indent=4, ensure_ascii=False)) + + asyncio.run(main_async()) diff --git a/opengsq/responses/trackmania_nations/__init__.py b/opengsq/responses/trackmania_nations/__init__.py index e9a9bd1..df1d457 100644 --- a/opengsq/responses/trackmania_nations/__init__.py +++ b/opengsq/responses/trackmania_nations/__init__.py @@ -1,3 +1,3 @@ -from .server_info import ServerInfo +from .server_info import Challenge, Player, ServerInfo, strip_formatting -__all__ = ["ServerInfo"] +__all__ = ["Challenge", "Player", "ServerInfo", "strip_formatting"] diff --git a/opengsq/responses/trackmania_nations/server_info.py b/opengsq/responses/trackmania_nations/server_info.py index d55b1d0..f6255a4 100644 --- a/opengsq/responses/trackmania_nations/server_info.py +++ b/opengsq/responses/trackmania_nations/server_info.py @@ -1,19 +1,63 @@ -from dataclasses import dataclass, field -from typing import Optional +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass, field +from typing import List, Optional + +# TrackMania text formatting: "$$" is a literal dollar sign, "$" + up to three hex +# digits is a colour, "$l[...]"/"$h[...]" carry a link target and every other +# "$x" is a single-character style code. +_FORMATTING_PATTERN = re.compile( + r"\$(\$|[0-9a-fA-F]{1,3}|[lLhHpP]\[[^\]]*\]|.)", re.DOTALL +) + + +def strip_formatting(text: str) -> str: + """Removes TrackMania formatting codes ($fff, $o, $l[...], ...) from a string.""" + return _FORMATTING_PATTERN.sub( + lambda match: "$" if match.group(1) == "$" else "", text + ) + + +@dataclass +class Player: + """A player connected to the server.""" + + name: str + """Nickname, including TrackMania formatting codes.""" + + ladder_ranking: int + """Ladder ranking (-1 = not ranked, 0 = unknown).""" + + +@dataclass +class Challenge: + """A challenge (map) of the server playlist.""" + + name: str + """Challenge name. The server truncates it to 15 characters.""" + + gold_time: int + """Gold medal time in milliseconds.""" + + copper_price: int + """Copper price (display cost) of the challenge.""" + + environment: str + """Environment (collection) of the challenge, e.g. Stadium.""" @dataclass class ServerInfo: """ - Trackmania Nations Server Information - Erweitert basierend auf Reverse-Engineering der #SRV# Server-Announcement-Payloads + Trackmania Nations/United Forever Server Information """ name: str - """Name of the server.""" + """Name of the server, including TrackMania formatting codes.""" map: str - """Current map being played.""" + """Current challenge (truncated to 15 characters by the server).""" players: int """Current number of players on the server.""" @@ -22,71 +66,111 @@ class ServerInfo: """Maximum number of players the server can hold.""" game_mode: str - """Current game mode (e.g., Time Attack, Rounds, Cup, etc.).""" + """Current game mode (TimeAttack, Rounds, Team, Laps, Stunts or Cup).""" password_protected: bool = False - """Whether the server requires a password.""" + """Whether joining as player requires a password.""" version: Optional[str] = None - """Server/game version.""" + """Not part of the query response, always None.""" - # Erweiterte Felder aus der Dokumentation environment: str = "Unknown" - """Map environment (Stadium, Canyon, Valley, etc.).""" + """Environment of the current challenge, e.g. Stadium.""" comment: str = "" - """Server comment/description.""" + """Server comment.""" server_login: str = "" - """Server login name.""" + """Login of the server host. In LAN mode this is the computer name.""" pc_guid: str = "" - """PC GUID identifier.""" + """Deprecated alias of server_login.""" time_limit: int = 0 - """Time limit in milliseconds.""" + """Time limit in milliseconds (TimeAttack and Stunts).""" nb_laps: int = 0 - """Number of laps for lap-based modes.""" + """Number of laps (Laps mode).""" spectator_slots: int = 0 - """Maximum number of spectator slots.""" + """Deprecated alias of max_spectators.""" build_number: int = 0 - """Game build number.""" + """Not part of the query response, always 0.""" private_server: bool = False - """Whether the server is private.""" + """Deprecated alias of password_protected.""" ladder_server: bool = False - """Whether the server is a ladder server.""" + """Whether the ladder mode is active (ladder_mode != 0).""" status_flags: int = 0 - """Server status flags bitfield.""" + """Not part of the query response, always 0.""" challenge_crc: int = 0 - """Challenge/Map CRC checksum.""" + """Not part of the query response, always 0.""" public_ip: str = "" - """Public IP address of the server.""" + """Not part of the query response, always empty.""" local_ip: str = "" - """Local IP address of the server.""" + """Deprecated alias of server_address.""" + + spectators: int = 0 + """Current number of spectators.""" + + max_spectators: int = 0 + """Maximum number of spectators.""" + + spectator_password_protected: bool = False + """Whether joining as spectator requires a password.""" + + ladder_mode: int = 0 + """Ladder mode (0 = inactive, 1 = forced).""" + + game_mode_id: int = 0 + """Internal game mode id (1 TimeAttack, 3 Rounds, 6 Team, 7 Laps, 8 Stunts, 9 Cup).""" + + points_limit: int = 0 + """Points limit (Rounds, Team and Cup).""" + + pack_mask: str = "" + """Pack mask of the server, e.g. Stadium for Nations servers.""" + + nb_challenges: int = 0 + """Number of challenges in the playlist (capped at 255 by the server).""" + + challenges: List[Challenge] = field(default_factory=list) + """Current challenge followed by the next ones (at most 20).""" + + player_list: List[Player] = field(default_factory=list) + """Players connected to the server.""" + + server_address: str = "" + """IP address the server announces for itself.""" + + server_port: int = 0 + """Port the server announces for itself.""" raw_data: Optional[str] = field(default=None, repr=False) - """Raw response data from the server as hex string.""" + """Decompressed server info payload as hex string.""" + + @property + def plain_name(self) -> str: + """Server name without TrackMania formatting codes.""" + return strip_formatting(self.name) def __str__(self) -> str: """ Returns a human-readable string representation of the server info. """ return ( - f"Trackmania Nations Server: {self.name}\n" + f"Trackmania Nations Server: {self.plain_name}\n" f"Map: {self.map} ({self.environment})\n" f"Players: {self.players}/{self.max_players}\n" + f"Spectators: {self.spectators}/{self.max_spectators}\n" f"Game Mode: {self.game_mode}\n" f"Password Protected: {self.password_protected}\n" - f"Version: {self.version}\n" f"Comment: {self.comment}" ) @@ -94,26 +178,6 @@ def to_dict(self) -> dict: """ Convert to dictionary for JSON serialization, excluding raw_data. """ - return { - "name": self.name, - "map": self.map, - "players": self.players, - "max_players": self.max_players, - "game_mode": self.game_mode, - "password_protected": self.password_protected, - "version": self.version, - "environment": self.environment, - "comment": self.comment, - "server_login": self.server_login, - "pc_guid": self.pc_guid, - "time_limit": self.time_limit, - "nb_laps": self.nb_laps, - "spectator_slots": self.spectator_slots, - "build_number": self.build_number, - "private_server": self.private_server, - "ladder_server": self.ladder_server, - "status_flags": self.status_flags, - "challenge_crc": self.challenge_crc, - "public_ip": self.public_ip, - "local_ip": self.local_ip, - } + result = asdict(self) + result.pop("raw_data", None) + return result diff --git a/opengsq/responses/trackmania_sunrise/__init__.py b/opengsq/responses/trackmania_sunrise/__init__.py new file mode 100644 index 0000000..03f1b16 --- /dev/null +++ b/opengsq/responses/trackmania_sunrise/__init__.py @@ -0,0 +1,3 @@ +from .server_info import Challenge, Player, ServerInfo, SessionInfo + +__all__ = ["Challenge", "Player", "ServerInfo", "SessionInfo"] diff --git a/opengsq/responses/trackmania_sunrise/server_info.py b/opengsq/responses/trackmania_sunrise/server_info.py new file mode 100644 index 0000000..b932794 --- /dev/null +++ b/opengsq/responses/trackmania_sunrise/server_info.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import List, Optional + +from opengsq.responses.trackmania_nations import strip_formatting + + +@dataclass +class Player: + """A player connected to the server.""" + + name: str + """Nickname, including TrackMania formatting codes.""" + + ladder_ranking: int + """Ladder ranking as sent by the server.""" + + +@dataclass +class Challenge: + """A challenge (map) of the server playlist.""" + + name: str + """Challenge name. The server truncates it to 15 characters.""" + + gold_time: int + """Gold medal time in milliseconds.""" + + copper_price: int + """Copper price (display cost) of the challenge.""" + + decoration_index: int + """ + 2 + index of the challenge decoration (environment and mood) in the + decoration table of the server, 0 if unknown. See TrackmaniaSunrise.decoration(). + """ + + environment: str = "" + """Environment of the challenge, e.g. Island. Empty if the game is unknown.""" + + mood: str = "" + """Decoration (mood) of the challenge, e.g. Night. Empty if unknown.""" + + +@dataclass +class ServerInfo: + """ + TrackMania Original/Sunrise/Nations ESWC Server Information + """ + + name: str + """Name of the server, including TrackMania formatting codes.""" + + map: str + """Current challenge (truncated to 15 characters by the server).""" + + players: int + """Current number of players on the server.""" + + max_players: int + """Maximum number of players the server can hold.""" + + game_mode: str + """Current game mode (TimeAttack, Rounds, Team, Laps or Stunts).""" + + game_id: str = "" + """ + Game the server runs. TmNationsESWC is recognised by the game tag, + TmOriginal and TmSunrise only when passed to get_info() (use get_session() + to find out), otherwise it is empty. + """ + + environment: str = "" + """Environment of the current challenge, e.g. Island. Empty if the game is unknown.""" + + mood: str = "" + """Decoration (mood) of the current challenge, e.g. Night.""" + + game_tag: int = 0 + """Game tag of the server info (0x07 Original/Sunrise, 0x09 Nations ESWC).""" + + protocol_version: int = 0 + """ConnectionAdmin version of the server (4 Original/Sunrise, 5 Nations ESWC).""" + + password_protected: bool = False + """Whether joining as player requires a password.""" + + spectator_password_protected: bool = False + """Whether joining as spectator requires a password.""" + + spectators: int = 0 + """Current number of spectators.""" + + max_spectators: int = 0 + """Maximum number of spectators.""" + + ladder_mode: int = 0 + """Ladder mode (0 = inactive).""" + + ladder_server: bool = False + """Whether the ladder mode is active (ladder_mode != 0).""" + + comment: str = "" + """Server comment.""" + + server_login: str = "" + """Login of the server host. In LAN mode this is the computer name.""" + + server_address: str = "" + """IP address the server announces for itself.""" + + server_port: int = 0 + """Port the server announces for itself.""" + + game_mode_id: int = 0 + """Internal game mode id (1 TimeAttack, 3 Rounds, 6 Team, 7 Laps, 8 Stunts).""" + + time_limit: int = 0 + """Time limit in milliseconds (TimeAttack and Stunts).""" + + nb_laps: int = 0 + """Number of laps (Laps mode).""" + + points_limit: int = 0 + """Points limit (Rounds and Team).""" + + nb_challenges: int = 0 + """Number of challenges in the playlist (capped at 255 by the server).""" + + challenges: List[Challenge] = field(default_factory=list) + """Current challenge followed by the next ones (at most 20).""" + + player_list: List[Player] = field(default_factory=list) + """Players connected to the server.""" + + raw_data: Optional[str] = field(default=None, repr=False) + """Decompressed server info payload as hex string.""" + + @property + def plain_name(self) -> str: + """Server name without TrackMania formatting codes.""" + return strip_formatting(self.name) + + def __str__(self) -> str: + """ + Returns a human-readable string representation of the server info. + """ + return ( + f"TrackMania Server: {self.plain_name}\n" + f"Map: {self.map} ({self.environment or 'Unknown'})\n" + f"Players: {self.players}/{self.max_players}\n" + f"Spectators: {self.spectators}/{self.max_spectators}\n" + f"Game Mode: {self.game_mode}\n" + f"Password Protected: {self.password_protected}\n" + f"Comment: {self.comment}" + ) + + def to_dict(self) -> dict: + """ + Convert to dictionary for JSON serialization, excluding raw_data. + """ + result = asdict(self) + result.pop("raw_data", None) + return result + + +@dataclass +class SessionInfo: + """ + LAN session announced by a server in reply to a session query + (CNetFormEnumSessions). + """ + + game_id: str + """Game the server runs: TmOriginal, TmSunrise or TmNationsESWC.""" + + game: str + """Name of the game, e.g. TrackMania Sunrise.""" + + version: str + """Version string of the server's network layer, e.g. 1.043.""" + + host_name: str + """Computer name of the server host.""" + + application: str + """Name of the network application, GameNet for the game server.""" + + server_address: str + """IP address the server announces for itself.""" + + server_port: int + """Game port the server announces for itself.""" + + secondary_address: str = "" + """Second address announced by the server (usually the same).""" + + secondary_port: int = 0 + """Port of the second announced address.""" + + def to_dict(self) -> dict: + """Convert to dictionary for JSON serialization.""" + return asdict(self) diff --git a/tests/protocols/test_trackmania_nations.py b/tests/protocols/test_trackmania_nations.py index 3bf29b2..0e9cc13 100644 --- a/tests/protocols/test_trackmania_nations.py +++ b/tests/protocols/test_trackmania_nations.py @@ -1,5 +1,10 @@ +import asyncio +import struct + import pytest +from opengsq.exceptions import InvalidPacketException from opengsq.protocols.trackmania_nations import TrackmaniaNations +from opengsq.responses.trackmania_nations import strip_formatting from ..result_handler import ResultHandler @@ -8,13 +13,173 @@ handler.enable_save = True # Test server configuration -SERVER_IP = "172.29.100.29" +SERVER_IP = "10.10.100.212" SERVER_PORT = 2350 tmn = TrackmaniaNations(host=SERVER_IP, port=SERVER_PORT) +# Packets captured from a real game client and server (request id 0x00413DD5) +CLIENT_QUERY_MODE = bytes.fromhex("0e000000820399f895580700000008000000") +CLIENT_INFO_REQUEST = bytes.fromhex("1200000082033bd464400700000007000000d53d4100") +SERVER_INFO_REPLY = bytes.fromhex( + "9b0000008303681ac2009b0000000a0700000006000000d53d41008b5c00000b2d1d641dac2e09" + "0900000050432d636539623063050000002353525623500204000106000600094402074b617761" + "626f6e6761075001075374616469756d0100002c6c0001ffffffff940602e09304000178030001" + "080000004230322d526163657a710000b9020079020374040b000040070000005374616469756d" + "110000" +) + +# Server info of a dedicated server (Rounds, 15 challenges, UTF-8 comment with BOM) +DEDICATED_SERVER_INFO = bytes.fromhex( + "2d0100007f2e090c0000004c4956452d54455354535256050000002353525623000000000006" + "00040017000000246f2466303044474e20246666664c6976652d546573740700000053746164" + "69756d000000001d000000efbbbf4c6976652d54657374204b6f6d6d656e74617220c3a4c3b6" + "c3bc031e0000000f0f000000080000004230312d52616365986c0000a9070008000000423032" + "2d526163657a710000b90200080000004230332d52616365767000002105000d000000423034" + "2d4163726f6261746963de350000f60300080000004230352d52616365b26b00000a06000c00" + "00004230362d4f62737461636c65f67200004d0400080000004230372d52616365be7d0000db" + "06000d0000004230382d456e647572616e6365b8a501006f07000d0000004230392d4163726f" + "6261746963163a00000b0600090000004231302d5370656564449d00007a0800080000004231" + "312d52616365a4830000c60700080000004231322d526163658cb90000d104000c0000004231" + "332d4f62737461636c65226a00009d0400090000004231342d53706565647689000048070008" + "0000004231352d5261636540ab0000b307000100000003000000000000400700000053746164" + "69756d" +) + + +def _decode_captured_info() -> bytes: + message_type, payload = TrackmaniaNations.decode_message(SERVER_INFO_REPLY[4:]) + assert message_type == 3 + + return tmn._read_info_reply(payload, 0x00413DD5) + @pytest.mark.asyncio async def test_get_info(): result = await tmn.get_info() await handler.save_result("test_get_info", result) + + +def test_build_requests_match_game_client(): + assert TrackmaniaNations.build_connection_admin(8) == CLIENT_QUERY_MODE + assert ( + TrackmaniaNations.build_connection_admin(7, 0x00413DD5) == CLIENT_INFO_REQUEST + ) + + +def test_parse_captured_server_info(): + info = TrackmaniaNations.parse_server_info(_decode_captured_info()) + + assert info.name == "Kawabonga" + assert info.map == "B02-Race" + assert info.environment == "Stadium" + assert (info.players, info.max_players) == (1, 6) + assert (info.spectators, info.max_spectators) == (0, 6) + assert info.game_mode == "TimeAttack" + assert info.time_limit == 300000 + assert info.password_protected is False + assert info.spectator_password_protected is False + assert info.ladder_mode == 0 + assert info.pack_mask == "Stadium" + assert info.server_login == "PC-ce9b0c" + assert (info.server_address, info.server_port) == ("172.29.100.29", 2350) + assert info.comment == "" + assert [(p.name, p.ladder_ranking) for p in info.player_list] == [("Kawabonga", -1)] + assert info.nb_challenges == 1 + assert len(info.challenges) == 1 + assert info.challenges[0].gold_time == 29050 + assert info.challenges[0].copper_price == 697 + + +def test_parse_dedicated_server_info(): + info = TrackmaniaNations.parse_server_info(DEDICATED_SERVER_INFO) + + assert info.name == "$o$f00DGN $fffLive-Test" + assert info.plain_name == "DGN Live-Test" + assert info.comment == "Live-Test Kommentar äöü" + assert (info.players, info.max_players) == (0, 6) + assert (info.spectators, info.max_spectators) == (0, 4) + assert (info.game_mode, info.game_mode_id, info.points_limit) == ("Rounds", 3, 30) + assert info.time_limit == 0 + assert info.player_list == [] + assert info.nb_challenges == 15 + assert [c.name for c in info.challenges[:2]] == ["B01-Race", "B02-Race"] + assert info.challenges[7].name == "B08-Endurance" + assert info.challenges[7].gold_time == 107960 + assert {c.environment for c in info.challenges} == {"Stadium"} + assert info.map == "B01-Race" + + +def test_password_flags_from_server_login(): + data = bytearray(_decode_captured_info()) + login = data.index(b"#SRV#") + # Replace "#SRV#" by "#SRV#f" (player and spectator password) + data[login - 4 : login + 5] = struct.pack(" bytes: + message_type, payload = TrackmaniaSunrise.decode_message(SERVER_INFO_REPLY[4:]) + assert message_type == 3 + + return tms._read_info_reply(payload, 0x1234) + + +def _server_info( + name: str = "Sunrise LAN Server", + login: str = "#SRV#", + game_tag: int = 0x07, + players=(), +) -> bytes: + """Builds a server info payload like the server serialises it.""" + + def wstr(text: str) -> bytes: + data = text.encode("utf-8") + if text != text.encode("ascii", "ignore").decode(): + data = b"\xef\xbb\xbf" + data + return struct.pack(" bytes: + return struct.pack("