server

Telnet server implementation with command-line interface.

The main function here is wired to the command line tool by name telnetlib3-server. If this server’s PID receives the SIGTERM signal, it attempts to shutdown gracefully.

The TelnetServer class negotiates a character-at-a-time (WILL-SGA, WILL-ECHO) session with support for negotiation about window size, environment variables, terminal type name, and to automatically close connections clients after an idle period.

class TelnetServer(term='unknown', cols=80, rows=25, timeout=300, shell=None, _waiter_connected=None, encoding='utf8', encoding_errors='strict', force_binary=False, never_send_ga=False, line_mode=False, connect_maxwait=4.0, compression=None, limit=None, reader_factory=<class 'telnetlib3.stream_reader.TelnetReader'>, reader_factory_encoding=<class 'telnetlib3.stream_reader.TelnetReaderUnicode'>, writer_factory=<class 'telnetlib3.stream_writer.TelnetWriter'>, writer_factory_encoding=<class 'telnetlib3.stream_writer.TelnetWriterUnicode'>)

Telnet Server protocol performing common negotiation.

Initialize TelnetServer with terminal parameters.

TTYPE_LOOPMAX = 8

Maximum number of cycles to seek for all terminal types. We are seeking the repeat or cycle of a terminal table, choosing the first – but when negotiated by MUD clients, we chose the must Unix TERM appropriate,

connection_made(transport)

Handle new connection and wire up telnet option callbacks.

Return type:

None

data_received(data)

Process received data and reset timeout timer.

Return type:

None

begin_negotiation()

Begin telnet negotiation by requesting terminal type.

Return type:

None

begin_advanced_negotiation()

Request advanced telnet options from client.

DO NEW_ENVIRON is deferred until the TTYPE cycle completes so that Microsoft telnet (ANSI + VT100) can be detected first. See _negotiate_environ() and GitHub issue #24.

WILL ECHO is deferred until TTYPE reveals the client identity. MUD clients (Mudlet, TinTin++, etc.) interpret WILL ECHO as “password mode” and mask input. See _negotiate_echo().

Return type:

None

check_negotiation(final=False)

Check if negotiation is complete including encoding.

Return type:

bool

encoding(outgoing=None, incoming=None)

Return encoding for the given stream direction.

Parameters:

• outgoing (Optional[bool]) – Whether the return value is suitable for encoding bytes for transmission to client end.

• incoming (Optional[bool]) – Whether the return value is suitable for decoding bytes received from the client.

Raises:

TypeError – when a direction argument, either outgoing or incoming, was not set True.

Return type:

Union[str, bool]

Returns:

'US-ASCII' for the directions indicated, unless BINARY RFC 856 has been negotiated for the direction indicated or force_binary is set True.

set_timeout(duration=-1)

Restart or unset timeout for client.

Parameters:

duration (int) – When specified as a positive integer, schedules Future for callback of on_timeout(). When -1, the value of self.get_extra_info('timeout') is used. When non-True, it is canceled.

Return type:

None

on_timeout()

Callback received on session timeout.

Default implementation writes “Timeout.” bound by CRLF and closes.

This can be disabled by calling set_timeout() with duration value of 0.

Return type:

None

on_naws(rows, cols)

Callback receives NAWS response, RFC 1073.

Parameters:

• rows (int) – screen size, by number of cells in height.

• cols (int) – screen size, by number of cells in width.

Return type:

None

on_request_environ()

Definition for NEW_ENVIRON request of client, RFC 1572.

This method is a callback from request_environ(), first entered on receipt of (WILL, NEW_ENVIRON) by server. The return value defines the request made to the client for environment values.

Return type:

List[Union[str, bytes]]

Returns:

A list of US-ASCII character strings indicating the environment keys the server requests of the client. If this list contains the special byte constants, USERVAR or VAR, the client is allowed to volunteer any other additional user or system values. An empty return value indicates that no request should be made.

The default return value requests only common variables needed for session setup. Override this method or see ENVIRON_EXTENDED for a larger set used during client fingerprinting.

Note

USER is excluded when the client is Microsoft telnet (ttype1=ANSI, ttype2=VT100) because requesting it crashes telnet.exe. See GitHub issue #24.

on_environ(mapping)

Callback receives NEW_ENVIRON response, RFC 1572.

Return type:

None

on_request_charset()

Definition for CHARSET request by client, RFC 2066.

This method is a callback from request_charset(), first entered on receipt of (WILL, CHARSET) by server. The return value defines the request made to the client for encodings.

Return type:

List[str]

Returns:

A list of US-ASCII character strings indicating the encodings offered by the server in its preferred order. An empty return value indicates that no encodings are offered.

The default return value includes common encodings for both Western and Eastern scripts:

['UTF-8', 'UTF-16', 'LATIN1', 'US-ASCII', 'CP1252', 'ISO-8859-15', 'CP437',
 'SHIFT_JIS', 'CP932', 'BIG5', 'CP950', 'GBK', 'GB2312', 'CP936', 'EUC-KR', 'CP949']

on_charset(charset)

Callback for CHARSET response, RFC 2066.

Return type:

None

on_tspeed(rx, tx)

Callback for TSPEED response, RFC 1079.

Return type:

None

on_ttype(ttype)

Callback for TTYPE response, RFC 930.

Return type:

None

on_xdisploc(xdisploc)

Callback for XDISPLOC response, RFC 1096.

Return type:

None

class LinemodeServer(term='unknown', cols=80, rows=25, timeout=300, shell=None, _waiter_connected=None, encoding='utf8', encoding_errors='strict', force_binary=False, never_send_ga=False, line_mode=False, connect_maxwait=4.0, compression=None, limit=None, reader_factory=<class 'telnetlib3.stream_reader.TelnetReader'>, reader_factory_encoding=<class 'telnetlib3.stream_reader.TelnetReaderUnicode'>, writer_factory=<class 'telnetlib3.stream_writer.TelnetWriter'>, writer_factory_encoding=<class 'telnetlib3.stream_writer.TelnetWriterUnicode'>)

TelnetServer subclass that negotiates LINEMODE EDIT.

In addition to the standard options negotiated by TelnetServer, this server sends DO LINEMODE during advanced negotiation, proposes LINEMODE EDIT (local line editing by the client), and suppresses WILL ECHO so the client performs local echoing via its LINEMODE buffer.

Use with create_server() to enable RFC 1184 LINEMODE EDIT on a telnet_server_shell() session or any custom shell.

Initialize TelnetServer with terminal parameters.

default_linemode = <b'\x01': lit_echo:False, soft_tab:False, ack:False, trapsig:False, remote:False, local:True>

Propose LINEMODE EDIT (local line editing) instead of remote mode.

begin_advanced_negotiation()

Negotiate standard options plus DO LINEMODE.

Return type:

None

class Server(server)

Telnet server that tracks connected clients.

Wraps asyncio.Server with protocol tracking and connection waiting. Returned by create_server().

Initialize wrapper around asyncio.Server.

close()

Close the server, stop accepting new connections, and close all clients.

Return type:

None

async wait_closed()

Wait until the server and all client connections are closed.

Return type:

None

property sockets: Tuple[socket, ...] | None

Return list of socket objects the server is listening on.

is_serving()

Return True if the server is accepting new connections.

Return type:

bool

property clients: List[BaseServer]

List of connected client protocol instances.

Returns:

List of protocol instances for all connected clients.

async wait_for_client()

Wait for a client to connect and complete negotiation.

Return type:

BaseServer

Returns:

The protocol instance for the connected client.

Example:

server = await telnetlib3.create_server(port=6023)
client = await server.wait_for_client()
client.writer.write("Welcome!\r\n")

async create_server(host=None, port=23, protocol_factory=<class 'telnetlib3.server.TelnetServer'>, shell=None, encoding='utf8', encoding_errors='strict', force_binary=False, never_send_ga=False, line_mode=False, connect_maxwait=4.0, compression=None, limit=None, term='unknown', cols=80, rows=25, timeout=300, ssl=None, tls_auto=False)

Create a TCP Telnet server.

Parameters:

• host (Union[str, Sequence[str], None]) – The host parameter can be a string, in that case the TCP server is bound to host and port. The host parameter can also be a sequence of strings, and in that case the TCP server is bound to all hosts of the sequence.

• port (int) – Listen port for TCP server.

• protocol_factory (Optional[Type[Protocol]]) – An alternate protocol factory for the server. When unspecified, TelnetServer is used.

• shell (Optional[Callable[[Union[TelnetReader, TelnetReaderUnicode], Union[TelnetWriter, TelnetWriterUnicode]], Coroutine[Any, Any, None]]]) – An async function that is called after negotiation completes, receiving arguments (reader, writer). Default is telnet_server_shell(). The reader is a TelnetReader instance, the writer is a TelnetWriter instance.

• encoding (Union[str, bool]) –

  The default assumed encoding, or False to disable unicode support. Encoding may be negotiated to another value by the client through NEW_ENVIRON RFC 1572 by sending environment value of LANG, or by any legal value for CHARSET RFC 2066 negotiation.

  The server’s attached reader, writer streams accept and return unicode, or natural strings, “hello world”, unless this value is explicitly set to False. In that case, the attached stream interfaces are bytes-only, b”hello world”.

• encoding_errors (str) – Same meaning as codecs.Codec.encode(). Default value is strict.

• force_binary (bool) –

  When True, the encoding specified is used for both directions even when BINARY mode, RFC 856, is not negotiated for the direction specified. This parameter has no effect when encoding=False.

  Note that when combined with a default encoding, use of this option may prematurely cause data transmitted in the default encoding immediately on-connect, before a “smart” telnet client or server can negotiate a different one.

  In most cases, so long as the initial login banner/etc is US-ASCII, this may be no problem at all. If an encoding is assumed, as in many MUD and BBS systems, the combination of force_binary with a default encoding is often preferred.

• line_mode (bool) – When True, the server does not send WILL SGA or WILL ECHO during negotiation. This keeps the client in NVT local (line) mode, where the client performs its own line editing and sends complete lines. Default is False (kludge mode).

• term (str) – Value returned for writer.get_extra_info('term') until negotiated by TTYPE RFC 930, or NAWS RFC 1572. Default value is 'unknown'.

• cols (int) – Value returned for writer.get_extra_info('cols') until negotiated by NAWS RFC 1572. Default value is 80 columns.

• rows (int) – Value returned for writer.get_extra_info('rows') until negotiated by NAWS RFC 1572. Default value is 25 rows.

• timeout (int) – Causes clients to disconnect if idle for this duration, in seconds. This ensures resources are freed on busy servers. When explicitly set to False, clients will not be disconnected for timeout. Default value is 300 seconds (5 minutes).

• connect_maxwait (float) – If the remote end is not compliant, or otherwise confused by our demands, the shell continues anyway after the greater of this value has elapsed. A client that is not answering option negotiation will delay the start of the shell by this amount.

• compression (Optional[bool]) – MCCP compression policy. None (default) passively accepts compression if requested by the client. True advertises MCCP2/MCCP3 during advanced negotiation. False rejects all compression offers.

• limit (Optional[int]) – The buffer limit for the reader stream.

• ssl (Optional[SSLContext]) – An ssl.SSLContext for TLS-encrypted connections (TELNETS, RFC 855 over TLS). When provided, the server performs a TLS handshake before any telnet data is exchanged. None (default) creates a plain TCP server.

• tls_auto (Union[bool, float]) – When truthy and ssl is provided, the server accepts both TLS and plain telnet clients on the same port. A float value sets the number of seconds to wait for a TLS ClientHello (0x16) before assuming a plain telnet connection; True uses a default of 0.5 seconds. TLS clients send ClientHello immediately; plain telnet clients typically wait for the server to speak first, so the timeout distinguishes the two. False or 0 (default) disables auto-detection. Requires ssl to be an ssl.SSLContext.

Return type:

Server

Returns:

A Server instance that wraps the asyncio.Server and provides access to connected client protocols via Server.wait_for_client() and Server.clients.

async run_server(host='localhost', port=6023, loglevel='info', logfile=None, logfmt='%(levelname)s %(filename)s:%(lineno)d %(message)s', shell=<function telnet_server_shell>, encoding='utf8', force_binary=False, timeout=300, connect_maxwait=1.5, pty_exec=None, pty_args=None, pty_raw=True, robot_check=False, pty_fork_limit=0, status_interval=20, never_send_ga=False, line_mode=False, compression=None, protocol_factory=None, ssl=None, tls_auto=False)

Program entry point for server daemon.

This function configures a logger and creates a telnet server for the given keyword arguments, serving forever, completing only upon receipt of SIGTERM.

Return type:

None

parse_server_args(extra_args_fn=None)

Parse command-line arguments for telnet server.

Parameters:

extra_args_fn (Optional[Callable[[ArgumentParser], None]]) – Optional callback to add extra arguments to the parser before parsing. Used by telnetlib3-fingerprint-server to inject --data-dir.

Return type:

Dict[str, Any]