diff --git a/gns3server/compute/base_node.py b/gns3server/compute/base_node.py index 182e10f9b..020588e4f 100644 --- a/gns3server/compute/base_node.py +++ b/gns3server/compute/base_node.py @@ -397,11 +397,7 @@ class BaseNode: echo=True ) # warning: this will raise OSError exception if there is a problem... - self._wrapper_telnet_server = await asyncio.start_server( - server.run, - self._manager.port_manager.console_host, - self.console - ) + self._wrapper_telnet_server = await server.start(self._manager.port_manager.console_host, self.console) async def stop_wrap_console(self): """ diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py index 79b5f640b..edfb46ee0 100644 --- a/gns3server/compute/docker/docker_vm.py +++ b/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 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("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 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("Could not start Telnet server on socket {}:{}: {}".format(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 9089d8e68..c545472bc 100644 --- a/gns3server/compute/iou/iou_vm.py +++ b/gns3server/compute/iou/iou_vm.py @@ -598,8 +598,7 @@ class IOUVM(BaseNode): server = AsyncioTelnetServer(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 5273bfe1c..e3ec7f5c4 100644 --- a/gns3server/compute/virtualbox/virtualbox_vm.py +++ b/gns3server/compute/virtualbox/virtualbox_vm.py @@ -982,7 +982,7 @@ class VirtualBoxVM(BaseNode): 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", {"message": "Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.console, e)}) diff --git a/gns3server/compute/vmware/vmware_vm.py b/gns3server/compute/vmware/vmware_vm.py index 52ebcea3d..9f2962998 100644 --- a/gns3server/compute/vmware/vmware_vm.py +++ b/gns3server/compute/vmware/vmware_vm.py @@ -881,7 +881,7 @@ class VMwareVM(BaseNode): 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", {"message": "Could not start Telnet server on socket {}:{}: {}".format(self._manager.port_manager.console_host, self.console, e)}) diff --git a/gns3server/utils/asyncio/embed_shell.py b/gns3server/utils/asyncio/embed_shell.py index 05e2f5ea7..1d1577d02 100644 --- a/gns3server/utils/asyncio/embed_shell.py +++ b/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) - 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 cc34a29a6..cf4a61870 100644 --- a/gns3server/utils/asyncio/telnet_server.py +++ b/gns3server/utils/asyncio/telnet_server.py @@ -1,4 +1,3 @@ -# -*- coding: utf-8 -*- # # Copyright (C) 2014 GNS3 Technologies Inc. # @@ -15,52 +14,47 @@ # 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, 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 TelnetConnection(object): +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: """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 @@ -85,59 +79,66 @@ class TelnetConnection(object): 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, + writer=None, + binary=True, + echo=False, + naws=False, + window_size_changed_callback=None, + connection_factory=None, + keepalive_interval=KEEPALIVE_INTERVAL, + ): + """ + Initialize telnet server. - def __init__(self, reader=None, writer=None, binary=True, echo=False, naws=False, window_size_changed_callback=None, connection_factory=None): + :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). """ - 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 - """ - 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." + + 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) @@ -149,317 +150,304 @@ 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()) + + 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): - 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._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: + 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() - 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__': +if __name__ == "__main__": logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() - process = loop.run_until_complete(asyncio.ensure_future(asyncio.subprocess.create_subprocess_exec("/bin/sh", "-i", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.STDOUT, - stdin=asyncio.subprocess.PIPE))) + process = loop.run_until_complete( + asyncio.ensure_future( + asyncio.subprocess.create_subprocess_exec( + "/bin/sh", + "-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) - 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 9c322ec53..e337e4723 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,3 +10,4 @@ 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 diff --git a/tests/utils/asyncio/test_telnet_server.py b/tests/utils/asyncio/test_telnet_server.py new file mode 100644 index 000000000..d91fb91c9 --- /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 _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