1 """Telnet functions and constants for the mudpy engine."""
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.
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)
18 TELOPT_BINARY: "8-bit binary",
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",
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)
49 SE: "subnegotiation end",
51 SB: "subnegotiation begin",
56 IAC: "interpret as command",
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
69 WANTNO: "demand disabling",
70 WANTYES: "request enabling",
71 WANTNO_OPPOSITE: "want no queued opposite",
72 WANTYES_OPPOSITE: "want yes queued opposite",
75 # RFC 1143 option negotiation parties
86 ttype_command_names = {
92 def log(message, user):
93 """Log debugging info for Telnet client/server interactions."""
95 client = user.account.get("name", user)
98 mudpy.misc.log('[telnet] %s %s.' % (message, client), 0)
101 def telnet_proto(*arguments):
102 """Return a concatenated series of Telnet protocol commands."""
103 return bytes((arguments))
106 def translate_action(*command):
107 """Convert a Telnet command sequence into text suitable for logging."""
109 command_name = command_names[command[0]]
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])
116 option_name = option_names[command[1]]
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)
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)
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[
140 def enable(user, telopt, party):
141 """Negotiates enabling a Telnet option for the indicated user's socket."""
146 if not (telopt, party) in user.telopts or user.telopts[
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
157 def disable(user, telopt, party):
158 """Negotiates disabling a Telnet option for the user's socket."""
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
174 def request_ttype(user):
175 """Clear and request the terminal type."""
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
181 user.send(telnet_proto(IAC, SB, TELOPT_TTYPE, SEND, IAC, SE), raw=True)
182 log('Sent terminal type request to', user)
185 def negotiate_telnet_options(user):
186 """Reply to and remove telnet negotiation options from partial_input."""
188 # make a local copy to play with
189 text = user.partial_input
191 # start at the beginning of the input
194 # as long as we haven't checked it all
196 while position < len_text:
198 # jump to the first IAC you find
199 position = text.find(telnet_proto(IAC), position)
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:
205 # the byte following the IAC is our command
206 command = text[position+1]
208 # replace a double (literal) IAC if there's a CR+NUL or CR+LF later
211 text.find(b"\r\0", position) > 0 or
212 text.find(b"\r\n", position) > 0):
214 text = text[:position] + text[position + 1:]
215 log('Escaped IAC from', user)
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:
234 if (telopt, party) not in user.telopts:
235 user.telopts[(telopt, party)] = NO
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)
248 user.telopts[(telopt, party)] = YES
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)
257 user.telopts[(telopt, party)] = NO
258 elif command is WILL:
259 send_command(user, DONT, telopt)
261 send_command(user, WONT, telopt)
262 text = text[:position] + text[position + 3:]
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:
271 text[position + 3] * 256 + text[position + 4])
273 text[position + 5] * 256 + text[position + 6])
274 elif telopt is TELOPT_TTYPE and text[position + 3] is IS:
276 text[position + 4:end_subnegotiation]).decode("ascii")
277 text = text[:position] + text[end_subnegotiation + 2:]
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:]
286 # and this means we got the beginning of an IAC
290 # replace the input with our cleaned-up text
291 user.partial_input = text