1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
"""Low-level protocol stuff."""
from __future__ import annotations
from asyncio import StreamReader
from enum import Enum
from functools import partial
from logging import getLogger
from random import randint
from typing import IO, NamedTuple
from rcon.exceptions import EmptyResponse, UnexpectedTerminator
__all__ = ["LittleEndianSignedInt32", "Type", "Packet", "random_request_id"]
LOGGER = getLogger(__file__)
TERMINATOR = b"\x00\x00"
class LittleEndianSignedInt32(int):
"""A little-endian, signed int32."""
MIN = -2_147_483_648
MAX = 2_147_483_647
def __init__(self, *_):
"""Check the boundaries."""
super().__init__()
if not self.MIN <= self <= self.MAX:
raise ValueError("Signed int32 out of bounds:", int(self))
def __bytes__(self):
"""Return the integer as signed little endian."""
return self.to_bytes(4, "little", signed=True)
@classmethod
async def aread(cls, reader: StreamReader) -> LittleEndianSignedInt32:
"""Read the integer from an asynchronous file-like object."""
return cls.from_bytes(await reader.read(4), "little", signed=True)
@classmethod
def read(cls, file: IO) -> LittleEndianSignedInt32:
"""Read the integer from a file-like object."""
return cls.from_bytes(file.read(4), "little", signed=True)
class Type(LittleEndianSignedInt32, Enum):
"""RCON packet types."""
SERVERDATA_AUTH = LittleEndianSignedInt32(3)
SERVERDATA_AUTH_RESPONSE = LittleEndianSignedInt32(2)
SERVERDATA_EXECCOMMAND = LittleEndianSignedInt32(2)
SERVERDATA_RESPONSE_VALUE = LittleEndianSignedInt32(0)
def __int__(self):
"""Return the actual integer value."""
return int(self.value)
def __bytes__(self):
"""Return the integer value as little endian."""
return bytes(self.value)
@classmethod
async def aread(cls, reader: StreamReader, *, prefix: str = "") -> Type:
"""Read the type from an asynchronous file-like object."""
LOGGER.debug("%sReading type asynchronously.", prefix)
value = await LittleEndianSignedInt32.aread(reader)
LOGGER.debug("%s => value: %i", prefix, value)
return cls(value)
@classmethod
def read(cls, file: IO, *, prefix: str = "") -> Type:
"""Read the type from a file-like object."""
LOGGER.debug("%sReading type.", prefix)
value = LittleEndianSignedInt32.read(file)
LOGGER.debug("%s => value: %i", prefix, value)
return cls(value)
class Packet(NamedTuple):
"""An RCON packet."""
id: LittleEndianSignedInt32
type: Type
payload: bytes
terminator: bytes = TERMINATOR
def __add__(self, other: Packet | None) -> Packet:
if other is None:
return self
if other.id != self.id:
raise ValueError("Can only add packages with same id.")
if other.type != self.type:
raise ValueError("Can only add packages of same type.")
if other.terminator != self.terminator:
raise ValueError("Can only add packages with same terminator.")
return Packet(self.id, self.type, self.payload + other.payload, self.terminator)
def __radd__(self, other: Packet | None) -> Packet:
if other is None:
return self
return other.__add__(self)
def __bytes__(self):
"""Return the packet as bytes with prepended length."""
payload = bytes(self.id)
payload += bytes(self.type)
payload += self.payload
payload += self.terminator
size = bytes(LittleEndianSignedInt32(len(payload)))
return size + payload
@classmethod
async def aread(
cls, reader: StreamReader, raise_unexpected_terminator: bool = False
) -> Packet:
"""Read a packet from an asynchronous file-like object."""
LOGGER.debug("Reading packet asynchronously.")
size = await LittleEndianSignedInt32.aread(reader)
LOGGER.debug(" => size: %i", size)
if not size:
raise EmptyResponse()
id_ = await LittleEndianSignedInt32.aread(reader)
LOGGER.debug(" => id: %i", id_)
type_ = await Type.aread(reader, prefix=" ")
LOGGER.debug(" => type: %i", type_)
payload = await reader.read(size - 10)
LOGGER.debug(" => payload: %s", payload)
terminator = await reader.read(2)
LOGGER.debug(" => terminator: %s", terminator)
if terminator != TERMINATOR:
if raise_unexpected_terminator:
raise UnexpectedTerminator(terminator)
LOGGER.warning("Unexpected terminator: %s", terminator)
return cls(id_, type_, payload, terminator)
@classmethod
def read(cls, file: IO, raise_unexpected_terminator: bool = False) -> Packet:
"""Read a packet from a file-like object."""
LOGGER.debug("Reading packet.")
size = LittleEndianSignedInt32.read(file)
LOGGER.debug(" => size: %i", size)
if not size:
raise EmptyResponse()
id_ = LittleEndianSignedInt32.read(file)
LOGGER.debug(" => id: %i", id_)
type_ = Type.read(file, prefix=" ")
LOGGER.debug(" => type: %i", type_)
payload = file.read(size - 10)
LOGGER.debug(" => payload: %s", payload)
terminator = file.read(2)
LOGGER.debug(" => terminator: %s", terminator)
if terminator != TERMINATOR:
if raise_unexpected_terminator:
raise UnexpectedTerminator(terminator)
LOGGER.warning("Unexpected terminator: %s", terminator)
return cls(id_, type_, payload, terminator)
@classmethod
def make_command(cls, *args: str, encoding: str = "utf-8") -> Packet:
"""Create a command packet."""
return cls(
random_request_id(),
Type.SERVERDATA_EXECCOMMAND,
b" ".join(map(partial(str.encode, encoding=encoding), args)),
)
@classmethod
def make_empty_response(cls) -> Packet:
"""Create an empty response packet."""
return cls(random_request_id(), Type.SERVERDATA_RESPONSE_VALUE, b"")
@classmethod
def make_login(cls, passwd: str, *, encoding: str = "utf-8") -> Packet:
"""Create a login packet."""
return cls(random_request_id(), Type.SERVERDATA_AUTH, passwd.encode(encoding))
def random_request_id() -> LittleEndianSignedInt32:
"""Generate a random request ID."""
return LittleEndianSignedInt32(randint(0, LittleEndianSignedInt32.MAX))
|
