使用 Paramiko 实现 SSH 远程连接和命令执行

在这里插入图片描述

每当灶火燃起，香气弥漫，熟悉的味道植入记忆深处，家，才获得完整的意义。万户千家，味道迥异，但幸福的滋味，却何其相同。
——《风味人间》

在现代网络管理中，远程管理设备变得越来越重要。本文将详细介绍如何使用 Python 的 Paramiko 库通过 SSH 连接到交换机，并执行命令以获取其状态。我们将分析代码的每个部分，以及使用该代码的实际场景。

一、代码概述

我们编写的代码主要实现了以下功能：

通过 SSH 连接到网络设备（如交换机）。
发送命令并获取设备的响应。
将设备状态记录到日志中。

代码结构

代码由几个主要部分组成：

导入模块：
```
import paramiko  
import time  
import logging
```
- paramiko：用于实现 SSH 连接。
- time：用于控制延迟和超时。
- logging：用于记录日志，便于后续审查。
日志设置：
```
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
```
这里我们设置了日志级别和格式，使得日志输出更易于阅读和分析。
发送命令的函数：
```
def send_command(remote_conn, command, timeout=2):
```
该函数的主要任务是向远程设备发送命令并接收其输出。它通过循环检查输出是否准备好，并在必要时实现超时机制，以避免长时间等待。
获取交换机状态的函数：
```
def get_switch_status(ip, username, password):
```
该函数通过提供的 IP 地址、用户名和密码连接到交换机，并发送命令以获取交换机版本信息。连接失败时会记录错误日志。
交换机列表和批量检查：
```
switches = [("192.168.249.100", "root", "888888")]
```
在这个例子中，我们只连接了一个交换机，但可以轻松扩展为多个设备。

二、代码分析

1. 连接和会话

通过 paramiko.SSHClient 创建 SSH 客户端并连接设备。使用 invoke_shell() 方法获取一个交互式会话，允许我们发送命令并接收输出。

2. 发送命令

send_command 函数中使用 remote_conn.send(command + "\n") 发送命令。接下来，通过循环检查 recv_ready() 方法来获取命令输出。

3. 超时控制

为了避免在接收命令输出时陷入无限等待，我们实现了一个简单的超时机制，确保程序在指定时间内完成。

4. 错误处理

通过 try-except-finally 块处理可能的异常，确保即使发生错误也能正常关闭 SSH 连接并记录日志。

三、使用场景

这种 SSH 远程管理的方案适合于多种场景，例如：

网络设备监控：
在大型网络环境中，管理员需要定期检查设备状态。使用该脚本可以轻松获取交换机的版本信息，确保设备正常运行。
自动化运维：
在日常运维中，管理员可以将此脚本集成到更复杂的自动化工作流中，实现设备配置、状态检查等操作的自动化。
故障排除：
当网络出现问题时，快速获取设备状态可以帮助管理员迅速定位问题源。

四、完整代码

以下是完整的代码实现：

# -*- coding: utf-8 -*-
import paramiko  
import time  
import logging  # 设置日志  
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')  def send_command(remote_conn, command, timeout=2):  """发送命令并接收响应"""  remote_conn.send(command + "\n")  time.sleep(1)  # 等待命令执行  output = ""  # 可能需要多次接收来确保获取完整的输出  while True:  if remote_conn.recv_ready():  data = remote_conn.recv(65535).decode('utf-8')  if not data:  break  output += data  else:  # 设置超时以避免无限等待  time.sleep(0.1)  timeout -= 0.1  if timeout <= 0:  break  return output  def get_switch_status(ip, username, password):  client = paramiko.SSHClient()  client.set_missing_host_key_policy(paramiko.AutoAddPolicy())  try:  client.connect(ip, username=username, password=password, look_for_keys=False, allow_agent=False, timeout=10)  remote_conn = client.invoke_shell()  time.sleep(1)  # 等待会话稳定  # 发送命令并接收输出  output = send_command(remote_conn, "display version")  # 打印输出  logging.info(f"====== {ip} ======")  logging.info(output)  logging.info("===================")  except Exception as e:  logging.error(f"Failed to connect to {ip}: {str(e)}")  finally:  client.close()  # 交换机列表  
switches = [  ("192.168.249.100", "root", "888888")  
]  # 批量检查交换机状态  
for ip, username, password in switches:  get_switch_status(ip, username, password)