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
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
|
"""GTK based GUI."""
from argparse import ArgumentParser, Namespace
from json import dump, load
from logging import DEBUG, INFO, basicConfig, getLogger
from os import getenv, name
from pathlib import Path
from socket import gaierror, timeout
from typing import Iterable, NamedTuple, Type
from gi import require_version
require_version("Gtk", "3.0")
from gi.repository import Gtk
from rcon import battleye, source
from rcon.client import BaseClient
from rcon.config import LOG_FORMAT
from rcon.exceptions import SessionTimeout, WrongPassword
__all__ = ["main"]
if name == "posix":
CACHE_DIR = Path.home().joinpath(".cache")
elif name == "nt":
CACHE_DIR = Path(getenv("TEMP") or getenv("TMP"))
else:
raise NotImplementedError("Unsupported operating system.")
CACHE_FILE = CACHE_DIR.joinpath("rcongui.json")
LOGGER = getLogger("rcongui")
def get_args() -> Namespace:
"""Parse and return the command line arguments."""
parser = ArgumentParser(description="A minimalistic, GTK-based RCON GUI.")
parser.add_argument(
"-B",
"--battleye",
action="store_true",
help="use BattlEye RCon instead of Source RCON",
)
parser.add_argument(
"-d", "--debug", action="store_true", help="print additional debug information"
)
parser.add_argument(
"-t",
"--timeout",
type=float,
metavar="seconds",
help="connection timeout in seconds",
)
return parser.parse_args()
class RCONParams(NamedTuple):
"""Represent the RCON parameters."""
host: str
port: int
passwd: str
command: Iterable[str]
class GUI(Gtk.Window):
"""A GTK based GUI for RCON."""
def __init__(self, args: Namespace):
"""Initialize the GUI."""
super().__init__(title="RCON GUI")
self.args = args
self.set_position(Gtk.WindowPosition.CENTER)
self.grid = Gtk.Grid()
self.add(self.grid)
self.host = Gtk.Entry()
self.host.set_placeholder_text("Host")
self.grid.attach(self.host, 0, 0, 1, 1)
self.port = Gtk.SpinButton.new_with_range(0, 65535, 1)
self.port.set_placeholder_text("Port")
self.grid.attach(self.port, 1, 0, 1, 1)
self.passwd = Gtk.Entry()
self.passwd.set_placeholder_text("Password")
self.passwd.set_visibility(False)
self.grid.attach(self.passwd, 2, 0, 1, 1)
self.command = Gtk.Entry()
self.command.set_placeholder_text("Command")
self.grid.attach(self.command, 0, 1, 2, 1)
self.button = Gtk.Button(label="Run")
self.button.connect("clicked", self.on_button_clicked)
self.grid.attach(self.button, 2, 1, 1, 1)
self.result = Gtk.TextView()
self.result.set_wrap_mode(Gtk.WrapMode.WORD)
self.result.set_property("editable", False)
self.grid.attach(self.result, 0, 2, 2, 1)
self.savepw = Gtk.CheckButton(label="Save password")
self.grid.attach(self.savepw, 2, 2, 1, 1)
self.load_gui_settings()
@property
def client_cls(self) -> Type[BaseClient]:
"""Return the client class."""
return battleye.Client if self.args.battleye else source.Client
@property
def result_text(self) -> str:
"""Return the result text."""
if (buf := self.result.get_buffer()) is not None:
return buf.get_text(
buf.get_iter_at_line(0),
buf.get_iter_at_line(buf.get_line_count()),
True,
)
return ""
@result_text.setter
def result_text(self, text: str):
"""Set the result text."""
if (buf := self.result.get_buffer()) is not None:
buf.set_text(text)
@property
def gui_settings(self) -> dict:
"""Return the GUI settings as a dict."""
json = {
"host": self.host.get_text(),
"port": self.port.get_value_as_int(),
"command": self.command.get_text(),
"result": self.result_text,
"savepw": (savepw := self.savepw.get_active()),
}
if savepw:
json["passwd"] = self.passwd.get_text()
return json
@gui_settings.setter
def gui_settings(self, json: dict):
"""Set the GUI settings."""
self.host.set_text(json.get("host", ""))
self.port.set_value(json.get("port", 0))
self.passwd.set_text(json.get("passwd", ""))
self.command.set_text(json.get("command", ""))
self.result_text = json.get("result", "")
self.savepw.set_active(json.get("savepw", False))
def load_gui_settings(self) -> None:
"""Load the GUI settings from the cache file."""
try:
with CACHE_FILE.open("rb") as cache:
self.gui_settings = load(cache)
except FileNotFoundError:
LOGGER.warning("Cache file not found: %s", CACHE_FILE)
except PermissionError:
LOGGER.error("Insufficient permissions to read: %s", CACHE_FILE)
except ValueError:
LOGGER.error("Cache file contains garbage: %s", CACHE_FILE)
def save_gui_settings(self):
"""Save the GUI settings to the cache file."""
try:
with CACHE_FILE.open("w", encoding="utf-8") as cache:
dump(self.gui_settings, cache, indent=2)
except PermissionError:
LOGGER.error("Insufficient permissions to read: %s", CACHE_FILE)
def show_error(self, message: str):
"""Show an error message."""
message_dialog = Gtk.MessageDialog(
transient_for=self,
message_type=Gtk.MessageType.ERROR,
buttons=Gtk.ButtonsType.OK,
text=message,
)
message_dialog.run()
message_dialog.destroy()
def run_rcon(self) -> str:
"""Return the current RCON settings."""
with self.client_cls(
self.host.get_text().strip(),
self.port.get_value_as_int(),
timeout=self.args.timeout,
passwd=self.passwd.get_text(),
) as client:
return client.run(*self.command.get_text().strip().split())
def on_button_clicked(self, _):
"""Run the client."""
try:
result = self.run_rcon()
except ValueError as error:
self.show_error(str(error))
except gaierror as error:
self.show_error(error.strerror)
except ConnectionRefusedError:
self.show_error("Connection refused.")
except (TimeoutError, timeout):
self.show_error("Connection timed out.")
except WrongPassword:
self.show_error("Wrong password.")
except SessionTimeout:
self.show_error("Session timed out.")
else:
self.result_text = result
def terminate(self, *args, **kwargs):
"""Save the settings and terminates the application."""
self.save_gui_settings()
Gtk.main_quit(*args, **kwargs)
def main() -> None:
"""Start the GUI."""
args = get_args()
basicConfig(format=LOG_FORMAT, level=DEBUG if args.debug else INFO)
win = GUI(args)
win.connect("destroy", win.terminate)
win.show_all()
Gtk.main()
|
