sync

Synchronous (blocking) interface for telnetlib3.

This module provides a non-asyncio interface that wraps the async telnetlib3 implementation. The asyncio event loop runs in a background thread, and blocking methods wait on thread-safe futures.

Example client usage:

from telnetlib3.sync import TelnetConnection

with TelnetConnection('localhost', 6023) as conn:
    conn.write('hello\r\n')
    print(conn.readline())

Example server usage:

from telnetlib3.sync import BlockingTelnetServer
import threading

def handler(conn):
    conn.write('Hello!\r\n')
    while line := conn.readline():
        conn.write(f'Echo: {line}')

server = BlockingTelnetServer('localhost', 6023, handler=handler)
server.serve_forever()

class TelnetConnection(host, port=23, timeout=None, encoding='utf8', **kwargs)[source]

Bases: object

Blocking telnet client connection.

Wraps async telnetlib3.open_connection() with blocking methods. The asyncio event loop runs in a daemon thread.

Parameters:

host (str) – Remote server hostname or IP address.
port (int) – Remote server port (default 23).
timeout (Optional[float]) – Default timeout for operations in seconds.
encoding (str) – Character encoding (default ‘utf8’).
connect_timeout – Timeout in seconds for the TCP connection to be established. Passed to telnetlib3.open_connection().
kwargs (Any) – Additional arguments passed to telnetlib3.open_connection().

Example:

with TelnetConnection('localhost', 6023) as conn:
    conn.write('hello\r\n')
    response = conn.readline()

Initialize connection parameters without connecting.

connect()[source]

Establish connection to the server.

Blocks until connected or timeout expires.

Raises:

RuntimeError – If already connected.
TimeoutError – If connection times out.
ConnectionError – If connection fails.
Exception – If connection fails for other reasons.

Return type:

None

read(n=-1, timeout=None)[source]

Read up to n bytes/characters from the connection.

Blocks until data is available or timeout expires.

Parameters:

n (int) – Maximum bytes to read (-1 for any available data).
timeout (Optional[float]) – Timeout in seconds (uses default if None).

Return type:

Union[str, bytes]

Returns:

Data read from connection.

Raises:

TimeoutError – If timeout expires before data available.
EOFError – If connection closed.

read_some(timeout=None)[source]

Read some available data from the connection.

Unlike read() with n=-1, this returns as soon as any data is available rather than waiting for EOF.

Parameters:: timeout (Optional[float]) – Timeout in seconds (uses default if None).
Return type:: Union[str, bytes]
Returns:: Data read from connection.

readline(timeout=None)[source]

Read one line from the connection.

Blocks until a complete line is received or timeout expires.

Parameters:

timeout (Optional[float]) – Timeout in seconds (uses default if None).

Return type:

Union[str, bytes]

Returns:

Line including terminator.

Raises:

TimeoutError – If timeout expires.
EOFError – If connection closed before line complete.

read_until(match, timeout=None)[source]

Read until match is found.

Like old telnetlib’s read_until method.

Parameters:

match (Union[str, bytes]) – String or bytes to match.
timeout (Optional[float]) – Timeout in seconds (uses default if None).

Return type:

Union[str, bytes]

Returns:

Data up to and including match.

Raises:

TimeoutError – If timeout expires before match found.
EOFError – If connection closed before match found.

write(data)[source]

Write data to the connection.

This method buffers data and returns immediately. Use flush() to ensure data is sent.

Parameters:: data (Union[str, bytes]) – String or bytes to write.
Return type:: None

flush(timeout=None)[source]

Flush buffered data to the connection.

Blocks until all buffered data has been sent.

Parameters:: timeout (Optional[float]) – Timeout in seconds (uses default if None).
Raises:: TimeoutError – If timeout expires.
Return type:: None

close()[source]

Close the connection and stop the event loop.

Return type:: None

get_extra_info(name, default=None)[source]

Get extra information about the connection.

After negotiation completes, provides access to negotiated values:

'TERM': Terminal type (e.g., ‘xterm-256color’)
'cols': Terminal width in columns
'rows': Terminal height in rows
'peername': Remote address tuple (host, port)
'LANG': Language/locale setting

Parameters:

name (str) – Information key.
default (Any) – Default value if key not found.

Return type:

Any

Returns:

Information value or default.

wait_for(remote=None, local=None, pending=None, timeout=None)[source]

Wait for telnet option negotiation states.

This method blocks until the specified options reach their desired states, or timeout expires. This is not possible with the legacy telnetlib module.

Parameters:

remote (Optional[dict[str, bool]]) – Dict of options for remote (client WILL) state. Example: {'NAWS': True, 'TTYPE': True}
local (Optional[dict[str, bool]]) – Dict of options for local (client DO) state. Example: {'BINARY': True, 'ECHO': True}
pending (Optional[dict[str, bool]]) – Dict of options for pending negotiation state. Example: {'TTYPE': False} (wait for negotiation to complete)
timeout (Optional[float]) – Timeout in seconds (uses default if None).

Raises:

TimeoutError – If timeout expires before conditions met.

Return type:

None

Example - wait for terminal info before proceeding:

conn = TelnetConnection('localhost', 6023)
conn.connect()

# Wait for NAWS and TTYPE negotiation to complete
conn.wait_for(remote={'NAWS': True, 'TTYPE': True}, timeout=5.0)

# Now terminal info is available
term = conn.get_extra_info('TERM')
cols = conn.get_extra_info('cols')
rows = conn.get_extra_info('rows')
print(f"Terminal: {term} ({cols}x{rows})")

property writer: TelnetWriter

Access the underlying TelnetWriter for advanced operations.

This provides access to telnet protocol features not available in the legacy telnetlib:

Option state inspection (writer.remote_option, writer.local_option)
Mode detection (writer.mode - ‘local’, ‘remote’, ‘kludge’)
Protocol constants and negotiation methods

Returns:: The underlying TelnetWriter instance.

class BlockingTelnetServer(host='localhost', port=6023, handler=None, **kwargs)[source]

Bases: object

Blocking telnet server.

Wraps async telnetlib3.create_server() with a blocking interface. Each client connection can be handled in a separate thread.

Parameters:

host (str) – Address to bind to.
port (int) – Port to bind to (default 6023).
handler (Optional[Callable[[ServerConnection], None]]) – Function called for each client connection. Receives a TelnetConnection-like object as argument.
kwargs (Any) – Additional arguments passed to telnetlib3.create_server().

Example with handler:

def handle_client(conn):
    conn.write('Welcome!\r\n')
    while line := conn.readline():
        conn.write(f'Echo: {line}')

server = BlockingTelnetServer('localhost', 6023, handler=handle_client)
server.serve_forever()

Example with manual accept loop:

server = BlockingTelnetServer('localhost', 6023)
server.start()
while True:
    conn = server.accept()
    threading.Thread(target=handle_client, args=(conn,)).start()

Initialize server parameters without starting.

start()[source]

Start the server.

Non-blocking. Use accept() or serve_forever() to handle clients.

Raises:: RuntimeError – If already started.
Return type:: None

accept(timeout=None)[source]

Accept a client connection.

Blocks until a client connects.

Parameters:

timeout (Optional[float]) – Timeout in seconds (None for no timeout).

Return type:

ServerConnection

Returns:

Connection object for the client.

Raises:

TimeoutError – If timeout expires.
RuntimeError – If server not started.

serve_forever()[source]

Serve clients forever.

Blocks and handles each client in a new thread using the handler function provided at construction.

Raises:: RuntimeError – If no handler was provided.
Return type:: None

shutdown()[source]

Shutdown the server.

Stops accepting new connections and closes the server.

Return type:: None

class ServerConnection(reader, writer, loop)[source]

Bases: object

Blocking interface for a server-side client connection.

This is similar to TelnetConnection but for server-side use. Created automatically when a client connects to BlockingTelnetServer.

Provides miniboa-compatible properties for easier migration:

active - Connection state (set to False to disconnect)
address, port - Client address info
terminal_type, columns, rows - Terminal info
send() - Alias for write()
addrport() - Returns “IP:PORT” string
idle(), duration() - Timing information
deactivate() - Set active=False to queue disconnection

Initialize connection from reader/writer pair.

read(n=-1, timeout=None)[source]

Read up to n bytes/characters from the connection.

Parameters:

n (int) – Maximum bytes to read (-1 for any available data).
timeout (Optional[float]) – Timeout in seconds.

Return type:

Union[str, bytes]

Returns:

Data read from connection.

Raises:

RuntimeError – If connection already closed.
TimeoutError – If timeout expires.
EOFError – If connection closed.

read_some(timeout=None)[source]

Read some available data from the connection.

Unlike read() with n=-1, this returns as soon as any data is available rather than waiting for EOF.

Parameters:: timeout (Optional[float]) – Timeout in seconds.
Return type:: Union[str, bytes]
Returns:: Data read from connection.

readline(timeout=None)[source]

Read one line from the connection.

Parameters:

timeout (Optional[float]) – Timeout in seconds.

Return type:

Union[str, bytes]

Returns:

Line including terminator.

Raises:

RuntimeError – If connection already closed.
TimeoutError – If timeout expires.
EOFError – If connection closed.

read_until(match, timeout=None)[source]

Read until match is found.

Parameters:

match (Union[str, bytes]) – String or bytes to match.
timeout (Optional[float]) – Timeout in seconds.

Return type:

Union[str, bytes]

Returns:

Data up to and including match.

Raises:

RuntimeError – If connection already closed.
TimeoutError – If timeout expires.
EOFError – If connection closed.

write(data)[source]

Write data to the connection.

Parameters:: data (Union[str, bytes]) – String or bytes to write.
Raises:: RuntimeError – If connection already closed.
Return type:: None

flush(timeout=None)[source]

Flush buffered data to the connection.

Parameters:

timeout (Optional[float]) – Timeout in seconds.

Raises:

RuntimeError – If connection already closed.
TimeoutError – If timeout expires.

Return type:

None

close()[source]

Close the connection.

Return type:: None

get_extra_info(name, default=None)[source]

Get extra information about the connection.

After negotiation completes, provides access to negotiated values:

'TERM': Terminal type (e.g., ‘xterm-256color’)
'cols': Terminal width in columns
'rows': Terminal height in rows
'peername': Remote address tuple (host, port)

Parameters:

name (str) – Information key.
default (Any) – Default value if key not found.

Return type:

Any

Returns:

Information value or default.

wait_for(remote=None, local=None, pending=None, timeout=None)[source]

Wait for telnet option negotiation states.

Blocks until the specified options reach their desired states.

Parameters:

remote (Optional[dict[str, bool]]) – Dict of options for remote state. Example: {'NAWS': True, 'TTYPE': True}
local (Optional[dict[str, bool]]) – Dict of options for local state.
pending (Optional[dict[str, bool]]) – Dict of options for pending state.
timeout (Optional[float]) – Timeout in seconds.

Raises:

RuntimeError – If connection already closed.
TimeoutError – If timeout expires.

Return type:

None

Example:

conn = server.accept()
conn.wait_for(remote={'NAWS': True}, timeout=5.0)
print(f"Window: {conn.columns}x{conn.rows}")

property writer: TelnetWriter

Access the underlying TelnetWriter for advanced operations.

Returns:: The underlying TelnetWriter instance.

property active: bool

Connection health status (miniboa-compatible).

Set to False to disconnect on next opportunity.

property address: str: Remote IP address of the connected client (miniboa-compatible).

property port: int: Remote port number of the connected client (miniboa-compatible).

property terminal_type: str: Client terminal type (miniboa-compatible).

property columns: int: Terminal width (miniboa-compatible).

property rows: int: Terminal height (miniboa-compatible).

property connect_time: float: Timestamp when connection was established (miniboa-compatible).

property last_input_time: float: Timestamp of last input received (miniboa-compatible).

send(text)[source]

Send text to the client (miniboa-compatible).

Alias for write(). Normalizes newlines to rn like miniboa.

Parameters:: text (Union[str, bytes]) – Text to send.
Return type:: None

addrport()[source]

Return client’s IP:PORT as string (miniboa-compatible).

Return type:: str
Returns:: String in format “IP:PORT”.

idle()[source]

Seconds since last input received (miniboa-compatible).

Return type:: float
Returns:: Idle time in seconds.

duration()[source]

Seconds since connection was established (miniboa-compatible).

Return type:: float
Returns:: Connection duration in seconds.

deactivate()[source]

Set connection to disconnect on next opportunity (miniboa-compatible).

Same as setting active = False.

Return type:: None