1 """Telnet functions and constants for the mudpy engine."""
3 # Copyright (c) 2004-2018 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",
27 # TODO(fungi): implement support for RFC 1091 terminal type information
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
84 def log(message, user):
85 """Log debugging info for Telnet client/server interactions."""
87 client = user.account.get("name", user)
90 mudpy.misc.log('[telnet] %s %s.' % (message, client), 0)
93 def telnet_proto(*arguments):
94 """Return a concatenated series of Telnet protocol commands."""
95 return bytes((arguments))
98 def translate_action(*command):
99 """Convert a Telnet command sequence into text suitable for logging."""
100 return "%s %s" % (command_names[command[0]], option_names[command[1]])
103 def send_command(user, *command):
104 """Sends a Telnet command string to the specified user's socket."""
105 user.send(telnet_proto(IAC, *command), raw=True)
106 log('Sent "%s" to' % translate_action(*command), user)
109 def is_enabled(user, telopt, party, state=YES):
110 """Indicates whether a specified Telnet option is enabled."""
111 if (telopt, party) in user.telopts and user.telopts[
119 def enable(user, telopt, party):
120 """Negotiates enabling a Telnet option for the indicated user's socket."""
125 if not (telopt, party) in user.telopts or user.telopts[
128 user.telopts[(telopt, party)] = WANTYES
129 send_command(user, txpos, telopt)
130 elif user.telopts[(telopt, party)] is WANTNO:
131 user.telopts[(telopt, party)] = WANTNO_OPPOSITE
132 elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
133 user.telopts[(telopt, party)] = WANTYES
136 def disable(user, telopt, party):
137 """Negotiates disabling a Telnet option for the user's socket."""
142 if not (telopt, party) in user.telopts:
143 user.telopts[(telopt, party)] = NO
144 elif user.telopts[(telopt, party)] is YES:
145 user.telopts[(telopt, party)] = WANTNO
146 send_command(user, txneg, telopt)
147 elif user.telopts[(telopt, party)] is WANTYES:
148 user.telopts[(telopt, party)] = WANTYES_OPPOSITE
149 elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
150 user.telopts[(telopt, party)] = WANTNO
153 def negotiate_telnet_options(user):
154 """Reply to and remove telnet negotiation options from partial_input."""
156 # make a local copy to play with
157 text = user.partial_input
159 # start at the begining of the input
162 # as long as we haven't checked it all
164 while position < len_text:
166 # jump to the first IAC you find
167 position = text.find(telnet_proto(IAC), position)
169 # if there wasn't an IAC in the input or it's at the end, we're done
170 if position < 0 or position >= len_text - 1:
173 # the byte following the IAC is our command
174 command = text[position+1]
176 # replace a double (literal) IAC if there's a CR+NUL or CR+LF later
179 text.find(b"\r\0", position) > 0 or
180 text.find(b"\r\n", position) > 0):
182 text = text[:position] + text[position + 1:]
183 log('Escaped IAC from', user)
187 # implement an RFC 1143 option negotiation queue here
188 elif len_text > position + 2 and WILL <= command <= DONT:
189 telopt = text[position+2]
190 log('Received "%s" from' % translate_action(command, telopt), user)
191 if telopt in supported:
202 if (telopt, party) not in user.telopts:
203 user.telopts[(telopt, party)] = NO
205 if user.telopts[(telopt, party)] is NO:
206 user.telopts[(telopt, party)] = YES
207 send_command(user, txpos, telopt)
208 elif user.telopts[(telopt, party)] is WANTNO:
209 user.telopts[(telopt, party)] = NO
210 elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
211 user.telopts[(telopt, party)] = YES
212 elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
213 user.telopts[(telopt, party)] = WANTNO
214 send_command(user, txneg, telopt)
216 user.telopts[(telopt, party)] = YES
218 if user.telopts[(telopt, party)] is YES:
219 user.telopts[(telopt, party)] = NO
220 send_command(user, txneg, telopt)
221 elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
222 user.telopts[(telopt, party)] = WANTYES
223 send_command(user, txpos, telopt)
225 user.telopts[(telopt, party)] = NO
226 elif command is WILL:
227 send_command(user, DONT, telopt)
229 send_command(user, WONT, telopt)
230 text = text[:position] + text[position + 3:]
232 # subnegotiation options
233 elif len_text > position + 4 and command is SB:
234 telopt = text[position + 2]
235 if telopt is TELOPT_NAWS:
237 text[position + 3] * 256 + text[position + 4])
238 end_subnegotiation = text.find(telnet_proto(IAC, SE), position)
239 if end_subnegotiation > 0:
240 text = text[:position] + text[end_subnegotiation + 2:]
244 # otherwise, strip out a two-byte IAC command
245 elif len_text > position + 2:
246 log("Ignored unknown command %s from" % command, user)
247 text = text[:position] + text[position + 2:]
249 # and this means we got the begining of an IAC
253 # replace the input with our cleaned-up text
254 user.partial_input = text