Rewizja Q3 Rcon CLI - single file, zero dependencies

onyx_online zrewidował ten Gist 1772128903. Przejdź do rewizji

1 file changed, 30 insertions, 30 deletions

q3rcon.py

			@@ -62,7 +62,7 @@ logger = logging.getLogger(__name__)
62	62
63	63
64	64		class Packet:
65		-	MAGIC = b"\xff\xff\xff\xff"
	65	+	MAGIC = b'\xff\xff\xff\xff'
66	66
67	67
68	68		class Request(Packet):
			@@ -85,14 +85,14 @@ class Request(Packet):
85	85		self.password = password
86	86
87	87		def _header(self) -> bytes:
88		-	return Packet.MAGIC + b"rcon"
	88	+	return Packet.MAGIC + b'rcon'
89	89
90	90		@classmethod
91		-	def from_credentials(cls, password: str) -> "Request":
	91	+	def from_credentials(cls, password: str) -> 'Request':
92	92		return cls(password)
93	93
94	94		def to_bytes(self, cmd: str) -> bytes:
95		-	return self._header() + f" {self.password} {cmd}\n".encode()
	95	+	return self._header() + f' {self.password} {cmd}\n'.encode()
96	96
97	97
98	98		class ResponseFragment(Packet):
			@@ -109,13 +109,13 @@ class ResponseFragment(Packet):
109	109		"""
110	110
111	111		def _header(self) -> bytes:
112		-	return self.MAGIC + b"print\n"
	112	+	return self.MAGIC + b'print\n'
113	113
114	114		@classmethod
115	115		def from_bytes(cls, data: bytes) -> str:
116	116		instance = cls()
117	117		if not data.startswith(instance._header()):
118		-	raise ValueError("Invalid fragment: does not start with expected header")
	118	+	raise ValueError('Invalid fragment: does not start with expected header')
119	119		return data.removeprefix(instance._header()).decode()
120	120
121	121
			@@ -136,7 +136,7 @@ def send(sock: socket.socket, args: argparse.Namespace, cmd: str) -> str:
136	136		ValueError: If any of the response fragments received from the server are invalid.
137	137		"""
138	138		raw_packet = Request.from_credentials(args.password).to_bytes(cmd)
139		-	logger.debug("Sending packet: %s", raw_packet)
	139	+	logger.debug('Sending packet: %s', raw_packet)
140	140		sock.sendto(raw_packet, (args.host, args.port))
141	141
142	142		fragments = []
			@@ -145,30 +145,30 @@ def send(sock: socket.socket, args: argparse.Namespace, cmd: str) -> str:
145	145		try:
146	146		fragment = ResponseFragment.from_bytes(resp)
147	147		except ValueError:
148		-	logger.debug("Received invalid fragment, skipping: %s", resp)
	148	+	logger.debug('Received invalid fragment, skipping: %s', resp)
149	149		continue
150	150		fragments.append(fragment)
151	151		except socket.timeout:
152		-	logger.debug("Finished receiving fragments (socket timeout)")
	152	+	logger.debug('Finished receiving fragments (socket timeout)')
153	153
154		-	return "".join(fragments)
	154	+	return ''.join(fragments)
155	155
156	156
157	157		def parse_args() -> argparse.Namespace:
158	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")
	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	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)",
	165	+	'--loglevel',
	166	+	'-l',
	167	+	default='info',
	168	+	choices=['debug', 'info', 'warning', 'error', 'critical'],
	169	+	help='Set the logging level (default: info)',
170	170		)
171		-	parser.add_argument("cmds", nargs="+", default=["status"])
	171	+	parser.add_argument('cmds', nargs='+', default=['status'])
172	172		args = parser.parse_args()
173	173		return args
174	174
			@@ -178,24 +178,24 @@ def main(args: argparse.Namespace):
178	178		Main function to handle command-line arguments and interact with a remote server using RCON protocol.
179	179		This function parses command-line arguments, creates a UDP socket, and sends requests to a remote server.
180	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	181		Args:
184		-	args (argparse.Namespace): The command-line arguments containing host, port, and password.
185		-	Returns:
186		-	None
	182	+	args (argparse.Namespace): The command-line arguments containing
	183	+	host, port, password, timeout, interactive mode flag, log level, and commands to be sent.
	184	+	Behavior:
	185	+	- If interactive mode is enabled, it prompts the user for commands until 'Q' is entered.
	186	+	- If interactive mode is not enabled, it sends each command from the list of commands provided in the command-line arguments and prints the responses.
187	187		"""
188	188
189	189		def cleaned_response(resp: str) -> str:
190		-	return re.sub(r"\^[0-9]", "", resp)
	190	+	return re.sub(r'\^[0-9]', '', resp)
191	191
192	192		with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
193	193		sock.settimeout(args.timeout)
194	194
195	195		if args.interactive:
196	196		print("Entering interactive mode. Type 'Q' to quit.")
197		-	while cmd := input("cmd: ").strip():
198		-	if cmd == "Q":
	197	+	while cmd := input('cmd: ').strip():
	198	+	if cmd == 'Q':
199	199		break
200	200
201	201		if resp := send(sock, args, cmd):
			@@ -206,7 +206,7 @@ def main(args: argparse.Namespace):
206	206		print(cleaned_response(resp))
207	207
208	208
209		-	if __name__ == "__main__":
	209	+	if __name__ == '__main__':
210	210		args = parse_args()
211	211
212	212		logging.basicConfig(level=getattr(logging, args.loglevel.upper()))

onyx_online zrewidował ten Gist 1772014761. Przejdź do rewizji

1 file changed, 4 insertions, 4 deletions

q3rcon.py

			@@ -23,7 +23,7 @@
23	23		# SOFTWARE.
24	24
25	25		"""
26		-	rcon.py
	26	+	q3rcon.py
27	27
28	28		This script provides functionality to interact with a remote server using the RCON (Remote Console) protocol.
29	29
			@@ -45,12 +45,12 @@ Functions:
45	45
46	46		Usage:
47	47		To send a single command to the server:
48		-	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command>
	48	+	python q3rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command>
49	49		To specify multiple commands:
50		-	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command1> <command2> ...
	50	+	python q3rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command1> <command2> ...
51	51
52	52		To use interactive mode:
53		-	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> -i
	53	+	python q3rcon.py --host <server_host> --port <server_port> --password <rcon_password> -i
54	54		"""
55	55
56	56		import argparse

onyx_online zrewidował ten Gist 1772014665. Przejdź do rewizji

1 file changed, 3 insertions, 3 deletions

rcon.py zmieniono nazwę na q3rcon.py

			@@ -157,8 +157,8 @@ def send(sock: socket.socket, args: argparse.Namespace, cmd: str) -> str:
157	157		def parse_args() -> argparse.Namespace:
158	158		parser = argparse.ArgumentParser()
159	159		parser.add_argument("--host", "-H", default="localhost")
160		-	parser.add_argument("--port", "-P", type=int, default=28960)
161		-	parser.add_argument("--password", "-p", default="secret")
	160	+	parser.add_argument("--port", "-P", type=int, default=27960)
	161	+	parser.add_argument("--password", "-p", default="")
162	162		parser.add_argument("--timeout", "-t", type=float, default=0.2)
163	163		parser.add_argument("--interactive", "-i", action="store_true")
164	164		parser.add_argument(
			@@ -211,4 +211,4 @@ if __name__ == "__main__":
211	211
212	212		logging.basicConfig(level=getattr(logging, args.loglevel.upper()))
213	213
214		-	main(args)
	214	+	main(args)

onyx_online zrewidował ten Gist 1771934370. Przejdź do rewizji

1 file changed, 85 insertions, 84 deletions

rcon.py

			@@ -33,7 +33,7 @@ The script also provides a command-line interface for sending RCON commands to a
33	33		Classes:
34	34		Packet: Base class for RCON packets.
35	35		Request: Class to encode a request packet for RCON communication.
36		-	ResponseBuilder: Class to build and decode a response packet from multiple fragments.
	36	+	ResponseFragment: Class to build and decode a response packet from multiple fragments received from the RCON server.
37	37
38	38		Functions:
39	39		send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str:
			@@ -46,18 +46,23 @@ Functions:
46	46		Usage:
47	47		To send a single command to the server:
48	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> ...
49	51
50	52		To use interactive mode:
51		-	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> --interactive
	53	+	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> -i
52	54		"""
53	55
54	56		import argparse
	57	+	import logging
55	58		import re
56	59		import socket
57	60
	61	+	logger = logging.getLogger(__name__)
	62	+
58	63
59	64		class Packet:
60		-	MAGIC = bytearray([0xFF, 0xFF, 0xFF, 0xFF])
	65	+	MAGIC = b"\xff\xff\xff\xff"
61	66
62	67
63	68		class Request(Packet):
			@@ -69,14 +74,11 @@ class Request(Packet):
69	74
70	75		Methods:
71	76		_header() -> bytes:
72		-	Generates the header for the RCON packet.
73		-
74		-	encode(cmd: str) -> bytes:
75		-	Encodes the command into a byte string suitable for sending to the RCON server.
76		-	Args:
77		-	cmd (str): The command to be sent to the RCON server.
78		-	Returns:
79		-	bytes: The encoded command as a byte string.
	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.
80	82		"""
81	83
82	84		def __init__(self, password: str):
			@@ -85,48 +87,39 @@ class Request(Packet):
85	87		def _header(self) -> bytes:
86	88		return Packet.MAGIC + b"rcon"
87	89
88		-	def encode(self, cmd: str) -> bytes:
89		-	return self._header() + f" {self.password} {cmd}".encode()
	90	+	@classmethod
	91	+	def from_credentials(cls, password: str) -> "Request":
	92	+	return cls(password)
90	93
	94	+	def to_bytes(self, cmd: str) -> bytes:
	95	+	return self._header() + f" {self.password} {cmd}\n".encode()
91	96
92		-	class ResponseBuilder(Packet):
93		-	"""
94		-	Class used to build and decode a response packet from multiple fragments.
95	97
96		-	Attributes:
97		-	_fragments (bytearray): A bytearray to store the concatenated fragments.
	98	+	class ResponseFragment(Packet):
	99	+	"""
	100	+	Class used to build and decode a response packet from multiple fragments received from the RCON server.
98	101
99	102		Methods:
100	103		_header() -> bytes:
101		-	Returns the header bytes that each fragment should start with.
102		-
103		-	is_valid_fragment(fragment: bytes) -> bool:
104		-	Checks if the given fragment starts with the expected header.
105		-
106		-	add_fragment(fragment: bytes):
107		-	Adds a fragment to the internal storage after removing the header.
108		-
109		-	build() -> str:
110		-	Decodes and returns the concatenated fragments as a string.
	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.
111	109		"""
112	110
113		-	def __init__(self):
114		-	self._fragments = bytearray()
115		-
116	111		def _header(self) -> bytes:
117		-	return Packet.MAGIC + b"print\n"
118		-
119		-	def is_valid_fragment(self, fragment: bytes) -> bool:
120		-	return fragment.startswith(self._header())
	112	+	return self.MAGIC + b"print\n"
121	113
122		-	def add_fragment(self, fragment: bytes):
123		-	self._fragments.extend(fragment.removeprefix(self._header()))
	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()
124	120
125		-	def build(self) -> str:
126		-	return self._fragments.decode()
127	121
128		-
129		-	def send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str:
	122	+	def send(sock: socket.socket, args: argparse.Namespace, cmd: str) -> str:
130	123		"""
131	124		Send a single RCON (Remote Console) command to the server and return the response.
132	125
			@@ -134,54 +127,61 @@ def send(sock: socket.socket, request: Request, host: str, port: int, cmd: str)
134	127		It waits for the server's response, collects it in fragments if necessary, and returns the complete response.
135	128
136	129		Args:
137		-	sock (socket.socket): The UDP socket object used to send and receive data.
138		-	request (Request): An instance of the Request class used to encode the command.
139		-	host (str): The hostname or IP address of the server.
140		-	port (int): The port number on which the server is listening.
141		-	cmd (str): The RCON command to be sent to the server.
142		-
	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.
143	133		Returns:
144		-	str: The complete response from the server after processing all received fragments.
145		-
	134	+	str: The complete response from the server after sending the command.
146	135		Raises:
147		-	socket.timeout: If the socket times out while waiting for a response from the server.
	136	+	ValueError: If any of the response fragments received from the server are invalid.
148	137		"""
149		-	sock.sendto(request.encode(cmd), (host, port))
150		-
151		-	response_builder = ResponseBuilder()
152		-	while True:
153		-	try:
154		-	data, _ = sock.recvfrom(4096)
155		-	except socket.timeout:
156		-	break
	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))
157	141
158		-	if response_builder.is_valid_fragment(data):
159		-	response_builder.add_fragment(data)
	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)")
160	153
161		-	return response_builder.build()
	154	+	return "".join(fragments)
162	155
163	156
164	157		def parse_args() -> argparse.Namespace:
165	158		parser = argparse.ArgumentParser()
166		-	parser.add_argument("--host", default="localhost")
167		-	parser.add_argument("--port", type=int, default=27960)
168		-	parser.add_argument("--password", default="secret")
169		-	parser.add_argument("--timeout", type=float, default=0.2)
170		-	parser.add_argument("--interactive", action="store_true")
171		-	parser.add_argument("cmd", nargs="?", default="status")
	159	+	parser.add_argument("--host", "-H", default="localhost")
	160	+	parser.add_argument("--port", "-P", type=int, default=28960)
	161	+	parser.add_argument("--password", "-p", default="secret")
	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	172		args = parser.parse_args()
173	173		return args
174	174
175	175
176		-	def main():
	176	+	def main(args: argparse.Namespace):
177	177		"""
178	178		Main function to handle command-line arguments and interact with a remote server using RCON protocol.
179	179		This function parses command-line arguments, creates a UDP socket, and sends requests to a remote server.
180	180		It supports both interactive and non-interactive modes.
181		-	In non-interactive mode, it sends a single command to the server and prints the response.
182	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	183		Args:
184		-	None
	184	+	args (argparse.Namespace): The command-line arguments containing host, port, and password.
185	185		Returns:
186	186		None
187	187		"""
			@@ -189,25 +189,26 @@ def main():
189	189		def cleaned_response(resp: str) -> str:
190	190		return re.sub(r"\^[0-9]", "", resp)
191	191
192		-	args = parse_args()
193		-
194	192		with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
195	193		sock.settimeout(args.timeout)
196	194
197		-	request = Request(args.password)
	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
198	200
199		-	if not args.interactive:
200		-	if resp := send(sock, request, args.host, args.port, args.cmd):
201		-	print(cleaned_response(resp))
202		-	return
	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))
203	207
204		-	while cmd := input("cmd: ").strip():
205		-	if cmd == "Q":
206		-	break
207	208
208		-	if resp := send(sock, request, args.host, args.port, cmd):
209		-	print(cleaned_response(resp))
	209	+	if __name__ == "__main__":
	210	+	args = parse_args()
210	211
	212	+	logging.basicConfig(level=getattr(logging, args.loglevel.upper()))
211	213
212		-	if __name__ == "__main__":
213		-	main()
	214	+	main(args)

onyx-and-iris zrewidował ten Gist 1738730389. Przejdź do rewizji

1 file changed, 29 insertions

rcon.py

			@@ -22,6 +22,35 @@
22	22		# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23	23		# SOFTWARE.
24	24
	25	+	"""
	26	+	rcon.py
	27	+
	28	+	This script provides functionality to interact with a remote server using the RCON (Remote Console) protocol.
	29	+
	30	+	It includes classes to encode and decode RCON packets, send commands to the server, and handle responses.
	31	+	The script also provides a command-line interface for sending RCON commands to a server.
	32	+
	33	+	Classes:
	34	+	Packet: Base class for RCON packets.
	35	+	Request: Class to encode a request packet for RCON communication.
	36	+	ResponseBuilder: Class to build and decode a response packet from multiple fragments.
	37	+
	38	+	Functions:
	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	+
	46	+	Usage:
	47	+	To send a single command to the server:
	48	+	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> <command>
	49	+
	50	+	To use interactive mode:
	51	+	python rcon.py --host <server_host> --port <server_port> --password <rcon_password> --interactive
	52	+	"""
	53	+
25	54		import argparse
26	55		import re
27	56		import socket

onyx-and-iris zrewidował ten Gist 1736893124. Przejdź do rewizji

1 file changed, 6 insertions, 2 deletions

rcon.py

			@@ -23,6 +23,7 @@
23	23		# SOFTWARE.
24	24
25	25		import argparse
	26	+	import re
26	27		import socket
27	28
28	29
			@@ -156,6 +157,9 @@ def main():
156	157		None
157	158		"""
158	159
	160	+	def cleaned_response(resp: str) -> str:
	161	+	return re.sub(r"\^[0-9]", "", resp)
	162	+
159	163		args = parse_args()
160	164
161	165		with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
			@@ -165,7 +169,7 @@ def main():
165	169
166	170		if not args.interactive:
167	171		if resp := send(sock, request, args.host, args.port, args.cmd):
168		-	print(resp)
	172	+	print(cleaned_response(resp))
169	173		return
170	174
171	175		while cmd := input("cmd: ").strip():
			@@ -173,7 +177,7 @@ def main():
173	177		break
174	178
175	179		if resp := send(sock, request, args.host, args.port, cmd):
176		-	print(resp)
	180	+	print(cleaned_response(resp))
177	181
178	182
179	183		if __name__ == "__main__":

onyx-and-iris zrewidował ten Gist 1736632704. Przejdź do rewizji

1 file changed, 61 insertions, 11 deletions

rcon.py

			@@ -31,7 +31,23 @@ class Packet:
31	31
32	32
33	33		class Request(Packet):
34		-	"""Class used to encode a request packet."""
	34	+	"""
	35	+	Class used to encode a request packet for remote console (RCON) communication.
	36	+
	37	+	Attributes:
	38	+	password (str): The password used for authentication with the RCON server.
	39	+
	40	+	Methods:
	41	+	_header() -> bytes:
	42	+	Generates the header for the RCON packet.
	43	+
	44	+	encode(cmd: str) -> bytes:
	45	+	Encodes the command into a byte string suitable for sending to the RCON server.
	46	+	Args:
	47	+	cmd (str): The command to be sent to the RCON server.
	48	+	Returns:
	49	+	bytes: The encoded command as a byte string.
	50	+	"""
35	51
36	52		def __init__(self, password: str):
37	53		self.password = password
			@@ -44,8 +60,24 @@ class Request(Packet):
44	60
45	61
46	62		class ResponseBuilder(Packet):
47		-	"""Class used to build and decode a response packet.
48		-	The response may be built from multiple fragments.
	63	+	"""
	64	+	Class used to build and decode a response packet from multiple fragments.
	65	+
	66	+	Attributes:
	67	+	_fragments (bytearray): A bytearray to store the concatenated fragments.
	68	+
	69	+	Methods:
	70	+	_header() -> bytes:
	71	+	Returns the header bytes that each fragment should start with.
	72	+
	73	+	is_valid_fragment(fragment: bytes) -> bool:
	74	+	Checks if the given fragment starts with the expected header.
	75	+
	76	+	add_fragment(fragment: bytes):
	77	+	Adds a fragment to the internal storage after removing the header.
	78	+
	79	+	build() -> str:
	80	+	Decodes and returns the concatenated fragments as a string.
49	81		"""
50	82
51	83		def __init__(self):
			@@ -65,17 +97,24 @@ class ResponseBuilder(Packet):
65	97
66	98
67	99		def send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str:
68		-	"""Send a single rcon command to the server and return the response.
	100	+	"""
	101	+	Send a single RCON (Remote Console) command to the server and return the response.
	102	+
	103	+	This function sends a command to a game server using the RCON protocol over a UDP socket.
	104	+	It waits for the server's response, collects it in fragments if necessary, and returns the complete response.
69	105
70	106		Args:
71		-	sock (socket.socket): UDP socket object
72		-	request (Request): reference to a Request object
73		-	host (str): hostname or IP address
74		-	port (int): port number
75		-	cmd (str): the rcon command to send
	107	+	sock (socket.socket): The UDP socket object used to send and receive data.
	108	+	request (Request): An instance of the Request class used to encode the command.
	109	+	host (str): The hostname or IP address of the server.
	110	+	port (int): The port number on which the server is listening.
	111	+	cmd (str): The RCON command to be sent to the server.
76	112
77	113		Returns:
78		-	str: the response from the server
	114	+	str: The complete response from the server after processing all received fragments.
	115	+
	116	+	Raises:
	117	+	socket.timeout: If the socket times out while waiting for a response from the server.
79	118		"""
80	119		sock.sendto(request.encode(cmd), (host, port))
81	120
			@@ -105,7 +144,18 @@ def parse_args() -> argparse.Namespace:
105	144
106	145
107	146		def main():
108		-	"""Fire command in one-shot mode or interactive mode."""
	147	+	"""
	148	+	Main function to handle command-line arguments and interact with a remote server using RCON protocol.
	149	+	This function parses command-line arguments, creates a UDP socket, and sends requests to a remote server.
	150	+	It supports both interactive and non-interactive modes.
	151	+	In non-interactive mode, it sends a single command to the server and prints the response.
	152	+	In interactive mode, it continuously prompts the user for commands until the user quits by entering 'Q'.
	153	+	Args:
	154	+	None
	155	+	Returns:
	156	+	None
	157	+	"""
	158	+
109	159		args = parse_args()
110	160
111	161		with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:

onyx-and-iris zrewidował ten Gist 1736632279. Przejdź do rewizji

1 file changed, 4 insertions, 4 deletions

rcon.py

			@@ -108,13 +108,13 @@ def main():
108	108		"""Fire command in one-shot mode or interactive mode."""
109	109		args = parse_args()
110	110
111		-	with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
112		-	s.settimeout(args.timeout)
	111	+	with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
	112	+	sock.settimeout(args.timeout)
113	113
114	114		request = Request(args.password)
115	115
116	116		if not args.interactive:
117		-	if resp := send(s, request, args.host, args.port, args.cmd):
	117	+	if resp := send(sock, request, args.host, args.port, args.cmd):
118	118		print(resp)
119	119		return
120	120
			@@ -122,7 +122,7 @@ def main():
122	122		if cmd == "Q":
123	123		break
124	124
125		-	if resp := send(s, request, args.host, args.port, cmd):
	125	+	if resp := send(sock, request, args.host, args.port, cmd):
126	126		print(resp)
127	127
128	128

onyx_online zrewidował ten Gist 1736632007. Przejdź do rewizji

1 file changed, 130 insertions

rcon.py(stworzono plik)

		@@ -0,0 +1,130 @@
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	+	import argparse
26	+	import socket
27	+
28	+
29	+	class Packet:
30	+	MAGIC = bytearray([0xFF, 0xFF, 0xFF, 0xFF])
31	+
32	+
33	+	class Request(Packet):
34	+	"""Class used to encode a request packet."""
35	+
36	+	def __init__(self, password: str):
37	+	self.password = password
38	+
39	+	def _header(self) -> bytes:
40	+	return Packet.MAGIC + b"rcon"
41	+
42	+	def encode(self, cmd: str) -> bytes:
43	+	return self._header() + f" {self.password} {cmd}".encode()
44	+
45	+
46	+	class ResponseBuilder(Packet):
47	+	"""Class used to build and decode a response packet.
48	+	The response may be built from multiple fragments.
49	+	"""
50	+
51	+	def __init__(self):
52	+	self._fragments = bytearray()
53	+
54	+	def _header(self) -> bytes:
55	+	return Packet.MAGIC + b"print\n"
56	+
57	+	def is_valid_fragment(self, fragment: bytes) -> bool:
58	+	return fragment.startswith(self._header())
59	+
60	+	def add_fragment(self, fragment: bytes):
61	+	self._fragments.extend(fragment.removeprefix(self._header()))
62	+
63	+	def build(self) -> str:
64	+	return self._fragments.decode()
65	+
66	+
67	+	def send(sock: socket.socket, request: Request, host: str, port: int, cmd: str) -> str:
68	+	"""Send a single rcon command to the server and return the response.
69	+
70	+	Args:
71	+	sock (socket.socket): UDP socket object
72	+	request (Request): reference to a Request object
73	+	host (str): hostname or IP address
74	+	port (int): port number
75	+	cmd (str): the rcon command to send
76	+
77	+	Returns:
78	+	str: the response from the server
79	+	"""
80	+	sock.sendto(request.encode(cmd), (host, port))
81	+
82	+	response_builder = ResponseBuilder()
83	+	while True:
84	+	try:
85	+	data, _ = sock.recvfrom(4096)
86	+	except socket.timeout:
87	+	break
88	+
89	+	if response_builder.is_valid_fragment(data):
90	+	response_builder.add_fragment(data)
91	+
92	+	return response_builder.build()
93	+
94	+
95	+	def parse_args() -> argparse.Namespace:
96	+	parser = argparse.ArgumentParser()
97	+	parser.add_argument("--host", default="localhost")
98	+	parser.add_argument("--port", type=int, default=27960)
99	+	parser.add_argument("--password", default="secret")
100	+	parser.add_argument("--timeout", type=float, default=0.2)
101	+	parser.add_argument("--interactive", action="store_true")
102	+	parser.add_argument("cmd", nargs="?", default="status")
103	+	args = parser.parse_args()
104	+	return args
105	+
106	+
107	+	def main():
108	+	"""Fire command in one-shot mode or interactive mode."""
109	+	args = parse_args()
110	+
111	+	with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
112	+	s.settimeout(args.timeout)
113	+
114	+	request = Request(args.password)
115	+
116	+	if not args.interactive:
117	+	if resp := send(s, request, args.host, args.port, args.cmd):
118	+	print(resp)
119	+	return
120	+
121	+	while cmd := input("cmd: ").strip():
122	+	if cmd == "Q":
123	+	break
124	+
125	+	if resp := send(s, request, args.host, args.port, cmd):
126	+	print(resp)
127	+
128	+
129	+	if __name__ == "__main__":
130	+	main()

Nowsze Starsze