我正在尝试找到以下问题的解决方案:
我想从服务器 A(完全权限)通过 Jumhost B(无 sudo)使用 Python 连接到多个网络设备(一个接一个就足够了,不必同时连接)。仅使用 SSH 不会有问题,但很多设备仅使用 Telnet(我知道这不安全,但我没有决定这样做)。
经过研究,我找到了多种链式 SSH 连接的解决方案,例如 Paramiko、Netmiko、Pxssh 等。但我找不到使用 Telnet 实现最后一步的正确方法。目前我有以下代码:
class SSHTool(): def __init__(self, host, user, auth, via=None, via_user=None, via_auth=None): if via: t0 = ssh.Transport(via) t0.start_client() t0.auth_password(via_user, via_auth) # setup forwarding from 127.0.0.1:<free_random_port> to |host| channel = t0.open_channel('direct-tcpip', host, ('127.0.0.1', 0)) self.transport = ssh.Transport(channel) else: self.transport = ssh.Transport(host) self.transport.start_client() self.transport.auth_password(user, auth) def run(self, cmd): ch = self.transport.open_session() ch.set_combine_stderr(True) ch.exec_command(cmd) retcode = ch.recv_exit_status() buf = '' while ch.recv_ready(): buf += str(ch.recv(1024)) return (buf, retcode) host = ('192.168.0.136', 22) via_host = ('192.168.0.213', 22) ssht = SSHTool(host, '', '', via=via_host, via_user='', via_auth='') output=ssht.run('ls') print(output)
通过这个,我能够链接我的 Jumphost,但我不知道如何实现 Telnet 连接。有人知道合适的解决方案吗?
为了从服务器 A 通过跳板机 B(Jumphost)连接到支持 Telnet 的网络设备,可以使用以下方法:
telnetlib
下面是完整的代码示例:
import paramiko import telnetlib import threading from socket import socket class SSHTool: def __init__(self, jumphost, jumphost_user, jumphost_password): """ 初始化 SSH 客户端并连接到跳板机 """ self.ssh_client = paramiko.SSHClient() self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh_client.connect(jumphost[0], jumphost[1], username=jumphost_user, password=jumphost_password) self.local_port = None def setup_tunnel(self, telnet_host, telnet_port): """ 设置从本地端口到目标 Telnet 设备的 SSH 隧道 """ transport = self.ssh_client.get_transport() local_socket = socket() local_socket.bind(('127.0.0.1', 0)) # 分配任意可用端口 self.local_port = local_socket.getsockname()[1] local_socket.close() def forward_tunnel(): transport.request_port_forward('127.0.0.1', self.local_port) channel = transport.open_channel('direct-tcpip', (telnet_host, telnet_port), ('127.0.0.1', self.local_port)) while True: # 保持隧道连接 if not channel.recv_ready(): break # 使用线程保持端口转发 threading.Thread(target=forward_tunnel, daemon=True).start() def close(self): """关闭 SSH 连接""" self.ssh_client.close() def connect_telnet(local_port): """ 使用 Telnet 连接到目标设备 """ try: telnet = telnetlib.Telnet('127.0.0.1', local_port) telnet.read_until(b"Username: ") # 根据设备提示调整 telnet.write(b"admin\n") telnet.read_until(b"Password: ") telnet.write(b"password\n") print(telnet.read_some().decode('utf-8')) # 输出设备初始提示 telnet.close() except Exception as e: print(f"Telnet 连接失败: {e}") # 示例用法 jumphost = ('192.168.0.213', 22) # 跳板机地址和端口 telnet_device = ('192.168.0.136', 23) # Telnet 设备地址和端口 # 跳板机登录信息 jumphost_user = "jumphost_user" jumphost_password = "jumphost_password" # 初始化 SSH 工具并设置隧道 ssh_tool = SSHTool(jumphost, jumphost_user, jumphost_password) ssh_tool.setup_tunnel(telnet_device[0], telnet_device[1]) # 使用隧道连接到 Telnet 设备 connect_telnet(ssh_tool.local_port) # 关闭 SSH 连接 ssh_tool.close()
paramiko.SSHClient
设置 self.local_port 为本地可用端口,用于 SSH 隧道的转发。
self.local_port
SSH 隧道设置:
隧道将本地 127.0.0.1:<local_port> 的流量转发到目标 Telnet 设备。
127.0.0.1:<local_port>
Telnet 连接:
telnet.read_until() 用于等待设备的登录提示,telnet.write() 用于发送用户名和密码。
telnet.read_until()
telnet.write()
线程管理隧道:
forward_tunnel()
调整 Telnet 登录提示: 根据设备的实际登录提示(如 Username: 和 Password:),修改 telnet.read_until() 的参数。
Username:
Password:
线程安全: 线程用于保持隧道连接活跃,确保主程序不会阻塞。
安全性:
通过这种方法,可以实现从跳板机到多个 Telnet 设备的顺利连接。