lib/mudpy/telnet.py

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