diff --git a/evennia/server/portal/telnet.py b/evennia/server/portal/telnet.py index fc16ca6ac6..57049d9b78 100644 --- a/evennia/server/portal/telnet.py +++ b/evennia/server/portal/telnet.py @@ -44,7 +44,7 @@ _IDLE_COMMAND = str.encode(settings.IDLE_COMMAND + "\n") # identify HTTP indata _HTTP_REGEX = re.compile( - r"(GET|HEAD|POST|PUT|DELETE|TRACE|OPTIONS|CONNECT|PATCH) (.*? HTTP/[0-9]\.[0-9])", re.I + rb"(GET|HEAD|POST|PUT|DELETE|TRACE|OPTIONS|CONNECT|PATCH) (.*? HTTP/[0-9]\.[0-9])", re.I ) _HTTP_WARNING = bytes( @@ -330,8 +330,10 @@ class TelnetProtocol(Telnet, StatefulTelnetProtocol, _BASE_SESSION_CLASS): data = [_IDLE_COMMAND] else: data = _RE_LINEBREAK.split(data) - - if len(data) > 2 and _HTTP_REGEX.match(data[0]): + # Normalize to bytes for regex match if needed. + if len(data) > 2 and _HTTP_REGEX.match( + data[0].encode("utf-8", errors="replace") if isinstance(data[0], str) else data[0] + ): # guard against HTTP request on the Telnet port; we # block and kill the connection. self.transport.write(_HTTP_WARNING) diff --git a/evennia/server/portal/tests.py b/evennia/server/portal/tests.py index b26cf0644b..2487272c71 100644 --- a/evennia/server/portal/tests.py +++ b/evennia/server/portal/tests.py @@ -236,6 +236,22 @@ class TestTelnet(TwistedTestCase): self.transport = proto_helpers.StringTransport() self.addCleanup(factory.sessionhandler.disconnect_all) + @mock.patch("evennia.server.portal.portalsessionhandler.reactor", new=MagicMock()) + def test_command_stacking_no_type_error(self): + self.transport.client = ["localhost"] + self.transport.setTcpKeepAlive = Mock() + d = self.proto.makeConnection(self.transport) + # Mudlet sends multiple commands in one packet when command stacking + data = b"wave\r\nsay hi\r\n" + try: + self.proto.dataReceived(data) + except TypeError: + self.fail("dataReceived raised TypeError on stacked commands") + # clean up to prevent Unclean reactor + self.proto.nop_keep_alive.stop() + self.proto._handshake_delay.cancel() + return d + @mock.patch("evennia.server.portal.portalsessionhandler.reactor", new=MagicMock()) def test_mudlet_ttype(self): self.transport.client = ["localhost"]