Don't fail dist or docs builds on deprecations
[mudpy.git] / mudpy / telnet.py
1 """Telnet functions and constants for the mudpy engine."""
2
3 # Copyright (c) 2004-2020 mudpy authors. Permission to use, copy,
4 # modify, and distribute this software is granted under terms
5 # provided in the LICENSE file distributed with this software.
6
7 import mudpy
8
9 # telnet options (from bsd's arpa/telnet.h since telnetlib's are ambiguous)
10 TELOPT_BINARY = 0  # transmit 8-bit data by the receiver (rfc 856)
11 TELOPT_ECHO = 1  # echo received data back to the sender (rfc 857)
12 TELOPT_SGA = 3  # suppress transmission of the go ahead character (rfc 858)
13 TELOPT_TTYPE = 24  # exchange terminal type information (rfc 1091)
14 TELOPT_EOR = 25  # transmit end-of-record after transmitting data (rfc 885)
15 TELOPT_NAWS = 31  # negotiate about window size (rfc 1073)
16 TELOPT_LINEMODE = 34  # cooked line-by-line input mode (rfc 1184)
17 option_names = {
18     TELOPT_BINARY: "8-bit binary",
19     TELOPT_ECHO: "echo",
20     TELOPT_SGA: "suppress go-ahead",
21     TELOPT_TTYPE: "terminal type",
22     TELOPT_EOR: "end of record",
23     TELOPT_NAWS: "negotiate about window size",
24     TELOPT_LINEMODE: "line mode",
25 }
26
27 supported = (
28     TELOPT_BINARY,
29     TELOPT_ECHO,
30     TELOPT_SGA,
31     TELOPT_TTYPE,
32     TELOPT_EOR,
33     TELOPT_NAWS,
34     TELOPT_LINEMODE
35 )
36
37 # telnet commands
38 EOR = 239  # end-of-record signal (rfc 885)
39 SE = 240  # end of subnegotiation parameters (rfc 854)
40 GA = 249  # go ahead signal (rfc 854)
41 SB = 250  # what follows is subnegotiation of the indicated option (rfc 854)
42 WILL = 251  # desire or confirmation of performing an option (rfc 854)
43 WONT = 252  # refusal or confirmation of performing an option (rfc 854)
44 DO = 253  # request or confirm performing an option (rfc 854)
45 DONT = 254  # demand or confirm no longer performing an option (rfc 854)
46 IAC = 255  # interpret as command escape character (rfc 854)
47 command_names = {
48     EOR: "end of record",
49     SE: "subnegotiation end",
50     GA: "go ahead",
51     SB: "subnegotiation begin",
52     WILL: "will",
53     WONT: "won't",
54     DO: "do",
55     DONT: "don't",
56     IAC: "interpret as command",
57 }
58
59 # RFC 1143 option negotiation states
60 NO = 0  # option is disabled
61 YES = 1  # option is enabled
62 WANTNO = 2  # demanded disabling option
63 WANTYES = 3  # requested enabling option
64 WANTNO_OPPOSITE = 4  # demanded disabling option but queued an enable after it
65 WANTYES_OPPOSITE = 5  # requested enabling option but queued a disable after it
66 state_names = {
67     NO: "disabled",
68     YES: "enabled",
69     WANTNO: "demand disabling",
70     WANTYES: "request enabling",
71     WANTNO_OPPOSITE: "want no queued opposite",
72     WANTYES_OPPOSITE: "want yes queued opposite",
73 }
74
75 # RFC 1143 option negotiation parties
76 HIM = 0
77 US = 1
78 party_names = {
79     HIM: "him",
80     US: "us",
81 }
82
83 # RFC 1091 commands
84 IS = 0
85 SEND = 1
86 ttype_command_names = {
87     IS: "is",
88     SEND: "send",
89 }
90
91
92 def log(message, user):
93     """Log debugging info for Telnet client/server interactions."""
94     if user.account:
95         client = user.account.get("name", user)
96     else:
97         client = user
98     mudpy.misc.log('[telnet] %s %s.' % (message, client), 0)
99
100
101 def telnet_proto(*arguments):
102     """Return a concatenated series of Telnet protocol commands."""
103     return bytes((arguments))
104
105
106 def translate_action(*command):
107     """Convert a Telnet command sequence into text suitable for logging."""
108     try:
109         command_name = command_names[command[0]]
110     except KeyError:
111         # This should never happen since we filter unknown commands from
112         # the input queue, but added here for completeness since logging
113         # should never crash the process
114         command_name = str(command[0])
115     try:
116         option_name = option_names[command[1]]
117     except KeyError:
118         # This can happen for any of the myriad of Telnet options missing
119         # from the option_names dict
120         option_name = str(command[1])
121     return "%s %s" % (command_name, option_name)
122
123
124 def send_command(user, *command):
125     """Sends a Telnet command string to the specified user's socket."""
126     user.send(telnet_proto(IAC, *command), raw=True)
127     log('Sent "%s" to' % translate_action(*command), user)
128
129
130 def is_enabled(user, telopt, party, state=YES):
131     """Indicates whether a specified Telnet option is enabled."""
132     if (telopt, party) in user.telopts and user.telopts[
133        (telopt, party)
134        ] is state:
135         return True
136     else:
137         return False
138
139
140 def enable(user, telopt, party):
141     """Negotiates enabling a Telnet option for the indicated user's socket."""
142     if party is HIM:
143         txpos = DO
144     else:
145         txpos = WILL
146     if not (telopt, party) in user.telopts or user.telopts[
147        (telopt, party)
148        ] is NO:
149         user.telopts[(telopt, party)] = WANTYES
150         send_command(user, txpos, telopt)
151     elif user.telopts[(telopt, party)] is WANTNO:
152         user.telopts[(telopt, party)] = WANTNO_OPPOSITE
153     elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
154         user.telopts[(telopt, party)] = WANTYES
155
156
157 def disable(user, telopt, party):
158     """Negotiates disabling a Telnet option for the user's socket."""
159     if party is HIM:
160         txneg = DONT
161     else:
162         txneg = WONT
163     if not (telopt, party) in user.telopts:
164         user.telopts[(telopt, party)] = NO
165     elif user.telopts[(telopt, party)] is YES:
166         user.telopts[(telopt, party)] = WANTNO
167         send_command(user, txneg, telopt)
168     elif user.telopts[(telopt, party)] is WANTYES:
169         user.telopts[(telopt, party)] = WANTYES_OPPOSITE
170     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
171         user.telopts[(telopt, party)] = WANTNO
172
173
174 def request_ttype(user):
175     """Clear and request the terminal type."""
176
177     # only actually request if the corresponding telopt is enabled
178     if is_enabled(user, TELOPT_TTYPE, HIM):
179         # set to the empty string to indicate it's been requested
180         user.ttype = ""
181         user.send(telnet_proto(IAC, SB, TELOPT_TTYPE, SEND, IAC, SE), raw=True)
182         log('Sent terminal type request to', user)
183
184
185 def negotiate_telnet_options(user):
186     """Reply to and remove telnet negotiation options from partial_input."""
187
188     # make a local copy to play with
189     text = user.partial_input
190
191     # start at the beginning of the input
192     position = 0
193
194     # as long as we haven't checked it all
195     len_text = len(text)
196     while position < len_text:
197
198         # jump to the first IAC you find
199         position = text.find(telnet_proto(IAC), position)
200
201         # if there wasn't an IAC in the input or it's at the end, we're done
202         if position < 0 or position >= len_text - 1:
203             break
204
205         # the byte following the IAC is our command
206         command = text[position+1]
207
208         # replace a double (literal) IAC if there's a CR+NUL or CR+LF later
209         if command is IAC:
210             if (
211                     text.find(b"\r\0", position) > 0 or
212                     text.find(b"\r\n", position) > 0):
213                 position += 1
214                 text = text[:position] + text[position + 1:]
215                 log('Escaped IAC from', user)
216             else:
217                 position += 2
218
219         # implement an RFC 1143 option negotiation queue here
220         elif len_text > position + 2 and WILL <= command <= DONT:
221             telopt = text[position+2]
222             log('Received "%s" from' % translate_action(command, telopt), user)
223             if telopt in supported:
224                 if command <= WONT:
225                     party = HIM
226                     rxpos = WILL
227                     txpos = DO
228                     txneg = DONT
229                 else:
230                     party = US
231                     rxpos = DO
232                     txpos = WILL
233                     txneg = WONT
234                 if (telopt, party) not in user.telopts:
235                     user.telopts[(telopt, party)] = NO
236                 if command is rxpos:
237                     if user.telopts[(telopt, party)] is NO:
238                         user.telopts[(telopt, party)] = YES
239                         send_command(user, txpos, telopt)
240                     elif user.telopts[(telopt, party)] is WANTNO:
241                         user.telopts[(telopt, party)] = NO
242                     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
243                         user.telopts[(telopt, party)] = YES
244                     elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
245                         user.telopts[(telopt, party)] = WANTNO
246                         send_command(user, txneg, telopt)
247                     else:
248                         user.telopts[(telopt, party)] = YES
249                 else:
250                     if user.telopts[(telopt, party)] is YES:
251                         user.telopts[(telopt, party)] = NO
252                         send_command(user, txneg, telopt)
253                     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
254                         user.telopts[(telopt, party)] = WANTYES
255                         send_command(user, txpos, telopt)
256                     else:
257                         user.telopts[(telopt, party)] = NO
258             elif command is WILL:
259                 send_command(user, DONT, telopt)
260             else:
261                 send_command(user, WONT, telopt)
262             text = text[:position] + text[position + 3:]
263
264         # subnegotiation options
265         elif len_text > position + 4 and command is SB:
266             telopt = text[position + 2]
267             end_subnegotiation = text.find(telnet_proto(IAC, SE), position)
268             if end_subnegotiation > 0:
269                 if telopt is TELOPT_NAWS:
270                     user.columns = (
271                         text[position + 3] * 256 + text[position + 4])
272                     user.rows = (
273                         text[position + 5] * 256 + text[position + 6])
274                 elif telopt is TELOPT_TTYPE and text[position + 3] is IS:
275                     user.ttype = (
276                         text[position + 4:end_subnegotiation]).decode("ascii")
277                 text = text[:position] + text[end_subnegotiation + 2:]
278             else:
279                 position += 1
280
281         # otherwise, strip out a two-byte IAC command
282         elif len_text > position + 2:
283             log("Ignored unknown command %s from" % command, user)
284             text = text[:position] + text[position + 2:]
285
286         # and this means we got the beginning of an IAC
287         else:
288             position += 1
289
290     # replace the input with our cleaned-up text
291     user.partial_input = text