From 83558d4d9fda0dae88c403c2a070f135b97b4f7c Mon Sep 17 00:00:00 2001 From: Cristi Date: Tue, 24 Mar 2026 13:23:45 +0200 Subject: [PATCH 1/5] Enhancement: Re-write current Telnet server implementation using telnetlib3 library --- gns3server/compute/base_node.py | 2 +- gns3server/compute/docker/docker_vm.py | 8 +- gns3server/compute/iou/iou_vm.py | 4 +- .../compute/virtualbox/virtualbox_vm.py | 4 +- gns3server/compute/vmware/vmware_vm.py | 4 +- gns3server/utils/asyncio/embed_shell.py | 4 +- gns3server/utils/asyncio/telnet_server.py | 587 ++++++++---------- requirements.txt | 1 + tests/utils/asyncio/test_telnet_server.py | 177 ++++++ 9 files changed, 454 insertions(+), 337 deletions(-) create mode 100644 tests/utils/asyncio/test_telnet_server.py diff --git a/gns3server/compute/base_node.py b/gns3server/compute/base_node.py index e09111de4..d67b96bef 100644 --- a/gns3server/compute/base_node.py +++ b/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): diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py index b78773203..5586cfb70 100644 --- a/gns3server/compute/docker/docker_vm.py +++ b/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( - await asyncio.start_server(server.run, self._manager.port_manager.console_host, self.aux) - ) + self._telnet_servers.append(await server.start(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( - await asyncio.start_server(telnet.run, self._manager.port_manager.console_host, self.console) - ) + self._telnet_servers.append(await telnet.start(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}" diff --git a/gns3server/compute/iou/iou_vm.py b/gns3server/compute/iou/iou_vm.py index 2322e7723..32dc01ef2 100644 --- a/gns3server/compute/iou/iou_vm.py +++ b/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( - server.run, self._manager.port_manager.console_host, self.console - ) + self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console) except OSError as e: await self.stop() raise IOUError( diff --git a/gns3server/compute/virtualbox/virtualbox_vm.py b/gns3server/compute/virtualbox/virtualbox_vm.py index 5f4cd2923..1cda3e180 100644 --- a/gns3server/compute/virtualbox/virtualbox_vm.py +++ b/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( - server.run, self._manager.port_manager.console_host, self.console - ) + self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console) except OSError as e: self.project.emit( "log.warning", diff --git a/gns3server/compute/vmware/vmware_vm.py b/gns3server/compute/vmware/vmware_vm.py index 22e766b63..6d39fd8c5 100644 --- a/gns3server/compute/vmware/vmware_vm.py +++ b/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( - server.run, self._manager.port_manager.console_host, self.console - ) + self._telnet_server = await server.start(self._manager.port_manager.console_host, self.console) except OSError as e: self.project.emit( "log.warning", diff --git a/gns3server/utils/asyncio/embed_shell.py b/gns3server/utils/asyncio/embed_shell.py index 930551cee..907ace2ff 100644 --- a/gns3server/utils/asyncio/embed_shell.py +++ b/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) - s = loop.run_until_complete(coro) + loop.run_until_complete(server.start("127.0.0.1", 4444)) try: loop.run_forever() except KeyboardInterrupt: pass + loop.run_until_complete(server.close()) # Demo using stdin # loop.run_until_complete(create_stdin_shell(Demo())) diff --git a/gns3server/utils/asyncio/telnet_server.py b/gns3server/utils/asyncio/telnet_server.py index ca6c1da35..8f571d1f1 100644 --- a/gns3server/utils/asyncio/telnet_server.py +++ b/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 . -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 - :return: - """ + """Handles incoming data.""" def send(self, data): - """ - Sending data back to client - :return: - """ + """Send data back to client.""" + data = data.decode().replace("\n", "\r\n") self.writer.write(data.encode()) def close(self): - """ - Closes current connection - :return: - """ + """Close current connection.""" + 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 - :param naws when True make a window size negotiation - :param connection_factory: when set it's possible to inject own implementation of connection + Initialize telnet server. + + :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): - # Send initial telnet session opening - if echo: - writer.write(bytes([IAC, WILL, ECHO])) - else: - writer.write(bytes([IAC, WONT, ECHO, IAC, DONT, ECHO])) + async def start(self, host, port): + """Start a telnetlib3-backed listener and return a managed server handle.""" - if binary: - 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])) + if self._server is not None: + raise RuntimeError("AsyncioTelnetServer is already started") - if naws: - writer.write(bytes([IAC, DO, NAWS])) - await writer.drain() + protocol_factory = self._build_protocol_factory() + self._server = await telnetlib3.create_server( + 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") - 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 - # Will keep a firewall from aging out telnet console. + """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._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) - s = loop.run_until_complete(coro) + loop.run_until_complete(server.start("127.0.0.1", 4444)) try: loop.run_forever() except KeyboardInterrupt: pass - # Close the server - s.close() - loop.run_until_complete(s.wait_closed()) + + loop.run_until_complete(server.close()) loop.close() diff --git a/requirements.txt b/requirements.txt index 4a32f8eda..c7e7def98 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/utils/asyncio/test_telnet_server.py b/tests/utils/asyncio/test_telnet_server.py new file mode 100644 index 000000000..dcd1c1efe --- /dev/null +++ b/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 From 7f3468dcd3f9a86795225c21ca587f6fb92a5379 Mon Sep 17 00:00:00 2001 From: Cristi Date: Tue, 24 Mar 2026 15:23:00 +0200 Subject: [PATCH 2/5] Improved exception handling and telnet client disconnect --- gns3server/utils/asyncio/telnet_server.py | 4 ++++ tests/utils/asyncio/test_telnet_server.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/gns3server/utils/asyncio/telnet_server.py b/gns3server/utils/asyncio/telnet_server.py index 8f571d1f1..a6b54d706 100644 --- a/gns3server/utils/asyncio/telnet_server.py +++ b/gns3server/utils/asyncio/telnet_server.py @@ -251,8 +251,12 @@ class AsyncioTelnetServer: 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) diff --git a/tests/utils/asyncio/test_telnet_server.py b/tests/utils/asyncio/test_telnet_server.py index dcd1c1efe..d91fb91c9 100644 --- a/tests/utils/asyncio/test_telnet_server.py +++ b/tests/utils/asyncio/test_telnet_server.py @@ -143,7 +143,7 @@ async def test_telnet_server_broadcast_survives_client_disconnect(): await _wait_for_connection_count(server, 2) client1_writer.close() - await asyncio.sleep(0.2) + await _wait_for_connection_count(server, 1) upstream_reader.feed_data(b"ok") data2 = await asyncio.wait_for(client2_reader.read(2), timeout=2) From 9a06830f88f9e8516b7bb8ee606bfda93a4b5600 Mon Sep 17 00:00:00 2001 From: grossmj Date: Tue, 31 Mar 2026 17:22:34 +0800 Subject: [PATCH 3/5] telnetlib3 is a shared dependency with AI Copilot --- ai-requirements.txt | 3 --- requirements.txt | 4 ++-- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/ai-requirements.txt b/ai-requirements.txt index 9c3c1f189..12254ae42 100644 --- a/ai-requirements.txt +++ b/ai-requirements.txt @@ -36,9 +36,6 @@ nornir-netmiko>=1.0.1 nornir-utils>=0.2.0 nornir-salt>=0.23.0 -# Telnet Client -telnetlib3>=2.0.8 - # Environment & Configuration python-dotenv>=1.2.1 diff --git a/requirements.txt b/requirements.txt index 3024c7eb2..e3f7e15df 100644 --- a/requirements.txt +++ b/requirements.txt @@ -23,11 +23,11 @@ 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) +telnetlib3==4.0.1 typing-extensions>=4.15.0 -requests>=2.32.5 +requests>=2.33.1 urllib3>=2.6.2 # ============================================================================== From 38ebaaa90a5e5bc1b9e7747a6c7af73538b13e63 Mon Sep 17 00:00:00 2001 From: grossmj Date: Tue, 31 Mar 2026 17:30:41 +0800 Subject: [PATCH 4/5] Remove requests dependency from dev-requirements.txt --- dev-requirements.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/dev-requirements.txt b/dev-requirements.txt index 9b6012aaa..476c7d108 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -3,6 +3,5 @@ flake8==7.3.0 pytest-timeout==2.4.0 pytest-asyncio==1.2.0; python_version == '3.9' # version 1.2.0 is the last one supporting Python 3.9 pytest-asyncio==1.3.0; python_version >= '3.10' -requests==2.32.5 httpx==0.28.1 httpx_ws==0.7.2 # upgrading leads to failures in tests \ No newline at end of file From 0b6f17cf8e69a732b4bc767206f3ec5ebf0ee01d Mon Sep 17 00:00:00 2001 From: grossmj Date: Tue, 31 Mar 2026 18:30:59 +0800 Subject: [PATCH 5/5] Add keepalive mechanism using the NOP command --- gns3server/utils/asyncio/telnet_server.py | 40 ++++++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/gns3server/utils/asyncio/telnet_server.py b/gns3server/utils/asyncio/telnet_server.py index a6b54d706..cf4a61870 100644 --- a/gns3server/utils/asyncio/telnet_server.py +++ b/gns3server/utils/asyncio/telnet_server.py @@ -22,12 +22,13 @@ import socket import telnetlib3 from telnetlib3.server import TelnetServer -from telnetlib3.telopt import DONT, ECHO, IAC, NAWS, WILL, WONT +from telnetlib3.telopt import DONT, ECHO, IAC, NAWS, NOP, WILL, WONT log = logging.getLogger(__name__) READ_SIZE = 1024 BROADCAST_DRAIN_TIMEOUT = 10 +KEEPALIVE_INTERVAL = 60 # Send NOP every 60 seconds class _ManagedTelnetListener: @@ -108,12 +109,14 @@ class AsyncioTelnetServer: 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. :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). """ assert connection_factory is None or ( @@ -126,12 +129,14 @@ class AsyncioTelnetServer: self._binary = binary 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 @@ -177,6 +182,9 @@ class AsyncioTelnetServer: 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): @@ -186,6 +194,13 @@ class AsyncioTelnetServer: 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 @@ -325,6 +340,29 @@ class AsyncioTelnetServer: 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())