如何用Python实现一个Telnet远程登录客户端

访客 网络编程 1

本文目录导读:

  1. 基础Telnet客户端
  2. 扩展功能版本
  3. 依赖安装
  4. 使用说明

我来为您提供一个完整的Python Telnet远程登录客户端实现。

基础Telnet客户端

import telnetlib
import time
from typing import Optional, List
class TelnetClient:
    """Telnet远程登录客户端"""
    def __init__(self, host: str, port: int = 23, timeout: int = 10):
        """
        初始化Telnet客户端
        Args:
            host: 目标主机地址
            port: 端口号,默认23
            timeout: 连接超时时间(秒)
        """
        self.host = host
        self.port = port
        self.timeout = timeout
        self.tn: Optional[telnetlib.Telnet] = None
        self.is_connected = False
    def connect(self) -> bool:
        """
        建立Telnet连接
        Returns:
            bool: 连接是否成功
        """
        try:
            print(f"正在连接到 {self.host}:{self.port}...")
            self.tn = telnetlib.Telnet(self.host, self.port, self.timeout)
            self.is_connected = True
            print("连接成功!")
            return True
        except Exception as e:
            print(f"连接失败: {e}")
            return False
    def login(self, username: str, password: str, 
              username_prompt: str = b"login: ",
              password_prompt: str = b"Password: ") -> bool:
        """
        登录到远程系统
        Args:
            username: 用户名
            password: 密码
            username_prompt: 用户名提示符
            password_prompt: 密码提示符
        Returns:
            bool: 登录是否成功
        """
        if not self.is_connected:
            print("未连接到服务器")
            return False
        try:
            # 等待登录提示
            self.tn.read_until(username_prompt, timeout=5)
            self.tn.write(username.encode('ascii') + b"\n")
            time.sleep(1)
            # 等待密码提示
            self.tn.read_until(password_prompt, timeout=5)
            self.tn.write(password.encode('ascii') + b"\n")
            time.sleep(1)
            print("登录成功!")
            return True
        except Exception as e:
            print(f"登录失败: {e}")
            return False
    def send_command(self, command: str, wait_time: float = 0.5) -> str:
        """
        发送命令并获取输出
        Args:
            command: 要执行的命令
            wait_time: 等待时间
        Returns:
            str: 命令输出结果
        """
        if not self.is_connected:
            return "未连接到服务器"
        try:
            self.tn.write(command.encode('ascii') + b"\n")
            time.sleep(wait_time)
            output = self.tn.read_very_eager().decode('ascii', errors='ignore')
            return output
        except Exception as e:
            return f"执行命令失败: {e}"
    def read_until(self, prompt: bytes = b"#", timeout: int = 10) -> str:
        """
        读取直到特定提示符
        Args:
            prompt: 提示符
            timeout: 超时时间
        Returns:
            str: 读取的内容
        """
        if not self.is_connected:
            return "未连接到服务器"
        try:
            data = self.tn.read_until(prompt, timeout)
            return data.decode('ascii', errors='ignore')
        except Exception as e:
            return f"读取失败: {e}"
    def interactive_mode(self):
        """交互式命令行模式"""
        print(f"\n进入交互模式 - {self.host}")
        print("输入 'exit' 或 'quit' 退出")
        print("-" * 50)
        while True:
            try:
                command = input(f"{self.host}> ")
                if command.lower() in ['exit', 'quit']:
                    break
                if command.strip():
                    output = self.send_command(command)
                    if output:
                        print(output)
            except KeyboardInterrupt:
                print("\n退出交互模式")
                break
            except Exception as e:
                print(f"错误: {e}")
    def close(self):
        """关闭连接"""
        if self.tn and self.is_connected:
            try:
                self.tn.close()
                print("连接已关闭")
            except:
                pass
            finally:
                self.is_connected = False
    def __enter__(self):
        """上下文管理器入口"""
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器出口"""
        self.close()
# 使用示例
def main():
    # 方式1:使用上下文管理器
    with TelnetClient("192.168.1.1", 23) as client:
        if client.connect():
            # 登录
            if client.login("admin", "password"):
                # 读取登录后的输出
                output = client.read_until(b">", timeout=5)
                print(f"登录后输出: {output}")
                # 执行命令
                result = client.send_command("show version")
                print(f"命令输出: {result}")
                # 进入交互模式
                client.interactive_mode()
    # 方式2:手动管理连接
    # client = TelnetClient("192.168.1.1", 23)
    # try:
    #     if client.connect():
    #         if client.login("admin", "password"):
    #             client.interactive_mode()
    # finally:
    #     client.close()
if __name__ == "__main__":
    main()

扩展功能版本

import telnetlib
import logging
import sys
from typing import Optional, Callable
class AdvancedTelnetClient:
    """增强版Telnet客户端"""
    def __init__(self, host: str, port: int = 23, timeout: int = 10,
                 debug: bool = False):
        """
        初始化增强版客户端
        Args:
            host: 目标主机
            port: 端口
            timeout: 超时时间
            debug: 是否启用调试模式
        """
        self.host = host
        self.port = port
        self.timeout = timeout
        self.tn: Optional[telnetlib.Telnet] = None
        # 配置日志
        self.logger = logging.getLogger(__name__)
        if debug:
            logging.basicConfig(level=logging.DEBUG)
        # 回调函数
        self.data_received_callback: Optional[Callable] = None
    def set_callback(self, callback: Callable):
        """设置数据接收回调"""
        self.data_received_callback = callback
    def expect(self, patterns: List[bytes], timeout: int = 10):
        """
        使用expect模式匹配
        Args:
            patterns: 要匹配的模式列表
            timeout: 超时时间
        Returns:
            tuple: (index, match, text)
        """
        if not self.tn:
            return (-1, None, b"")
        try:
            return self.tn.expect(patterns, timeout)
        except Exception as e:
            self.logger.error(f"Expect失败: {e}")
            return (-1, None, b"")
    def send_line(self, text: str):
        """发送一行文本"""
        if self.tn:
            self.tn.write(text.encode('ascii') + b"\n")
            self.logger.debug(f"发送: {text}")
    def read_all(self, timeout: int = 5) -> str:
        """读取所有可用数据"""
        if not self.tn:
            return ""
        try:
            import select
            data = b""
            while True:
                ready = select.select([self.tn], [], [], timeout)
                if ready[0]:
                    chunk = self.tn.read_eager()
                    if not chunk:
                        break
                    data += chunk
                    if self.data_received_callback:
                        self.data_received_callback(chunk)
                else:
                    break
            return data.decode('ascii', errors='ignore')
        except Exception as e:
            self.logger.error(f"读取数据失败: {e}")
            return ""
def main_advanced():
    """增强版使用示例"""
    # 配置日志
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s'
    )
    # 创建客户端
    client = AdvancedTelnetClient("192.168.1.1", 23, debug=True)
    # 设置数据接收回调
    def on_data(data: bytes):
        print(f"收到数据: {data}")
    client.set_callback(on_data)
    if client.connect():
        # 使用expect登录
        client.send_line("admin")
        index, match, text = client.expect([b"Password:", b">"])
        if index == 0:
            client.send_line("password")
            client.expect([b">"])
        # 执行命令并读取输出
        client.send_line("show ip interface brief")
        output = client.read_all()
        print(f"输出: {output}")
if __name__ == "__main__":
    # 运行基础版本
    main()
    # 运行增强版本
    # main_advanced()

依赖安装

# Telnet库是Python标准库,无需额外安装
# 如果需要更多功能,可以安装:
pip install telnetlib3  # 异步版本

使用说明

  1. 基本用法:创建客户端,连接,登录,执行命令,关闭连接
  2. 交互模式:提供类似命令行终端的交互体验
  3. 上下文管理器:支持with语句自动管理资源
  4. 错误处理:包含完整的异常处理机制
  5. 日志系统:支持调试模式查看详细信息

这个实现提供了完整的Telnet客户端功能,可以根据实际需求进行修改和扩展。

标签: Telnet Python

抱歉,评论功能暂时关闭!