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
|
"""An interactive console."""
from getpass import getpass
from typing import Type
from rcon.client import BaseClient
from rcon.config import Config
from rcon.exceptions import EmptyResponse, SessionTimeout, WrongPassword
__all__ = ["PROMPT", "rconcmd"]
EXIT_COMMANDS = {"exit", "quit"}
MSG_LOGIN_ABORTED = "\nLogin aborted. Bye."
MSG_EXIT = "\nBye."
MSG_SERVER_GONE = "Server has gone away."
MSG_SESSION_TIMEOUT = "Session timed out. Please login again."
PROMPT = "RCON {host}:{port}> "
VALID_PORTS = range(0, 65536)
def read_host() -> str:
"""Read the host."""
while True:
try:
return input("Host: ")
except KeyboardInterrupt:
print()
continue
def read_port() -> int:
"""Read the port."""
while True:
try:
port = input("Port: ")
except KeyboardInterrupt:
print()
continue
try:
port = int(port)
except ValueError:
print(f"Invalid integer: {port}")
continue
if port in VALID_PORTS:
return port
print(f"Invalid port: {port}")
def read_passwd() -> str:
"""Read the password."""
while True:
try:
return getpass("Password: ")
except KeyboardInterrupt:
print()
def get_config(host: str, port: int, passwd: str) -> Config:
"""Read the necessary arguments."""
if host is None:
host = read_host()
if port is None:
port = read_port()
if passwd is None:
passwd = read_passwd()
return Config(host, port, passwd)
def login(client: BaseClient, passwd: str) -> str:
"""Perform a login."""
while True:
try:
client.login(passwd)
except WrongPassword:
print("Wrong password.")
passwd = read_passwd()
continue
return passwd
def process_input(client: BaseClient, passwd: str, prompt: str) -> bool:
"""Process the CLI input."""
try:
command = input(prompt)
except KeyboardInterrupt:
print()
return True
except EOFError:
print(MSG_EXIT)
return False
try:
command, *args = command.split()
except ValueError:
return True
if command in EXIT_COMMANDS:
return False
try:
result = client.run(command, *args)
except EmptyResponse:
print(MSG_SERVER_GONE)
return False
except SessionTimeout:
print(MSG_SESSION_TIMEOUT)
try:
login(client, passwd)
except EOFError:
print(MSG_LOGIN_ABORTED)
return False
return True
if result:
print(result)
return True
def rconcmd(
client_cls: Type[BaseClient],
host: str,
port: int,
passwd: str,
*,
timeout: float | None = None,
prompt: str = PROMPT,
):
"""Initialize the console."""
try:
host, port, passwd = get_config(host, port, passwd)
except EOFError:
print(MSG_EXIT)
return
prompt = prompt.format(host=host, port=port)
with client_cls(host, port, timeout=timeout) as client:
try:
passwd = login(client, passwd)
except EOFError:
print(MSG_LOGIN_ABORTED)
return
while True:
if not process_input(client, passwd, prompt):
break
|
