TelnetServer recv error: Forcing an RST request to the telnet server can cause a crash

[R4be1]

TelnetServer and the local terminal start simultaneously.

#!/usr/bin/env python3

import asyncio
import logging
import errno

from prompt_toolkit import PromptSession
from prompt_toolkit.patch_stdout import patch_stdout
from prompt_toolkit.contrib.telnet.server import TelnetServer

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)

logger = logging.getLogger(__name__)


def install_asyncio_exception_filter(loop):
    old_handler = loop.get_exception_handler()

    def exception_handler(loop, context):
        exc = context.get("exception")
        print(1231231)

        if isinstance(exc, ConnectionResetError):
            return

        if isinstance(exc, OSError):
            if getattr(exc, "errno", None) in (
                errno.ECONNRESET,
                errno.EPIPE,
            ):
                return

        if old_handler:
            old_handler(loop, context)
        else:
            loop.default_exception_handler(context)

    loop.set_exception_handler(exception_handler)


async def interact(connection):
    peer = connection.addr
    print("666 Client connected:"+ str(peer))
    logger.info("Client connected: %s", peer)

    try:
        session = PromptSession()

        await connection.send(
            "\r\n"
            "====================================\r\n"
            " prompt_toolkit Telnet Demo\r\n"
            "====================================\r\n"
            "Type 'exit' to quit.\r\n\r\n"
        )

        while True:
            try:
                with patch_stdout():
                    text = await session.prompt_async(
                        "telnet> "
                    )

            except (EOFError, KeyboardInterrupt):
                break

            if text.lower() in ("exit", "quit"):
                break

            await connection.send(
                f"You typed: {text}\r\n"
            )
        '''
        except ConnectionResetError:
            logger.info("Client reset connection: %s", peer)

        except OSError as e:
            if e.errno in (
                errno.ECONNRESET,
                errno.EPIPE,
            ):
                logger.info(
                    "Client disconnected unexpectedly: %s",
                    peer,
                )
            else:
                raise
        '''

    except Exception:
        logger.exception(
            "Unhandled exception from %s",
            peer,
        )

    finally:
        logger.info("Client disconnected: %s", peer)


#!/usr/bin/env python3
from prompt_toolkit import PromptSession
from prompt_toolkit.patch_stdout import patch_stdout
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.styles import Style
import asyncio

style = Style.from_dict({
    "prompt": "ansicyan bold",
    "command": "ansigreen",
    "error": "ansired bold",
})

async def main():
    loop = asyncio.get_running_loop()
    install_asyncio_exception_filter(loop)

    server = TelnetServer(
        host="0.0.0.0",
        port=8023,
        interact=interact,
    )

    asyncio.create_task( server.run() )

    print("Listening on 0.0.0.0:8023")




    session = PromptSession()

    print("Welcome to the demo interactive session!")
    print("Type 'exit' or 'quit' to leave.\n")

    while True:
        try:
            with patch_stdout():
                text = await session.prompt_async(HTML('<prompt>MyPrompt> </prompt>'), style=style)

        except (EOFError, KeyboardInterrupt):
            print("\n[Session terminated]")
            break

        text = text.strip()
        if not text:
            continue

        if text.lower() in ("exit", "quit"):
            print("[Goodbye]")
            break

        print(HTML(f"<command>You typed:</command> {text}"), style=style)


if __name__ == "__main__":
    asyncio.run(main())

Exploit/Client request:

import socket
import time

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("127.0.0.1", 8023))

s.send(b"hello")
time.sleep(1)


s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
             (1, 0))   # linger=0 -> RST

s.close()

Crash:

MyPrompt> 
MyPrompt> 2026-06-02 01:18:59,623 INFO New connection '127.0.0.1' 60252
2026-06-02 01:18:59,624 INFO Initializing telnet connection
2026-06-02 01:18:59,624 INFO Starting interaction '127.0.0.1' 60252
Press ENTER to continue...2026-06-02 01:19:00,642 INFO Connection closed by client. '127.0.0.1' 60252

Unhandled exception in event loop:
  File "/home/debian/.pyenv/versions/3.12.7/lib/python3.12/asyncio/events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
  File "/home/debian/.pyenv/versions/3.12.7/lib/python3.12/site-packages/prompt_toolkit/contrib/telnet/server.py", line 185, in handle_incoming_data
    data = self.conn.recv(1024)
           ^^^^^^^^^^^^^^^^^^^^

Exception [Errno 104] Connection reset by peer
Press ENTER to continue...

[R4be1]

Temporary relief measures.

# ==================== 2. 硬核猴子补丁（Monkey Patch） ====================
# 直接修复原库中 run_application 内部 handle_incoming_data 没有捕获异常的 Bug

_original_run_application = TelnetConnection.run_application

async def safe_run_application(self: TelnetConnection) -> None:
    """重写底层的 run_application，为 handle_incoming_data 加上防线"""
    import contextvars
    from asyncio import get_running_loop
    from prompt_toolkit.application.current import create_app_session

    def handle_incoming_data() -> None:
        try:
            # 在这里死死掐住 RST 产生的各种网络底层异常
            data = self.conn.recv(1024)
        except (ConnectionResetError, BrokenPipeError, OSError) as e:
            logger.info("Client matrix connection broken (RST caught via Patch): %s", self.addr)
            self.close()
            return

        if data:
            self.feed(data)
        else:
            logger.info("Connection closed by client. {!r} {!r}".format(*self.addr))
            self.close()

    # 替换原生 add_reader 逻辑
    loop = get_running_loop()
    loop.add_reader(self.conn, handle_incoming_data)

    try:
        await self._ready.wait()
        with create_app_session(input=self.vt100_input, output=self.vt100_output):
            self.context = contextvars.copy_context()
            await self.interact(self)
    except (ConnectionResetError, BrokenPipeError, OSError):
        pass  # 消化掉协程层面的残存波动
    finally:
        self.close()

# 激活补丁：用我们的安全函数替换掉 prompt_toolkit 的原生错误实现
TelnetConnection.run_application = safe_run_application

[R4be1]

The _accept function in the TelnetServer class will also trigger an error when encountering 104, but it is within the expected try block, yet traceback.print_exc() is used to print it.

See: https://github.com/prompt-toolkit/python-prompt-toolkit/blob/main/src/prompt_toolkit/contrib/telnet/server.py#L421

The simplest way to reproduce it is nmap -p8023 127.0.0.1