Enhancement: Re-write current Telnet server implementation using telnetlib3 library

gns3server/compute/base_node.py

@@ -436,7 +436,7 @@ class BaseNode:
             echo=True
         )
         # warning: this will raise OSError exception if there is a problem...
-        telnet_server = await asyncio.start_server(server.run, self._manager.port_manager.console_host, external_port)
+        telnet_server = await server.start(self._manager.port_manager.console_host, external_port)
         self._wrapper_telnet_servers.append(telnet_server)
 
     async def start_wrap_console(self):

gns3server/compute/docker/docker_vm.py

@@ -704,9 +704,7 @@ class DockerVM(BaseNode):
             raise DockerError(f"Could not start auxiliary console process: {e}")
         server = AsyncioTelnetServer(reader=process.stdout, writer=process.stdin, binary=True, echo=True)
         try:
-            self._telnet_servers.append(
+            self._telnet_servers.append(await server.start(self._manager.port_manager.console_host, self.aux))
-                await asyncio.start_server(server.run, self._manager.port_manager.console_host, self.aux)
-            )
         except OSError as e:
             raise DockerError(
                 f"Could not start Telnet server on socket {self._manager.port_manager.console_host}:{self.aux}: {e}"
@@ -899,9 +897,7 @@ class DockerVM(BaseNode):
             window_size_changed_callback=self._window_size_changed_callback,
         )
         try:
-            self._telnet_servers.append(
+            self._telnet_servers.append(await telnet.start(self._manager.port_manager.console_host, self.console))
-                await asyncio.start_server(telnet.run, self._manager.port_manager.console_host, self.console)
-            )
         except OSError as e:
             raise DockerError(
                 f"Could not start Telnet server on socket {self._manager.port_manager.console_host}:{self.console}: {e}"

gns3server/compute/iou/iou_vm.py

@@ -651,9 +651,7 @@ class IOUVM(BaseNode):
                 reader=self._iou_process.stdout, writer=self._iou_process.stdin, binary=True, echo=True
             )
             try:
-                self._telnet_server = await asyncio.start_server(
+                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
-                    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

@@ -1019,9 +1019,7 @@ class VirtualBoxVM(BaseNode):
                 raise VirtualBoxError(f"Could not open serial pipe '{pipe_name}': {e}")
             server = AsyncioTelnetServer(reader=self._remote_pipe, writer=self._remote_pipe, binary=True, echo=True)
             try:
-                self._telnet_server = await asyncio.start_server(
+                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
-                    server.run, self._manager.port_manager.console_host, self.console
-                )
             except OSError as e:
                 self.project.emit(
                     "log.warning",

gns3server/compute/vmware/vmware_vm.py

@@ -920,9 +920,7 @@ class VMwareVM(BaseNode):
                 raise VMwareError(f"Could not open serial pipe '{pipe_name}': {e}")
             server = AsyncioTelnetServer(reader=self._remote_pipe, writer=self._remote_pipe, binary=True, echo=True)
             try:
-                self._telnet_server = await asyncio.start_server(
+                self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console)
-                    server.run, self._manager.port_manager.console_host, self.console
-                )
             except OSError as e:
                 self.project.emit(
                     "log.warning",

gns3server/utils/asyncio/embed_shell.py

@@ -357,12 +357,12 @@ if __name__ == "__main__":
     # Demo using telnet
     shell = Demo(welcome_message="Welcome!\n")
     server = create_telnet_shell(shell, loop=loop)
-    coro = asyncio.start_server(server.run, "127.0.0.1", 4444)
+    loop.run_until_complete(server.start("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

@@ -14,49 +14,41 @@
 # 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 struct
+import contextlib
-
 import logging
+import socket
+
+import telnetlib3
+from telnetlib3.server import TelnetServer
+from telnetlib3.telopt import DONT, ECHO, IAC, NAWS, 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
+
+
+class _ManagedTelnetListener:
+    """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:
@@ -86,36 +78,27 @@ class TelnetConnection:
 
     async def window_size_changed(self, columns, rows):
         """Method called when window size changed, only can occur when
-        `naws` flag is enable in server configuration."""
+        `naws` flag is enabled 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:
-    MAX_NEGOTIATION_READ = 10
-
     def __init__(
         self,
         reader=None,
@@ -127,29 +110,31 @@ class AsyncioTelnetServer:
         connection_factory=None,
     ):
         """
-        Initializes telnet server
+        Initialize telnet server.
-        :param naws when True make a window size negotiation
+
-        :param connection_factory: when set it's possible to inject own implementation of connection
+        :param naws: when True, window size negotiation callbacks are enabled.
+        :param connection_factory: optional factory to inject a custom connection implementation.
         """
+
         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._connections = {}
+        self._pending_window_sizes = {}
+        self._connections_lock = asyncio.Lock()
+        self._close_lock = asyncio.Lock()
+        self._broadcast_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)
 
@@ -160,277 +145,243 @@ class AsyncioTelnetServer:
 
     @staticmethod
     async def write_client_intro(writer, echo=False):
-        # Send initial telnet session opening
+        """Write a minimal telnet intro to an upstream console endpoint."""
+
         if echo:
-            writer.write(bytes([IAC, WILL, ECHO]))
+            writer.write(IAC + WILL + ECHO)
         else:
-            writer.write(bytes([IAC, WONT, ECHO, IAC, DONT, ECHO]))
+            writer.write(IAC + WONT + ECHO + IAC + DONT + ECHO)
         await writer.drain()
 
-    async def _write_intro(self, writer, binary=False, echo=False, naws=False):
+    async def start(self, host, port):
-        # Send initial telnet session opening
+        """Start a telnetlib3-backed listener and return a managed server handle."""
-        if echo:
-            writer.write(bytes([IAC, WILL, ECHO]))
-        else:
-            writer.write(bytes([IAC, WONT, ECHO, IAC, DONT, ECHO]))
 
-        if binary:
+        if self._server is not None:
-            writer.write(bytes([IAC, WILL, SGA, IAC, WILL, BINARY, IAC, DO, BINARY]))
+            raise RuntimeError("AsyncioTelnetServer is already started")
-        else:
-            writer.write(bytes([IAC, WONT, SGA, IAC, DONT, SGA, IAC, WONT, BINARY, IAC, DONT, BINARY]))
 
-        if naws:
+        protocol_factory = self._build_protocol_factory()
-            writer.write(bytes([IAC, DO, NAWS]))
+        self._server = await telnetlib3.create_server(
-        await writer.drain()
+            host=host,
+            port=port,
+            protocol_factory=protocol_factory,
+            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())
+
+        return self._server_handle
 
     async def run(self, network_reader, network_writer):
-        sock = network_writer.get_extra_info("socket")
+        """Backward-compatible entrypoint for asyncio.start_server(server.run, ...)."""
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+
-        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        await self._run_client_session(network_reader, network_writer)
-        # 60 sec keep alives, close tcp session after 4 missed
+
-        # Will keep a firewall from aging out telnet console.
+    async def close(self):
+        async with self._close_lock:
+            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 (ConnectionError, OSError):
+            connection.close()
+
+    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 _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:
+            return
+
+        with contextlib.suppress(OSError):
+            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("module 'socket' has no attribute 'TCP_KEEPIDLE' or 'TCP_KEEPALIVE'")
+                raise AttributeError("No TCP keepalive idle socket option is available")
+
             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()))
-
-        # 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()
-            # Use pop() to avoid KeyError if connection was already removed
-            self._connections.pop(network_writer, None)
-
-    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()
-                            self._connections.pop(connection_key, None)
-
-    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("!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: " "{:#x} {:#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(f"Telnet WON'T {iac_cmd[2]:#x}")
-                    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(f"Telnet WON'T {iac_cmd[2]:#x}")
-
-                elif iac_cmd[1] == DONT:
-                    log.debug("Unhandled DONT telnet command: " "{:#x} {:#x} {:#x}".format(*iac_cmd))
-                elif iac_cmd[1] == WILL:
-                    if iac_cmd[2] not in [BINARY, NAWS]:
-                        log.debug("Unhandled WILL telnet command: " "{:#x} {:#x} {:#x}".format(*iac_cmd))
-                elif iac_cmd[1] == WONT:
-                    log.debug("Unhandled WONT telnet command: " "{:#x} {:#x} {:#x}".format(*iac_cmd))
-                else:
-                    log.debug("Unhandled telnet command: " "{:#x} {:#x} {:#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)
@@ -449,14 +400,12 @@ if __name__ == "__main__":
     )
     server = AsyncioTelnetServer(reader=process.stdout, writer=process.stdin, binary=False, echo=False)
 
-    coro = asyncio.start_server(server.run, "127.0.0.1", 4444)
+    loop.run_until_complete(server.start("127.0.0.1", 4444))
-    s = loop.run_until_complete(coro)
 
     try:
         loop.run_forever()
     except KeyboardInterrupt:
         pass
-    # Close the server
+
-    s.close()
+    loop.run_until_complete(server.close())
-    loop.run_until_complete(s.wait_closed())
     loop.close()

requirements.txt

@@ -23,6 +23,7 @@ watchdog==6.0.0
 zstandard==0.25.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
 
 # Shared dependencies (also used by AI Copilot)
 typing-extensions>=4.15.0

tests/utils/asyncio/test_telnet_server.py

@@ -0,0 +1,177 @@
+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 asyncio.sleep(0.2)
+
+        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