telnetlib3: Async Telnet Server and Client
telnetlib3 is an asynchronous Python library for building Telnet servers and clients, supporting a wide array of Telnet Option specifications (RFCs). It offers both a programmatic API and command-line utilities. Currently at version 4.0.2, it maintains an active release cadence with frequent bug fixes and feature enhancements.
Warnings
- breaking The `color_filter` module was removed. Any code directly importing or using this module will break.
- breaking The default operating mode for `telnetlib3-client` (CLI) and `telnetlib3-server --pty-exec` (CLI) changed from line-buffered local echo to raw terminal mode (no line buffering, no local echo). This significantly alters user interaction.
- breaking The default value for the `connect_minwait` parameter in `open_connection()` and `create_server()` was changed from 1.0 seconds to 0 seconds. This affects how long the library waits before performing initial negotiation.
- gotcha Prior to version 4.0.2, MCCP2 decompression could fail on some MUD servers using raw deflate or gzip-wrapped compression, leading to garbled banners or corrupted data.
Install
-
pip install telnetlib3
Imports
- telnetlib3
import telnetlib3
- create_server
from telnetlib3 import create_server
- open_connection
from telnetlib3 import open_connection
Quickstart
import telnetlib3
import asyncio
import os
async def simple_shell(reader, writer):
peername = writer.get_extra_info('peername')
print(f'Client {peername} connected.')
writer.write(f'Hello, {peername}! Type something and press Enter (or Ctrl+C to quit).
')
while True:
try:
text = await reader.read(1024)
if not text:
break
print(f'Received from {peername}: {text!r}')
writer.write(f'You said: {text!r}
')
await writer.drain()
except asyncio.IncompleteReadError:
break # Client disconnected
except ConnectionResetError:
break # Client forcibly closed
print(f'Client {peername} disconnected.')
async def main():
# Use os.environ for port to make it runnable without hardcoding
port = int(os.environ.get('TELNET_PORT', '6023'))
server = await telnetlib3.create_server(port=port, shell=simple_shell)
print(f"Telnet server listening on port {port}")
async with server:
await server.serve_forever()
if __name__ == '__main__':
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Server stopped gracefully.")