本文目录导读:
我来为您提供一个完整的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 # 异步版本
使用说明
- 基本用法:创建客户端,连接,登录,执行命令,关闭连接
- 交互模式:提供类似命令行终端的交互体验
- 上下文管理器:支持
with语句自动管理资源 - 错误处理:包含完整的异常处理机制
- 日志系统:支持调试模式查看详细信息
这个实现提供了完整的Telnet客户端功能,可以根据实际需求进行修改和扩展。