From 0724c3341495ab4c590ac5d46dffad466beab001 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20=C5=A0koda?= Date: Thu, 7 Nov 2024 03:53:47 +0100 Subject: [PATCH] Switch from deprecated telnet to telnetlib3 (#410) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Joongi Kim --- aiomonitor/telnet.py | 47 ++++++++++++++++++++++---------------------- requirements-dev.txt | 1 + setup.cfg | 2 +- 3 files changed, 26 insertions(+), 24 deletions(-) diff --git a/aiomonitor/telnet.py b/aiomonitor/telnet.py index 976db84..1cb19dc 100644 --- a/aiomonitor/telnet.py +++ b/aiomonitor/telnet.py @@ -6,10 +6,11 @@ import os import struct import sys -import telnetlib import termios from typing import Dict, Optional, TextIO, Tuple +import telnetlib3 + ModeDef = collections.namedtuple( "ModeDef", ["iflag", "oflag", "cflag", "lflag", "ispeed", "ospeed", "cc"] ) @@ -61,7 +62,7 @@ def restore_mode(self) -> None: def determine_mode(self, mode: ModeDef) -> ModeDef: # Reference: https://github.com/jquast/telnetlib3/blob/b90616f/telnetlib3/client_shell.py#L76 - if not self._remote_options[telnetlib.ECHO]: + if not self._remote_options[telnetlib3.ECHO]: return mode iflag = mode.iflag & ~( termios.BRKINT @@ -160,31 +161,31 @@ async def _handle_user_input(self) -> None: pass async def _handle_nego(self, command: bytes, option: bytes) -> None: - if command == telnetlib.DO and option == telnetlib.TTYPE: - self._conn_writer.write(telnetlib.IAC + telnetlib.WILL + telnetlib.TTYPE) - self._conn_writer.write(telnetlib.IAC + telnetlib.SB + telnetlib.TTYPE) - self._conn_writer.write(telnetlib.BINARY + self._term.encode("ascii")) - self._conn_writer.write(telnetlib.IAC + telnetlib.SE) + if command == telnetlib3.DO and option == telnetlib3.TTYPE: + self._conn_writer.write(telnetlib3.IAC + telnetlib3.WILL + telnetlib3.TTYPE) + self._conn_writer.write(telnetlib3.IAC + telnetlib3.SB + telnetlib3.TTYPE) + self._conn_writer.write(telnetlib3.BINARY + self._term.encode("ascii")) + self._conn_writer.write(telnetlib3.IAC + telnetlib3.SE) await self._conn_writer.drain() - elif command == telnetlib.DO and option == telnetlib.NAWS: - self._conn_writer.write(telnetlib.IAC + telnetlib.WILL + telnetlib.NAWS) - self._conn_writer.write(telnetlib.IAC + telnetlib.SB + telnetlib.NAWS) + elif command == telnetlib3.DO and option == telnetlib3.NAWS: + self._conn_writer.write(telnetlib3.IAC + telnetlib3.WILL + telnetlib3.NAWS) + self._conn_writer.write(telnetlib3.IAC + telnetlib3.SB + telnetlib3.NAWS) fmt = "HHHH" buf = b"\x00" * struct.calcsize(fmt) try: buf = fcntl.ioctl(self._stdin.fileno(), termios.TIOCGWINSZ, buf) rows, cols, _, _ = struct.unpack(fmt, buf) self._conn_writer.write(struct.pack(">HH", cols, rows)) - self._conn_writer.write(telnetlib.IAC + telnetlib.SE) + self._conn_writer.write(telnetlib3.IAC + telnetlib3.SE) except OSError: rows, cols = 22, 80 await self._conn_writer.drain() - elif command == telnetlib.WILL: + elif command == telnetlib3.WILL: self._remote_options[option] = True - if option in (telnetlib.ECHO, telnetlib.BINARY, telnetlib.SGA): - self._conn_writer.write(telnetlib.IAC + telnetlib.DO + option) + if option in (telnetlib3.ECHO, telnetlib3.BINARY, telnetlib3.SGA): + self._conn_writer.write(telnetlib3.IAC + telnetlib3.DO + option) await self._conn_writer.drain() - elif command == telnetlib.WONT: + elif command == telnetlib3.WONT: self._remote_options[option] = False async def _handle_sb(self, option: bytes, chunk: bytes) -> None: @@ -196,7 +197,7 @@ async def _handle_received(self): while not self._conn_reader.at_eof(): buf += await self._conn_reader.read(128) while buf: - cmd_begin = buf.find(telnetlib.IAC) + cmd_begin = buf.find(telnetlib3.IAC) if cmd_begin == -1: self._stdout_writer.write(buf) await self._stdout_writer.drain() @@ -212,20 +213,20 @@ async def _handle_received(self): self._stdout_writer.write(buf_before) await self._stdout_writer.drain() if command in ( - telnetlib.WILL, - telnetlib.WONT, - telnetlib.DO, - telnetlib.DONT, + telnetlib3.WILL, + telnetlib3.WONT, + telnetlib3.DO, + telnetlib3.DONT, ): await self._handle_nego(command, option) - elif command == telnetlib.SB: - subnego_end = buf_after.find(telnetlib.IAC + telnetlib.SE) + elif command == telnetlib3.SB: + subnego_end = buf_after.find(telnetlib3.IAC + telnetlib3.SE) if subnego_end == -1: subnego_chunk = ( buf_after + ( await self._conn_reader.readuntil( - telnetlib.IAC + telnetlib.SE + telnetlib3.IAC + telnetlib3.SE ) )[:-2] ) diff --git a/requirements-dev.txt b/requirements-dev.txt index 1d20033..c0ed00a 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -15,5 +15,6 @@ ruff==0.7.2 terminaltables==3.1.10 towncrier==24.8.0 types-requests +telnetlib3==2.0.4 uvloop==0.21.0 build==1.2.2.post1 diff --git a/setup.cfg b/setup.cfg index 918d3fa..2b330fc 100644 --- a/setup.cfg +++ b/setup.cfg @@ -45,12 +45,12 @@ install_requires = janus>=1.0 jinja2>=3.1.2 backports.strenum>=1.2.4; python_version<"3.11" - standard-telnetlib>=3.13; python_version>="3.13" terminaltables trafaret>=2.1.1 typing-extensions>=4.1 prompt_toolkit>=3.0 aioconsole>=0.7.0 + telnetlib3>=2.0.4 [options.packages.find] exclude =