Последняя активность 1772128903

Use this script to send rcon commands to a game server supporting Q3 Rcon

Версия 2095a7db01b6d279db29b0d227ccdc97cc7b857b

q3rcon.py Исходник
1#!/usr/bin/env python3
2
3# MIT License
4#
5# Copyright (c) 2025 Onyx and Iris
6#
7# Permission is hereby granted, free of charge, to any person obtaining a copy
8# of this software and associated documentation files (the "Software"), to deal
9# in the Software without restriction, including without limitation the rights
10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11# copies of the Software, and to permit persons to whom the Software is
12# furnished to do so, subject to the following conditions:
13#
14# The above copyright notice and this permission notice shall be included in all
15# copies or substantial portions of the Software.
16#
17# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23# SOFTWARE.
24
25"""
26rcon.py
27
28This script provides functionality to interact with a remote server using the RCON (Remote Console) protocol.
29
30It includes classes to encode and decode RCON packets, send commands to the server, and handle responses.
31The script also provides a command-line interface for sending RCON commands to a server.
32
33Classes:
34 Packet: Base class for RCON packets.
35 Request: Class to encode a request packet for RCON communication.
36 ResponseFragment: Class to build and decode a response packet from multiple fragments received from the RCON server.
37
38Functions:
39 send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str:
40 Sends an RCON command to the server and returns the response.
41 parse_args() -> argparse.Namespace:
42 Parses command-line arguments.
43 main():
44 Main function to handle command-line arguments and interact with a remote server using RCON protocol.
45
46Usage:
47 To send a single command to the server:
48 python rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command>
49 To specify multiple commands:
50 python rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command1> <command2> ...
51
52 To use interactive mode:
53 python rcon.py --host <server_host> --port <server_port> --password <rcon_password> -i
54"""
55
56import argparse
57import logging
58import re
59import socket
60
61logger = logging.getLogger(__name__)
62
63
64class Packet:
65 MAGIC = b"\xff\xff\xff\xff"
66
67
68class Request(Packet):
69 """
70 Class used to encode a request packet for remote console (RCON) communication.
71
72 Attributes:
73 password (str): The password used for authentication with the RCON server.
74
75 Methods:
76 _header() -> bytes:
77 Returns the header bytes that should be prefixed to the request packet.
78 from_credentials(password: str) -> Request:
79 Class method to create a Request instance from the given password.
80 to_bytes(cmd: str) -> bytes:
81 Converts the request to bytes format, including the command to be sent to the server.
82 """
83
84 def __init__(self, password: str):
85 self.password = password
86
87 def _header(self) -> bytes:
88 return Packet.MAGIC + b"rcon"
89
90 @classmethod
91 def from_credentials(cls, password: str) -> "Request":
92 return cls(password)
93
94 def to_bytes(self, cmd: str) -> bytes:
95 return self._header() + f" {self.password} {cmd}\n".encode()
96
97
98class ResponseFragment(Packet):
99 """
100 Class used to build and decode a response packet from multiple fragments received from the RCON server.
101
102 Methods:
103 _header() -> bytes:
104 Returns the header bytes that should be prefixed to the response fragment.
105 from_bytes(data: bytes) -> str:
106 Class method to create a ResponseFragment instance from the given bytes data.
107 It checks if the data starts with the expected header and extracts the response content.
108 If the data does not start with the expected header, it raises a ValueError.
109 """
110
111 def _header(self) -> bytes:
112 return self.MAGIC + b"print\n"
113
114 @classmethod
115 def from_bytes(cls, data: bytes) -> str:
116 instance = cls()
117 if not data.startswith(instance._header()):
118 raise ValueError("Invalid fragment: does not start with expected header")
119 return data.removeprefix(instance._header()).decode()
120
121
122def send(sock: socket.socket, args: argparse.Namespace, cmd: str) -> str:
123 """
124 Send a single RCON (Remote Console) command to the server and return the response.
125
126 This function sends a command to a game server using the RCON protocol over a UDP socket.
127 It waits for the server's response, collects it in fragments if necessary, and returns the complete response.
128
129 Args:
130 sock (socket.socket): The UDP socket used for communication with the server.
131 args (argparse.Namespace): The command-line arguments containing host, port, and password.
132 cmd (str): The command to be sent to the server.
133 Returns:
134 str: The complete response from the server after sending the command.
135 Raises:
136 ValueError: If any of the response fragments received from the server are invalid.
137 """
138 raw_packet = Request.from_credentials(args.password).to_bytes(cmd)
139 logger.debug("Sending packet: %s", raw_packet)
140 sock.sendto(raw_packet, (args.host, args.port))
141
142 fragments = []
143 try:
144 while resp := sock.recv(2048):
145 try:
146 fragment = ResponseFragment.from_bytes(resp)
147 except ValueError:
148 logger.debug("Received invalid fragment, skipping: %s", resp)
149 continue
150 fragments.append(fragment)
151 except socket.timeout:
152 logger.debug("Finished receiving fragments (socket timeout)")
153
154 return "".join(fragments)
155
156
157def parse_args() -> argparse.Namespace:
158 parser = argparse.ArgumentParser()
159 parser.add_argument("--host", "-H", default="localhost")
160 parser.add_argument("--port", "-P", type=int, default=27960)
161 parser.add_argument("--password", "-p", default="")
162 parser.add_argument("--timeout", "-t", type=float, default=0.2)
163 parser.add_argument("--interactive", "-i", action="store_true")
164 parser.add_argument(
165 "--loglevel",
166 "-l",
167 default="info",
168 choices=["debug", "info", "warning", "error", "critical"],
169 help="Set the logging level (default: info)",
170 )
171 parser.add_argument("cmds", nargs="+", default=["status"])
172 args = parser.parse_args()
173 return args
174
175
176def main(args: argparse.Namespace):
177 """
178 Main function to handle command-line arguments and interact with a remote server using RCON protocol.
179 This function parses command-line arguments, creates a UDP socket, and sends requests to a remote server.
180 It supports both interactive and non-interactive modes.
181 In interactive mode, it continuously prompts the user for commands until the user quits by entering 'Q'.
182 In non-interactive mode, it sends a single command to the server and prints the response.
183 Args:
184 args (argparse.Namespace): The command-line arguments containing host, port, and password.
185 Returns:
186 None
187 """
188
189 def cleaned_response(resp: str) -> str:
190 return re.sub(r"\^[0-9]", "", resp)
191
192 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
193 sock.settimeout(args.timeout)
194
195 if args.interactive:
196 print("Entering interactive mode. Type 'Q' to quit.")
197 while cmd := input("cmd: ").strip():
198 if cmd == "Q":
199 break
200
201 if resp := send(sock, args, cmd):
202 print(cleaned_response(resp))
203 else:
204 for cmd in args.cmds:
205 if resp := send(sock, args, cmd):
206 print(cleaned_response(resp))
207
208
209if __name__ == "__main__":
210 args = parse_args()
211
212 logging.basicConfig(level=getattr(logging, args.loglevel.upper()))
213
214 main(args)
215