Safely log unknown Telnet options and commands

During Telnet negotiation, log unknown options by numeric value if
there is no listed name for them. Do the same for unknown Telnet
commands, though in reality this should never happen as they get
filtered by the existing implementation. Add regression testing to
make certain the crash bug which this fixes doesn't recur.

diff --git a/mudpy/telnet.py b/mudpy/telnet.py
index 0ce9aec..e483c13 100644 (file)
--- a/mudpy/telnet.py
+++ b/mudpy/telnet.py
@@ -1,6 +1,6 @@
 """Telnet functions and constants for the mudpy engine."""
 
-# Copyright (c) 2004-2018 mudpy authors. Permission to use, copy,
+# Copyright (c) 2004-2019 mudpy authors. Permission to use, copy,
 # modify, and distribute this software is granted under terms
 # provided in the LICENSE file distributed with this software.
 
@@ -97,7 +97,20 @@ def telnet_proto(*arguments):
 
 def translate_action(*command):
     """Convert a Telnet command sequence into text suitable for logging."""
-    return "%s %s" % (command_names[command[0]], option_names[command[1]])
+    try:
+        command_name = command_names[command[0]]
+    except KeyError:
+        # This should never happen since we filter unknown commands from
+        # the input queue, but added here for completeness since logging
+        # should never crash the process
+        command_name = str(command[0])
+    try:
+        option_name = option_names[command[1]]
+    except KeyError:
+        # This can happen for any of the myriad of Telnet options missing
+        # from the option_names dict
+        option_name = str(command[1])
+    return "%s %s" % (command_name, option_name)
 
 
 def send_command(user, *command):

diff --git a/mudpy/tests/selftest.py b/mudpy/tests/selftest.py
index 43e93ae..055e50a 100644 (file)
--- a/mudpy/tests/selftest.py
+++ b/mudpy/tests/selftest.py
@@ -184,13 +184,19 @@ test_telnet_iac = (
     (2, r"Non-ASCII characters from admin: b'say argle\\xffbargle'.*> ", ""),
 )
 
-test_telnet_unknown = (
+test_telnet_unknown_command = (
     # Send an unsupported negotiation command #127 which should get filtered
     # from the line of input
     (2, "> ", b"say glop\xff\x7fglyf\r\0"),
     (2, r'Ignored unknown command 127 from admin\..*"Glopglyf\.".*> ', ""),
 )
 
+test_telnet_unknown_option = (
+    # Send an unassigned negotiation option #127 which should get logged
+    (2, "> ", b"\xff\xfe\x7f\r\0"),
+    (2, r'''Received "don't 127" from admin\..*> ''', ""),
+)
+
 test_admin_restriction = (
     (0, "> ", "help halt"),
     (0, r"That is not an available command\.", "halt"),
@@ -317,7 +323,8 @@ dialogue = (
     (test_preferences, "set and show preferences"),
     (test_crlf_eol, "send crlf from the client as eol"),
     (test_telnet_iac, "escape stray telnet iac bytes"),
-    (test_telnet_unknown, "strip unknown telnet command"),
+    (test_telnet_unknown_command, "strip unknown telnet command"),
+    (test_telnet_unknown_option, "log unknown telnet option"),
     (test_admin_restriction, "restricted admin commands"),
     (test_admin_help, "admin help"),
     (test_reload, "reload"),
@@ -419,6 +426,17 @@ def tlog(message, quiet=False):
     return True
 
 
+def option_callback(telnet_socket, command, option):
+    if option == b'\x7f':
+        # We use this unassigned option value as a canary, so short-circuit
+        # any response to avoid endlessly looping
+        pass
+    elif command in (telnetlib.DO, telnetlib.DONT):
+        telnet_socket.send(b"%s%s%s" % (telnetlib.IAC, telnetlib.WONT, option))
+    elif command in (telnetlib.WILL, telnetlib.WONT):
+        telnet_socket.send(b"%s%s%s" % (telnetlib.IAC, telnetlib.DONT, option))
+
+
 def main():
     captures = ["", "", ""]
     lusers = [telnetlib.Telnet(), telnetlib.Telnet(), telnetlib.Telnet()]
@@ -430,6 +448,7 @@ def main():
         service = start_service(sys.argv[1])
     for luser in lusers:
         luser.open("::1", 4000)
+        luser.set_option_negotiation_callback(option_callback)
     for test, description in dialogue:
         tlog("\nTesting %s..." % description)
         test_start = time.time()