1+ from __future__ import annotations
2+
3+ import os
4+ from opengsq .binary_reader import BinaryReader
5+ from opengsq .exceptions import InvalidPacketException
6+ from opengsq .protocol_base import ProtocolBase
7+ from opengsq .protocol_socket import UdpClient
8+ from opengsq .responses .flatout2 import Status
9+
10+
11+ class Flatout2 (ProtocolBase ):
12+ """
13+ This class represents the Flatout 2 Protocol. It provides methods to interact with Flatout 2 game servers.
14+ The protocol uses broadcast packets to discover and query servers.
15+ """
16+
17+ full_name = "Flatout 2 Protocol"
18+ FLATOUT2_PORT = 23757 # Default broadcast port for Flatout 2
19+
20+ # Protocol specific constants
21+ REQUEST_HEADER = b"\x22 \x00 "
22+ RESPONSE_HEADER = b"\x59 \x00 "
23+ GAME_IDENTIFIER = b"FO14"
24+ SESSION_ID = b"\x99 \x72 \xcc \x8f "
25+ COMMAND_QUERY = b"\x18 \x0c "
26+ PACKET_END = b"\x2e \x55 \x19 \xb4 \xe1 \x4f \x81 \x4a "
27+
28+ def __init__ (self , host : str , port : int = FLATOUT2_PORT , timeout : float = 5.0 ):
29+ """
30+ Initialize the Flatout 2 protocol handler.
31+
32+ :param host: The broadcast address (usually '255.255.255.255' for LAN)
33+ :param port: The port to use (default: 23757)
34+ :param timeout: Connection timeout in seconds
35+ """
36+ if port != self .FLATOUT2_PORT :
37+ raise ValueError (f"Flatout 2 protocol requires port { self .FLATOUT2_PORT } )
38+ super ().__init__ (host , self .FLATOUT2_PORT , timeout )
39+ self ._allow_broadcast = True
40+
41+ async def get_status (self ) -> Status :
42+ """
43+ Asynchronously retrieves the status of Flatout 2 servers via broadcast.
44+ Expects a response packet with server information.
45+
46+ :return: A Status object containing the status of the game server.
47+ """
48+ # Build the request packet
49+ request_data = (
50+ self .REQUEST_HEADER + # Protocol header
51+ self .SESSION_ID + # Session ID
52+ b"\x00 " * 4 + # Padding pre-identifier
53+ self .GAME_IDENTIFIER + # "FO14"
54+ b"\x00 " * 8 + # Padding post-identifier
55+ self .COMMAND_QUERY + # Query command
56+ b"\x00 \x00 \x22 \x00 " + # Command data
57+ self .PACKET_END # Standard packet end
58+ )
59+
60+ # Send broadcast and receive response
61+ data = await UdpClient .communicate (self , request_data , source_port = self .FLATOUT2_PORT )
62+
63+ # Debug output for packet analysis
64+ print (f"Response header: { data [:2 ].hex ()} { self .RESPONSE_HEADER .hex ()} )
65+ print (f"Session ID: { data [2 :6 ].hex ()} )
66+ print (f"Game ID: { data [10 :14 ]} { self .GAME_IDENTIFIER } )
67+ print (f"Full packet length: { len (data )} )
68+
69+ # Verify response packet
70+ if not self ._verify_packet (data ):
71+ raise InvalidPacketException ("Invalid response packet received" )
72+
73+ br = BinaryReader (data )
74+ return self ._parse_response (br )
75+
76+ def _verify_packet (self , data : bytes ) -> bool :
77+ """
78+ Verifies that a packet is a valid Flatout 2 response.
79+
80+ :param data: The packet data to verify
81+ :return: True if the packet is valid, False otherwise
82+ """
83+ if len (data ) < 14 : # Minimum length for header + session ID + padding + game ID
84+ print (f"Packet too short: { len (data )} )
85+ return False
86+
87+ # Check response header
88+ header_matches = data .startswith (self .RESPONSE_HEADER )
89+ if not header_matches :
90+ print (f"Header mismatch: got { data [:2 ].hex ()} { self .RESPONSE_HEADER .hex ()} )
91+ return False
92+
93+ # Check game identifier (position 10-14, after session ID and padding)
94+ game_id = data [10 :14 ]
95+ game_id_matches = game_id == self .GAME_IDENTIFIER
96+ if not game_id_matches :
97+ print (f"Game ID mismatch: got { game_id } { self .GAME_IDENTIFIER } )
98+ return False
99+
100+ return True
101+
102+ def _read_utf16_string (self , br : BinaryReader ) -> str :
103+ """
104+ Reads a UTF-16 encoded string from the binary reader.
105+
106+ :param br: The binary reader to read from
107+ :return: The decoded string
108+ """
109+ bytes_list = []
110+ while True :
111+ # Read two bytes at a time (UTF-16)
112+ char_bytes = br .read_bytes (2 )
113+ if char_bytes == b"\x00 \x00 " : # End of string
114+ break
115+ bytes_list .extend (char_bytes )
116+
117+ return bytes (bytes_list ).decode ('utf-16-le' ).strip ()
118+
119+ def _parse_response (self , br : BinaryReader ) -> Status :
120+ """
121+ Parses the binary response into a Status object.
122+ The response contains UTF-16 encoded strings and various server information.
123+
124+ :param br: BinaryReader containing the response data
125+ :return: A Status object containing the parsed information
126+ """
127+ # Skip header (2), session ID (4), padding (4), game ID (4), padding (8), command (4), data (2), and packet end (8)
128+ br .read_bytes (36 ) # Skip to the server data section
129+
130+ info = {}
131+
132+ try :
133+ # Read server name (UTF-16 encoded)
134+ server_name = self ._read_utf16_string (br )
135+ info ["hostname" ] = server_name
136+
137+ # Read server information
138+ timestamp = br .read_long_long () # Server timestamp
139+ info ["timestamp" ] = str (timestamp )
140+
141+ server_flags = br .read_long (unsigned = True ) # Server configuration flags
142+ info ["flags" ] = str (server_flags )
143+
144+ # Skip reserved bytes
145+ br .read_bytes (8 )
146+
147+ # Read server status
148+ status_flags = br .read_long (unsigned = True )
149+ info ["status" ] = str (status_flags )
150+
151+ # Read server configuration
152+ config = br .read_bytes (12 ) # Remaining configuration data
153+ info ["config" ] = config .hex ()
154+
155+ except Exception as e :
156+ print (f"Error parsing response: { e } )
157+
158+ return Status (info = info )
159+
160+
161+ if __name__ == "__main__" :
162+ import asyncio
163+
164+ async def main_async ():
165+ # Use broadcast address for LAN discovery
166+ flatout2 = Flatout2 (host = "255.255.255.255" , port = 23757 , timeout = 5.0 )
167+ status = await flatout2 .get_status ()
168+ print (status )
0 commit comments