host.telnet¶

Telnet client for remote host connections.

TelnetClient handles the transport-level concerns: opening a telnet connection, negotiating protocol options (echo suppression), authenticating, and optionally sending NAWS (window-size) updates on SIGWINCH so remote TUIs reflow like they do under SSH. After login the reader/writer streams are handed off to a TelnetSession for command execution.

async otto.host.telnet.open_telnet_connection(*args: Any, **kwargs: Any) → Any¶

Lazy, patchable wrapper around telnetlib3.open_connection().

Kept as a module-level seam (tests monkeypatch it) while deferring the heavy telnetlib3 import to connect-time — see tests/unit/host/test_lazy_network_imports.py.

otto.host.telnet.abort_console_transports() → int¶

Synchronously abort every tracked single-client console transport.

Releases each FD (and the server-side single-client slot) via the transport’s own synchronous abort() — no event loop required, so this works even after a pytest-timeout signal aborts a test before its async close() could run. Best-effort and idempotent; returns the count aborted. Per-process.

class otto.host.telnet.TelnetClient(host: str, user: str, password: str, options: otto.host.options.TelnetOptions = <factory>, prompt: str | None = None, connect_port: int | None = None)¶

Bases: object

host : str¶

user : str¶

password : str¶

options : TelnetOptions¶

Connection options. Default reproduces otto’s historical behavior.

prompt : str | None = None¶

Shell prompt string the device displays after each command (e.g. ‘$ ‘ or ‘# ‘). Used during login to confirm authentication succeeded.

connect_port : int | None = None¶

Override port for ConnectionManager’s tunneled case. When None, options.port is used. Keeps tunnel port-forwarding transparent to the TelnetOptions carried through the rest of the stack.

reader : Any = None¶

writer : Any = None¶

async connect(interactive: bool = False) → None¶

Open the telnet connection, negotiate options, and log in.

Parameters:¶

interactive – When True, skip the DONT ECHO negotiation so the remote shell echoes keystrokes back — required for otto.host.interact.run_telnet_login() so the user sees what they type. Non-interactive callers (the default) get DONT ECHO so command echoes don’t mix with captured output.

async login() → None¶

Send credentials through an established telnet stream.

Readiness after the password is not confirmed here. When no prompt is configured, login() returns as soon as the password is written and the session’s marker handshake (ShellSession._ensure_initialized, which runs immediately after) is the deterministic readiness check — it reads through any banner/MOTD to a unique sentinel and is bounded by a timeout that surfaces a bad-credential login as a clear error.

property alive : bool¶

Whether the underlying TCP transport is still usable.

close() clears writer/reader to None, and the asyncio writer reports is_closing() after a peer-initiated EOF. Either signal means the next read/write would fail, so callers (notably ConnectionManager.telnet()) should treat the client as stale.

async close() → None¶

Gracefully close the telnet connection.