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 begining 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 begining of an IAC
 287         else:
 288             position += 1
 289
 290     # replace the input with our cleaned-up text
 291     user.partial_input = text