Revert "Backport telnetlib3 changes"

gns3server/compute/base_node.py

@@ -397,7 +397,11 @@ class BaseNode:
             echo=True
         )
         # warning: this will raise OSError exception if there is a problem...
-        self._wrapper_telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
+        self._wrapper_telnet_server = await asyncio.start_server(
+            server.run,
+            self._manager.port_manager.console_host,
+            self.console
+        )
 
     async def stop_wrap_console(self):
         """

gns3server/compute/docker/docker_vm.py

@@ -587,7 +587,7 @@ class DockerVM(BaseNode):
             raise DockerError("Could not start auxiliary console process: {}".format(e))
         server = AsyncioTelnetServer(reader=process.stdout, writer=process.stdin, binary=True, echo=True)
         try:
-            self._telnet_servers.append(await server.start(self._manager.port_manager.console_host, self.aux))
+            self._telnet_servers.append((await asyncio.start_server(server.run, self._manager.port_manager.console_host, self.aux)))
         except OSError as e:
             raise DockerError("Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.aux, e))
         log.debug("Docker container '%s' started listen for auxiliary telnet on %d", self.name, self.aux)
@@ -747,7 +747,7 @@ class DockerVM(BaseNode):
         input_stream = InputStream()
         telnet = AsyncioTelnetServer(reader=output_stream, writer=input_stream, echo=True, naws=True, window_size_changed_callback=self._window_size_changed_callback)
         try:
-            self._telnet_servers.append(await telnet.start(self._manager.port_manager.console_host, self.console))
+            self._telnet_servers.append((await asyncio.start_server(telnet.run, self._manager.port_manager.console_host, self.console)))
         except OSError as e:
             raise DockerError("Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.console, e))
 

gns3server/compute/iou/iou_vm.py

@@ -598,7 +598,8 @@ class IOUVM(BaseNode):
             server = AsyncioTelnetServer(reader=self._iou_process.stdout, writer=self._iou_process.stdin, binary=True,
                                          echo=True)
             try:
-                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
+                self._telnet_server = await asyncio.start_server(server.run, self._manager.port_manager.console_host,
+                                                                 self.console)
             except OSError as e:
                 await self.stop()
                 raise IOUError(

gns3server/compute/virtualbox/virtualbox_vm.py

@@ -982,7 +982,7 @@ class VirtualBoxVM(BaseNode):
                                          binary=True,
                                          echo=True)
             try:
-                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
+                self._telnet_server = await asyncio.start_server(server.run, self._manager.port_manager.console_host, self.console)
             except OSError as e:
                 self.project.emit("log.warning", {"message": "Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.console, e)})
 

gns3server/compute/vmware/vmware_vm.py

@@ -881,7 +881,7 @@ class VMwareVM(BaseNode):
                                          binary=True,
                                          echo=True)
             try:
-                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
+                self._telnet_server = await asyncio.start_server(server.run, self._manager.port_manager.console_host, self.console)
             except OSError as e:
                 self.project.emit("log.warning", {"message": "Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.console, e)})
 

gns3server/utils/asyncio/embed_shell.py

@@ -352,12 +352,12 @@ if __name__ == '__main__':
     # Demo using telnet
     shell = Demo(welcome_message="Welcome!\n")
     server = create_telnet_shell(shell, loop=loop)
-    loop.run_until_complete(server.start("127.0.0.1", 4444))
+    coro = asyncio.start_server(server.run, '127.0.0.1', 4444)
+    s = loop.run_until_complete(coro)
     try:
         loop.run_forever()
     except KeyboardInterrupt:
         pass
-    loop.run_until_complete(server.close())
 
     # Demo using stdin
     # loop.run_until_complete(create_stdin_shell(Demo()))

gns3server/utils/asyncio/telnet_server.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 #
 # Copyright (C) 2014 GNS3 Technologies Inc.
 #
@@ -14,47 +15,52 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import sys
+import socket
 import asyncio
 import asyncio.subprocess
-import contextlib
+import struct
+
 import logging
-import socket
-
-import telnetlib3
-from telnetlib3.server import TelnetServer
-from telnetlib3.telopt import DONT, ECHO, IAC, NAWS, NOP, WILL, WONT
-
 log = logging.getLogger(__name__)
 
+# Mostly from https://code.google.com/p/miniboa/source/browse/trunk/miniboa/telnet.py
+
+# Telnet Commands
+SE = 240    # End of sub-negotiation parameters
+NOP = 241    # No operation
+DATMK = 242    # Data stream portion of a sync.
+BREAK = 243    # NVT Character BRK
+IP = 244    # Interrupt Process
+AO = 245    # Abort Output
+AYT = 246    # Are you there
+EC = 247    # Erase Character
+EL = 248    # Erase Line
+GA = 249    # The Go Ahead Signal
+SB = 250    # Sub-option to follow
+WILL = 251    # Will; request or confirm option begin
+WONT = 252    # Wont; deny option request
+DO = 253    # Do = Request or confirm remote option
+DONT = 254    # Don't = Demand or confirm option halt
+IAC = 255    # Interpret as Command
+SEND = 1      # Sub-process negotiation SEND command
+IS = 0      # Sub-process negotiation IS command
+
+# Telnet Options
+BINARY = 0      # Transmit Binary
+ECHO = 1      # Echo characters back to sender
+RECON = 2      # Reconnection
+SGA = 3      # Suppress Go-Ahead
+TMARK = 6      # Timing Mark
+TTYPE = 24     # Terminal Type
+NAWS = 31     # Negotiate About Window Size
+LINEMO = 34     # Line Mode
+
 READ_SIZE = 1024
-BROADCAST_DRAIN_TIMEOUT = 10
-KEEPALIVE_INTERVAL = 60  # Send NOP every 60 seconds
 
 
-class _ManagedTelnetListener:
+class TelnetConnection(object):
-    """Compatibility wrapper that owns AsyncioTelnetServer shutdown."""
-
-    def __init__(self, telnet_server, listener):
-        self._telnet_server = telnet_server
-        self._listener = listener
-        self._close_task = None
-
-    def close(self):
-        if self._close_task is None:
-            self._close_task = asyncio.create_task(self._telnet_server.close())
-
-    async def wait_closed(self):
-        self.close()
-        with contextlib.suppress(asyncio.CancelledError):
-            await self._close_task
-
-    def __getattr__(self, attribute):
-        return getattr(self._listener, attribute)
-
-
-class TelnetConnection:
     """Default implementation of telnet connection which may but may not be used."""
-
     def __init__(self, reader, writer, window_size_changed_callback=None):
         self.is_closing = False
         self._reader = reader
@@ -79,66 +85,59 @@ class TelnetConnection:
 
     async def window_size_changed(self, columns, rows):
         """Method called when window size changed, only can occur when
-        `naws` flag is enabled in server configuration."""
+         `naws` flag is enable in server configuration."""
 
         if self._window_size_changed_callback:
             await self._window_size_changed_callback(columns, rows)
 
     async def feed(self, data):
-        """Handles incoming data."""
+        """
+        Handles incoming data
+        :return:
+        """
 
     def send(self, data):
-        """Send data back to client."""
+        """
-
+        Sending data back to client
+        :return:
+        """
         data = data.decode().replace("\n", "\r\n")
         self.writer.write(data.encode())
 
     def close(self):
-        """Close current connection."""
+        """
-
+        Closes current connection
+        :return:
+        """
         self.is_closing = True
 
 
 class AsyncioTelnetServer:
-    def __init__(
+    MAX_NEGOTIATION_READ = 10
-        self,
-        reader=None,
-        writer=None,
-        binary=True,
-        echo=False,
-        naws=False,
-        window_size_changed_callback=None,
-        connection_factory=None,
-        keepalive_interval=KEEPALIVE_INTERVAL,
-    ):
-        """
-        Initialize telnet server.
 
-        :param naws: when True, window size negotiation callbacks are enabled.
+    def __init__(self, reader=None, writer=None, binary=True, echo=False, naws=False, window_size_changed_callback=None, connection_factory=None):
-        :param connection_factory: optional factory to inject a custom connection implementation.
-        :param keepalive_interval: interval in seconds for sending NOP keep-alive (0 to disable).
         """
-
+        Initializes telnet server
-        assert connection_factory is None or (
+        :param naws when True make a window size negotiation
-            connection_factory is not None and reader is None and writer is None
+        :param connection_factory: when set it's possible to inject own implementation of connection
-        ), "Please use either reader and writer either connection_factory, otherwise duplicate data may be produced."
+        """
+        assert connection_factory is None or (connection_factory is not None and reader is None and writer is None), \
+            "Please use either reader and writer either connection_factory, otherwise duplicate data may be produced."
 
         self._reader = reader
         self._writer = writer
+        self._connections = dict()
+        self._lock = asyncio.Lock()
+        self._reader_process = None
+        self._current_read = None
         self._window_size_changed_callback = window_size_changed_callback
+
         self._binary = binary
+        # If echo is true when the client send data
+        # the data is echo on his terminal by telnet otherwise
+        # it's our job (or the wrapped app) to send back the data
         self._echo = echo
         self._naws = naws
-        self._keepalive_interval = keepalive_interval
-
-        self._connections = {}
-        self._pending_window_sizes = {}
-        self._connections_lock = asyncio.Lock()
-        self._close_lock = asyncio.Lock()
-        self._broadcast_task = None
-        self._keepalive_task = None
-        self._server = None
-        self._server_handle = None
 
         def default_connection_factory(reader, writer, window_size_changed_callback):
             return TelnetConnection(reader, writer, window_size_changed_callback)
@@ -150,304 +149,317 @@ class AsyncioTelnetServer:
 
     @staticmethod
     async def write_client_intro(writer, echo=False):
-        """Write a minimal telnet intro to an upstream console endpoint."""
+        # Send initial telnet session opening
-
         if echo:
-            writer.write(IAC + WILL + ECHO)
+            writer.write(bytes([IAC, WILL, ECHO]))
         else:
-            writer.write(IAC + WONT + ECHO + IAC + DONT + ECHO)
+            writer.write(bytes([
+                IAC, WONT, ECHO,
+                IAC, DONT, ECHO]))
         await writer.drain()
 
-    async def start(self, host, port):
+    async def _write_intro(self, writer, binary=False, echo=False, naws=False):
-        """Start a telnetlib3-backed listener and return a managed server handle."""
+        # Send initial telnet session opening
+        if echo:
+            writer.write(bytes([IAC, WILL, ECHO]))
+        else:
+            writer.write(bytes([
+                IAC, WONT, ECHO,
+                IAC, DONT, ECHO]))
 
-        if self._server is not None:
+        if binary:
-            raise RuntimeError("AsyncioTelnetServer is already started")
+            writer.write(bytes([
+                IAC, WILL, SGA,
+                IAC, WILL, BINARY,
+                IAC, DO, BINARY]))
+        else:
+            writer.write(bytes([
+                IAC, WONT, SGA,
+                IAC, DONT, SGA,
+                IAC, WONT, BINARY,
+                IAC, DONT, BINARY]))
 
-        protocol_factory = self._build_protocol_factory()
+        if naws:
-        self._server = await telnetlib3.create_server(
+            writer.write(bytes([
-            host=host,
+                IAC, DO, NAWS
-            port=port,
+            ]))
-            protocol_factory=protocol_factory,
+        await writer.drain()
-            shell=self._run_client_session,
-            encoding=False,
-            force_binary=self._binary,
-            never_send_ga=True,
-            line_mode=not self._binary,
-            timeout=0,
-            connect_maxwait=1.0,
-        )
-        self._server_handle = _ManagedTelnetListener(self, self._server)
-
-        if self._reader is not None and self._broadcast_task is None:
-            self._broadcast_task = asyncio.create_task(self._broadcast_from_upstream())
-
-        if self._keepalive_interval > 0 and self._keepalive_task is None:
-            self._keepalive_task = asyncio.create_task(self._send_keepalives())
-
-        return self._server_handle
 
     async def run(self, network_reader, network_writer):
-        """Backward-compatible entrypoint for asyncio.start_server(server.run, ...)."""
-
-        await self._run_client_session(network_reader, network_writer)
-
-    async def close(self):
-        async with self._close_lock:
-            if self._keepalive_task is not None:
-                keepalive_task = self._keepalive_task
-                self._keepalive_task = None
-                keepalive_task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await keepalive_task
-
-            if self._broadcast_task is not None:
-                broadcast_task = self._broadcast_task
-                self._broadcast_task = None
-                broadcast_task.cancel()
-                with contextlib.suppress(asyncio.CancelledError):
-                    await broadcast_task
-            else:
-                await self._disconnect_all_clients()
-
-            if self._server is not None:
-                self._server.close()
-                await self._server.wait_closed()
-                self._server = None
-
-            self._server_handle = None
-
-    async def client_connected_hook(self):
-        pass
-
-    def _build_protocol_factory(self):
-        parent = self
-
-        class GNS3TelnetServer(TelnetServer):
-            def _negotiate_echo(self):
-                if self._echo_negotiated:
-                    return
-                self._echo_negotiated = True
-
-                if self.line_mode:
-                    return
-
-                if parent._echo:
-                    self.writer.iac(WILL, ECHO)
-                else:
-                    self.writer.iac(WONT, ECHO)
-                    self.writer.iac(DONT, ECHO)
-
-            def begin_advanced_negotiation(self):
-                super().begin_advanced_negotiation()
-                if not parent._naws:
-                    self.writer.iac(DONT, NAWS)
-
-            def on_naws(self, rows, cols):
-                super().on_naws(rows, cols)
-                parent._handle_naws(self.writer, cols, rows)
-
-        return GNS3TelnetServer
-
-    def _handle_naws(self, writer, columns, rows):
-        if not self._naws:
-            return
-        asyncio.create_task(self._dispatch_window_size(writer, columns, rows))
-
-    async def _dispatch_window_size(self, writer, columns, rows):
-        async with self._connections_lock:
-            connection = self._connections.get(writer)
-            if connection is None:
-                self._pending_window_sizes[writer] = (columns, rows)
-                return
-
-        await self._invoke_window_size_changed(connection, columns, rows)
-
-    async def _invoke_window_size_changed(self, connection, columns, rows):
-        try:
-            await connection.window_size_changed(columns, rows)
-        except asyncio.CancelledError:
-            raise
-        except (ConnectionError, OSError):
-            connection.close()
-        except Exception:
-            log.exception("Unhandled exception in window_size_changed callback for %r", connection)
-
-    async def _run_client_session(self, network_reader, network_writer):
-        self._set_socket_options(network_writer)
-
-        connection = self._connection_factory(network_reader, network_writer, self._window_size_changed_callback)
-
-        async with self._connections_lock:
-            self._connections[network_writer] = connection
-            pending_window_size = self._pending_window_sizes.pop(network_writer, None)
-
-        if pending_window_size is not None:
-            columns, rows = pending_window_size
-            await self._invoke_window_size_changed(connection, columns, rows)
-
-        try:
-            await connection.connected()
-            await self.client_connected_hook()
-
-            while True:
-                data = await network_reader.read(READ_SIZE)
-                if not data:
-                    break
-
-                if not self._binary:
-                    data = data.replace(b"\r\n", b"\n")
-
-                if self._writer is not None:
-                    self._writer.write(data)
-                    await self._writer.drain()
-
-                await connection.feed(data)
-                if connection.is_closing:
-                    break
-
-        except asyncio.CancelledError:
-            raise
-        except (ConnectionError, OSError):
-            pass
-        finally:
-            await self._disconnect_client(network_writer)
-
-    async def _broadcast_from_upstream(self):
-        try:
-            while True:
-                data = await self._reader.read(READ_SIZE)
-                if not data:
-                    break
-
-                for network_writer, connection in await self._get_connections_snapshot():
-                    try:
-                        connection.writer.write(data)
-                        await asyncio.wait_for(connection.writer.drain(), timeout=BROADCAST_DRAIN_TIMEOUT)
-                    except (OSError, ConnectionError, asyncio.TimeoutError) as e:
-                        client_info = self._get_peername(network_writer)
-                        log.debug(
-                            "Error sending data to client %s: %s, closing and removing from connection table.",
-                            client_info,
-                            e,
-                        )
-                        connection.close()
-                        await self._disconnect_client(network_writer)
-        except asyncio.CancelledError:
-            raise
-        except (ConnectionError, OSError):
-            pass
-        finally:
-            await self._disconnect_all_clients()
-
-    async def _send_keepalives(self):
-        """Periodically send IAC NOP to all connected clients to keep sessions alive."""
-
-        try:
-            while True:
-                await asyncio.sleep(self._keepalive_interval)
-                for network_writer, connection in await self._get_connections_snapshot():
-                    client_info = self._get_peername(network_writer)
-                    try:
-                        log.debug("Sending keepalive to client %s", client_info)
-                        connection.writer.send_iac(IAC + NOP)
-                        await asyncio.wait_for(connection.writer.drain(), timeout=BROADCAST_DRAIN_TIMEOUT)
-                    except (OSError, ConnectionError, asyncio.TimeoutError) as e:
-                        log.debug(
-                            "Keepalive failed for client %s: %s, closing connection.",
-                            client_info,
-                            e,
-                        )
-                        connection.close()
-                        await self._disconnect_client(network_writer)
-        except asyncio.CancelledError:
-            raise
-
-    async def _get_connections_snapshot(self):
-        async with self._connections_lock:
-            return list(self._connections.items())
-
-    async def _disconnect_all_clients(self):
-        async with self._connections_lock:
-            writers = list(self._connections.keys())
-
-        for network_writer in writers:
-            await self._disconnect_client(network_writer)
-
-    async def _disconnect_client(self, network_writer):
-        async with self._connections_lock:
-            connection = self._connections.pop(network_writer, None)
-            self._pending_window_sizes.pop(network_writer, None)
-
-        if connection is not None:
-            with contextlib.suppress(Exception):
-                await connection.disconnected()
-
-        with contextlib.suppress(AttributeError, OSError):
-            network_writer.close()
-
-        wait_closed = getattr(network_writer, "wait_closed", None)
-        if callable(wait_closed):
-            with contextlib.suppress(ConnectionError, OSError):
-                await wait_closed()
-
-    @staticmethod
-    def _get_peername(network_writer):
-        with contextlib.suppress(OSError, AttributeError):
-            sock = network_writer.get_extra_info("socket")
-            if sock is not None:
-                return sock.getpeername()
-        return network_writer.get_extra_info("peername")
-
-    @staticmethod
-    def _set_socket_options(network_writer):
         sock = network_writer.get_extra_info("socket")
-        if sock is None:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-            return
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
+        # 60 sec keep alives, close tcp session after 4 missed
-        with contextlib.suppress(OSError):
+        # Will keep a firewall from aging out telnet console.
-            sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
-        # 60 sec keep alives, close tcp session after 4 missed.
-        # This keeps stateful firewalls from aging out long-lived sessions.
         try:
+            # Keepalive options are platform dependent: Linux uses TCP_KEEPIDLE,
+            # while macOS exposes TCP_KEEPALIVE (in Python >= 3.10).
             if hasattr(socket, "TCP_KEEPIDLE"):
                 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
             elif hasattr(socket, "TCP_KEEPALIVE"):
                 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPALIVE, 60)
             else:
-                raise AttributeError("No TCP keepalive idle socket option is available")
+                raise AttributeError("module 'socket' has no attribute 'TCP_KEEPIDLE' or 'TCP_KEEPALIVE'")
-
             sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
             sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 4)
         except (AttributeError, OSError):
             log.debug("Failed to tune TCP keepalive for telnet client; using OS defaults", exc_info=True)
 
+        #log.debug("New connection from {}".format(sock.getpeername()))
 
-if __name__ == "__main__":
+        # Keep track of connected clients
+        connection = self._connection_factory(network_reader, network_writer, self._window_size_changed_callback)
+        self._connections[network_writer] = connection
+
+        try:
+            await self._write_intro(network_writer, echo=self._echo, binary=self._binary, naws=self._naws)
+            await connection.connected()
+            await self._process(network_reader, network_writer, connection)
+        except (ConnectionError, OSError):
+            async with self._lock:
+                network_writer.close()
+                # await network_writer.wait_closed()  # this doesn't work in Python 3.6
+                if self._reader_process == network_reader:
+                    self._reader_process = None
+                    # Cancel current read from this reader
+                    if self._current_read is not None:
+                        self._current_read.cancel()
+
+            await connection.disconnected()
+            del self._connections[network_writer]
+
+    async def close(self):
+        for writer, connection in self._connections.items():
+            try:
+                writer.write_eof()
+                await writer.drain()
+                writer.close()
+                # await writer.wait_closed()  # this doesn't work in Python 3.6
+            except (AttributeError, ConnectionError):
+                continue
+
+    async def client_connected_hook(self):
+        pass
+
+    async def _get_reader(self, network_reader):
+        """
+        Get a reader or None if another reader is already reading.
+        """
+        async with self._lock:
+            if self._reader_process is None:
+                self._reader_process = network_reader
+            if self._reader:
+                if self._reader_process == network_reader:
+                    self._current_read = asyncio.ensure_future(self._reader.read(READ_SIZE))
+                    return self._current_read
+        return None
+
+    async def _process(self, network_reader, network_writer, connection):
+        network_read = asyncio.ensure_future(network_reader.read(READ_SIZE))
+        reader_read = await self._get_reader(network_reader)
+
+        while True:
+            if reader_read is None:
+                reader_read = await self._get_reader(network_reader)
+            if reader_read is None:
+                done, pending = await asyncio.wait(
+                    [
+                        network_read,
+                    ],
+                    timeout=1,
+                    return_when=asyncio.FIRST_COMPLETED)
+            else:
+                done, pending = await asyncio.wait(
+                    [
+                        network_read,
+                        reader_read
+                    ],
+                    return_when=asyncio.FIRST_COMPLETED)
+            for coro in done:
+                data = coro.result()
+                if coro == network_read:
+                    if network_reader.at_eof():
+                        raise ConnectionResetError()
+
+                    network_read = asyncio.ensure_future(network_reader.read(READ_SIZE))
+
+                    if IAC in data:
+                        data = await self._IAC_parser(data, network_reader, network_writer, connection)
+
+                    if len(data) == 0:
+                        continue
+
+                    if not self._binary:
+                        data = data.replace(b"\r\n", b"\n")
+
+                    if self._writer:
+                        self._writer.write(data)
+                        await self._writer.drain()
+
+                    await connection.feed(data)
+                    if connection.is_closing:
+                        raise ConnectionResetError()
+
+                elif coro == reader_read:
+                    if self._reader and self._reader.at_eof():
+                        raise ConnectionResetError()
+
+                    reader_read = await self._get_reader(network_reader)
+
+                    # Replicate the output on all clients
+                    for connection_key in list(self._connections.keys()):
+                        connection = self._connections[connection_key]
+                        client_info = None
+
+                        try:
+                            client_info = connection_key.get_extra_info("socket").getpeername()
+                            connection.writer.write(data)
+                            await asyncio.wait_for(connection.writer.drain(), timeout=10)
+                        except (OSError, ConnectionError, asyncio.TimeoutError) as e:
+                            log.debug(f"Error sending data to client {client_info}: {e}, closing and removing from connection table.")
+                            connection.close()
+                            del self._connections[connection_key]
+
+    async def _read(self, cmd, buffer, location, reader):
+        """ Reads next op from the buffer or reader"""
+        try:
+            op = buffer[location]
+            cmd.append(op)
+            return op
+        except IndexError:
+            op = await reader.read(1)
+            buffer.extend(op)
+            cmd.append(buffer[location])
+            return op
+
+    async def _negotiate(self, data, connection):
+        """ Performs negotiation commands"""
+
+        command, payload = data[0], data[1:]
+        if command == NAWS:
+            if len(payload) == 4:
+                columns, rows = struct.unpack(str('!HH'), bytes(payload))
+                await connection.window_size_changed(columns, rows)
+            else:
+                log.warning('Wrong number of NAWS bytes')
+        else:
+            log.debug("Not supported negotiation sequence, received {} bytes", len(data))
+
+    async def _IAC_parser(self, buf, network_reader, network_writer, connection):
+        """
+        Processes and removes any Telnet commands from the buffer.
+
+        :param buf: buffer
+        :returns: buffer minus Telnet commands
+        """
+
+
+        skip_to = 0
+        while True:
+            # Locate an IAC to process
+            iac_loc = buf.find(IAC, skip_to)
+            if iac_loc < 0:
+                break
+
+            # Get the TELNET command
+            iac_cmd = bytearray([IAC])
+            try:
+                iac_cmd.append(buf[iac_loc + 1])
+            except IndexError:
+                d = await network_reader.read(1)
+                buf.extend(d)
+                iac_cmd.append(buf[iac_loc + 1])
+
+            # Is this just a 2-byte TELNET command?
+            if iac_cmd[1] not in [WILL, WONT, DO, DONT, SB]:
+                if iac_cmd[1] == AYT:
+                    log.debug("Telnet server received Are-You-There (AYT)")
+                    network_writer.write(b'\r\nYour Are-You-There received. I am here.\r\n')
+                elif iac_cmd[1] == IAC:
+                    # It's data, not an IAC
+                    iac_cmd.pop()
+                    # This prevents the 0xff from being
+                    # interrupted as yet another IAC
+                    skip_to = iac_loc + 1
+                    log.debug("Received IAC IAC")
+                elif iac_cmd[1] == NOP:
+                    pass
+                else:
+                    log.debug("Unhandled telnet command: "
+                              "{0:#x} {1:#x}".format(*iac_cmd))
+            elif iac_cmd[1] == SB:  # starts negotiation commands
+                negotiation = []
+                for pos in range(2, self.MAX_NEGOTIATION_READ):
+                    op = await self._read(iac_cmd, buf, iac_loc + pos, network_reader)
+                    negotiation.append(op)
+                    if op == SE:
+                        # ends negotiation commands
+                        break
+
+                # SE command is followed by IAC, remove the last two operations from stack
+                await self._negotiate(negotiation[0:-2], connection)
+
+            # This must be a 3-byte TELNET command
+            else:
+                try:
+                    iac_cmd.append(buf[iac_loc + 2])
+                except IndexError:
+                    d = await network_reader.read(1)
+                    buf.extend(d)
+                    iac_cmd.append(buf[iac_loc + 2])
+                # We do ECHO, SGA, and BINARY. Period.
+                if iac_cmd[1] == DO:
+                    if iac_cmd[2] not in [ECHO, SGA, BINARY]:
+                        network_writer.write(bytes([IAC, WONT, iac_cmd[2]]))
+                        log.debug("Telnet WON'T {:#x}".format(iac_cmd[2]))
+                    else:
+                        if iac_cmd[2] == SGA:
+                            if self._binary:
+                                network_writer.write(bytes([IAC, WILL, iac_cmd[2]]))
+                            else:
+                                network_writer.write(bytes([IAC, WONT, iac_cmd[2]]))
+                                log.debug("Telnet WON'T {:#x}".format(iac_cmd[2]))
+
+                elif iac_cmd[1] == DONT:
+                    log.debug("Unhandled DONT telnet command: "
+                              "{0:#x} {1:#x} {2:#x}".format(*iac_cmd))
+                elif iac_cmd[1] == WILL:
+                    if iac_cmd[2] not in [BINARY, NAWS]:
+                        log.debug("Unhandled WILL telnet command: "
+                                  "{0:#x} {1:#x} {2:#x}".format(*iac_cmd))
+                elif iac_cmd[1] == WONT:
+                    log.debug("Unhandled WONT telnet command: "
+                              "{0:#x} {1:#x} {2:#x}".format(*iac_cmd))
+                else:
+                    log.debug("Unhandled telnet command: "
+                              "{0:#x} {1:#x} {2:#x}".format(*iac_cmd))
+
+            # Remove the entire TELNET command from the buffer
+            buf = buf.replace(iac_cmd, b'', 1)
+
+            await network_writer.drain()
+
+        # Return the new copy of the buffer, minus telnet commands
+        return buf
+
+if __name__ == '__main__':
     logging.basicConfig(level=logging.DEBUG)
     loop = asyncio.get_event_loop()
 
-    process = loop.run_until_complete(
+    process = loop.run_until_complete(asyncio.ensure_future(asyncio.subprocess.create_subprocess_exec("/bin/sh", "-i",
-        asyncio.ensure_future(
+                                                                                                      stdout=asyncio.subprocess.PIPE,
-            asyncio.subprocess.create_subprocess_exec(
+                                                                                                      stderr=asyncio.subprocess.STDOUT,
-                "/bin/sh",
+                                                                                                      stdin=asyncio.subprocess.PIPE)))
-                "-i",
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.STDOUT,
-                stdin=asyncio.subprocess.PIPE,
-            )
-        )
-    )
     server = AsyncioTelnetServer(reader=process.stdout, writer=process.stdin, binary=False, echo=False)
 
-    loop.run_until_complete(server.start("127.0.0.1", 4444))
+    coro = asyncio.start_server(server.run, '127.0.0.1', 4444)
+    s = loop.run_until_complete(coro)
 
     try:
         loop.run_forever()
     except KeyboardInterrupt:
         pass
-
+    # Close the server
-    loop.run_until_complete(server.close())
+    s.close()
+    loop.run_until_complete(s.wait_closed())
     loop.close()

requirements.txt

@@ -10,4 +10,3 @@ distro>=1.9.0
 py-cpuinfo>=9.0.0,<10.0
 platformdirs>=2.4.0,<3  # platformdirs >=3 conflicts when building Debian packages
 truststore>=0.10.4; python_version >= '3.10'
-telnetlib3==4.0.1

tests/utils/asyncio/test_telnet_server.py

@@ -1,177 +0,0 @@
-import asyncio
-import contextlib
-
-import pytest
-import telnetlib3
-
-from gns3server.utils.asyncio.telnet_server import AsyncioTelnetServer
-
-
-class DummyUpstreamWriter:
-    def __init__(self):
-        self.buffer = bytearray()
-        self.write_event = asyncio.Event()
-
-    def write(self, data):
-        self.buffer.extend(data)
-        self.write_event.set()
-
-    async def drain(self):
-        await asyncio.sleep(0)
-
-
-def _get_listen_port(server):
-    return server.sockets[0].getsockname()[1]
-
-
-async def _wait_for_connection_count(server, expected_count, timeout=2):
-    deadline = asyncio.get_running_loop().time() + timeout
-    last_count = 0
-    while asyncio.get_running_loop().time() < deadline:
-        async with server._connections_lock:
-            last_count = len(server._connections)
-        if last_count == expected_count:
-            return
-        await asyncio.sleep(0.05)
-    assert last_count == expected_count
-
-
-@pytest.mark.asyncio
-async def test_telnet_server_forwards_client_input_to_upstream_writer():
-    upstream_reader = asyncio.StreamReader()
-    upstream_writer = DummyUpstreamWriter()
-    server = AsyncioTelnetServer(reader=upstream_reader, writer=upstream_writer, binary=True, echo=True)
-    listener = await server.start("127.0.0.1", 0)
-    port = _get_listen_port(listener)
-
-    client_writer = None
-    try:
-        _, client_writer = await telnetlib3.open_connection("127.0.0.1", port, encoding=False)
-        await _wait_for_connection_count(server, 1)
-        client_writer.write(b"ping\r\n")
-        await client_writer.drain()
-
-        await asyncio.wait_for(upstream_writer.write_event.wait(), timeout=2)
-        assert b"ping\r\n" in bytes(upstream_writer.buffer)
-    finally:
-        if client_writer is not None:
-            client_writer.close()
-        listener.close()
-        await listener.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_telnet_server_broadcasts_upstream_output_to_all_clients():
-    upstream_reader = asyncio.StreamReader()
-    upstream_writer = DummyUpstreamWriter()
-    server = AsyncioTelnetServer(reader=upstream_reader, writer=upstream_writer, binary=True, echo=True)
-    listener = await server.start("127.0.0.1", 0)
-    port = _get_listen_port(listener)
-
-    client1_reader = client1_writer = None
-    client2_reader = client2_writer = None
-
-    try:
-        client1_reader, client1_writer = await telnetlib3.open_connection("127.0.0.1", port, encoding=False)
-        client2_reader, client2_writer = await telnetlib3.open_connection("127.0.0.1", port, encoding=False)
-        await _wait_for_connection_count(server, 2)
-
-        upstream_reader.feed_data(b"hello")
-
-        data1 = await asyncio.wait_for(client1_reader.read(5), timeout=2)
-        data2 = await asyncio.wait_for(client2_reader.read(5), timeout=2)
-
-        assert data1 == b"hello"
-        assert data2 == b"hello"
-    finally:
-        if client1_writer is not None:
-            client1_writer.close()
-        if client2_writer is not None:
-            client2_writer.close()
-        listener.close()
-        await listener.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_telnet_server_naws_callback_uses_negotiated_dimensions():
-    callback_event = asyncio.Event()
-    callback_result = {}
-
-    async def window_size_changed(columns, rows):
-        callback_result["columns"] = columns
-        callback_result["rows"] = rows
-        callback_event.set()
-
-    server = AsyncioTelnetServer(
-        binary=True,
-        echo=True,
-        naws=True,
-        window_size_changed_callback=window_size_changed,
-    )
-    listener = await server.start("127.0.0.1", 0)
-    port = _get_listen_port(listener)
-
-    client_writer = None
-    try:
-        _, client_writer = await telnetlib3.open_connection(
-            "127.0.0.1", port, encoding=False, cols=132, rows=44
-        )
-
-        await asyncio.wait_for(callback_event.wait(), timeout=2)
-        assert callback_result == {"columns": 132, "rows": 44}
-    finally:
-        if client_writer is not None:
-            client_writer.close()
-        listener.close()
-        await listener.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_telnet_server_broadcast_survives_client_disconnect():
-    upstream_reader = asyncio.StreamReader()
-    upstream_writer = DummyUpstreamWriter()
-    server = AsyncioTelnetServer(reader=upstream_reader, writer=upstream_writer, binary=True, echo=True)
-    listener = await server.start("127.0.0.1", 0)
-    port = _get_listen_port(listener)
-
-    client1_writer = None
-    client2_reader = client2_writer = None
-
-    try:
-        _, client1_writer = await telnetlib3.open_connection("127.0.0.1", port, encoding=False)
-        client2_reader, client2_writer = await telnetlib3.open_connection("127.0.0.1", port, encoding=False)
-        await _wait_for_connection_count(server, 2)
-
-        client1_writer.close()
-        await _wait_for_connection_count(server, 1)
-
-        upstream_reader.feed_data(b"ok")
-        data2 = await asyncio.wait_for(client2_reader.read(2), timeout=2)
-
-        assert data2 == b"ok"
-        async with server._connections_lock:
-            assert len(server._connections) <= 1
-    finally:
-        if client1_writer is not None:
-            with contextlib.suppress(Exception):
-                client1_writer.close()
-        if client2_writer is not None:
-            with contextlib.suppress(Exception):
-                client2_writer.close()
-        listener.close()
-        await listener.wait_closed()
-
-
-@pytest.mark.asyncio
-async def test_telnet_server_listener_close_cleans_internal_tasks():
-    upstream_reader = asyncio.StreamReader()
-    upstream_writer = DummyUpstreamWriter()
-    server = AsyncioTelnetServer(reader=upstream_reader, writer=upstream_writer, binary=True, echo=True)
-    listener = await server.start("127.0.0.1", 0)
-
-    assert server._broadcast_task is not None
-
-    listener.close()
-    await listener.wait_closed()
-
-    assert server._broadcast_task is None