mudpy/telnet.py

   1 """Telnet functions and constants for the mudpy engine."""
   2
   3 # Copyright (c) 2004-2018 Jeremy Stanley <fungi@yuggoth.org>. Permission
   4 # to use, copy, modify, and distribute this software is granted under
   5 # terms 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
  18 supported = (
  19     TELOPT_BINARY,
  20     TELOPT_ECHO,
  21     TELOPT_SGA,
  22     TELOPT_EOR,
  23     TELOPT_NAWS,
  24     TELOPT_LINEMODE
  25 )
  26
  27 # telnet commands
  28 EOR = 239  # end-of-record signal (rfc 885)
  29 SE = 240  # end of subnegotiation parameters (rfc 854)
  30 GA = 249  # go ahead signal (rfc 854)
  31 SB = 250  # what follows is subnegotiation of the indicated option (rfc 854)
  32 WILL = 251  # desire or confirmation of performing an option (rfc 854)
  33 WONT = 252  # refusal or confirmation of performing an option (rfc 854)
  34 DO = 253  # request or confirm performing an option (rfc 854)
  35 DONT = 254  # demand or confirm no longer performing an option (rfc 854)
  36 IAC = 255  # interpret as command escape character (rfc 854)
  37
  38 # RFC 1143 option negotiation states
  39 NO = 0  # option is disabled
  40 YES = 1  # option is enabled
  41 WANTNO = 2  # demanded disabling option
  42 WANTYES = 3  # requested enabling option
  43 WANTNO_OPPOSITE = 4  # demanded disabling option but queued an enable after it
  44 WANTYES_OPPOSITE = 5  # requested enabling option but queued a disable after it
  45
  46 # RFC 1143 option negotiation parties
  47 HIM = 0
  48 US = 1
  49
  50
  51 def telnet_proto(*arguments):
  52     """Return a concatenated series of Telnet protocol commands."""
  53     return bytes((arguments))
  54
  55
  56 def send_command(user, *command):
  57     """Sends a Telnet command string to the specified user's socket."""
  58     user.send(telnet_proto(IAC, *command), raw=True)
  59
  60
  61 def is_enabled(user, telopt, party, state=YES):
  62     """Indicates whether a specified Telnet option is enabled."""
  63     if (telopt, party) in user.telopts and user.telopts[
  64        (telopt, party)
  65        ] is state:
  66         return True
  67     else:
  68         return False
  69
  70
  71 def enable(user, telopt, party):
  72     """Negotiates enabling a Telnet option for the indicated user's socket."""
  73     if party is HIM:
  74         txpos = DO
  75     else:
  76         txpos = WILL
  77     if not (telopt, party) in user.telopts or user.telopts[
  78        (telopt, party)
  79        ] is NO:
  80         user.telopts[(telopt, party)] = WANTYES
  81         send_command(user, txpos, telopt)
  82     elif user.telopts[(telopt, party)] is WANTNO:
  83         user.telopts[(telopt, party)] = WANTNO_OPPOSITE
  84     elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
  85         user.telopts[(telopt, party)] = WANTYES
  86
  87
  88 def disable(user, telopt, party):
  89     """Negotiates disabling a Telnet option for the user's socket."""
  90     if party is HIM:
  91         txneg = DONT
  92     else:
  93         txneg = WONT
  94     if not (telopt, party) in user.telopts:
  95         user.telopts[(telopt, party)] = NO
  96     elif user.telopts[(telopt, party)] is YES:
  97         user.telopts[(telopt, party)] = WANTNO
  98         send_command(user, txneg, telopt)
  99     elif user.telopts[(telopt, party)] is WANTYES:
 100         user.telopts[(telopt, party)] = WANTYES_OPPOSITE
 101     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
 102         user.telopts[(telopt, party)] = WANTNO
 103
 104
 105 def negotiate_telnet_options(user):
 106     """Reply to and remove telnet negotiation options from partial_input."""
 107
 108     # make a local copy to play with
 109     text = user.partial_input
 110
 111     # start at the begining of the input
 112     position = 0
 113
 114     # as long as we haven't checked it all
 115     len_text = len(text)
 116     while position < len_text:
 117
 118         # jump to the first IAC you find
 119         position = text.find(telnet_proto(IAC), position)
 120
 121         # if there wasn't an IAC in the input or it's at the end, we're done
 122         if position < 0 or position >= len_text - 1:
 123             break
 124
 125         # the byte following the IAC is our command
 126         command = text[position+1]
 127
 128         # replace a double (literal) IAC if there's a CR+NUL or CR+LF later
 129         if command is IAC:
 130             if (
 131                     text.find(b"\r\0", position) > 0 or
 132                     text.find(b"\r\n", position) > 0):
 133                 position += 1
 134                 text = text[:position] + text[position + 1:]
 135             else:
 136                 position += 2
 137
 138         # implement an RFC 1143 option negotiation queue here
 139         elif len_text > position + 2 and WILL <= command <= DONT:
 140             telopt = text[position+2]
 141             if telopt in supported:
 142                 if command <= WONT:
 143                     party = HIM
 144                     rxpos = WILL
 145                     txpos = DO
 146                     txneg = DONT
 147                 else:
 148                     party = US
 149                     rxpos = DO
 150                     txpos = WILL
 151                     txneg = WONT
 152                 if (telopt, party) not in user.telopts:
 153                     user.telopts[(telopt, party)] = NO
 154                 if command is rxpos:
 155                     if user.telopts[(telopt, party)] is NO:
 156                         user.telopts[(telopt, party)] = YES
 157                         send_command(user, txpos, telopt)
 158                     elif user.telopts[(telopt, party)] is WANTNO:
 159                         user.telopts[(telopt, party)] = NO
 160                     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
 161                         user.telopts[(telopt, party)] = YES
 162                     elif user.telopts[(telopt, party)] is WANTYES_OPPOSITE:
 163                         user.telopts[(telopt, party)] = WANTNO
 164                         send_command(user, txneg, telopt)
 165                     else:
 166                         user.telopts[(telopt, party)] = YES
 167                 else:
 168                     if user.telopts[(telopt, party)] is YES:
 169                         user.telopts[(telopt, party)] = NO
 170                         send_command(user, txneg, telopt)
 171                     elif user.telopts[(telopt, party)] is WANTNO_OPPOSITE:
 172                         user.telopts[(telopt, party)] = WANTYES
 173                         send_command(user, txpos, telopt)
 174                     else:
 175                         user.telopts[(telopt, party)] = NO
 176             elif command is WILL:
 177                 send_command(user, DONT, telopt)
 178             else:
 179                 send_command(user, WONT, telopt)
 180             text = text[:position] + text[position + 3:]
 181
 182         # subnegotiation options
 183         elif len_text > position + 4 and command is SB:
 184             telopt = text[position + 2]
 185             if telopt is TELOPT_NAWS:
 186                 user.columns = (
 187                     text[position + 3] * 256 + text[position + 4])
 188             end_subnegotiation = text.find(telnet_proto(IAC, SE), position)
 189             if end_subnegotiation > 0:
 190                 text = text[:position] + text[end_subnegotiation + 2:]
 191             else:
 192                 position += 1
 193
 194         # otherwise, strip out a two-byte IAC command
 195         elif len_text > position + 2:
 196             mudpy.misc.log("Unknown Telnet IAC command %s ignored." % command)
 197             text = text[:position] + text[position + 2:]
 198
 199         # and this means we got the begining of an IAC
 200         else:
 201             position += 1
 202
 203     # replace the input with our cleaned-up text
 204     user.partial_input = text