第303篇：Paramiko 库：SSH 连接网络设备

关键词

Paramiko、SSH 连接、远程执行命令、SSH 客户端、密钥认证、交互式命令、异常处理、网络自动化 Python

一、Paramiko 简介

1.1 什么是 Paramiko

Paramiko 是 Python 实现的 SSHv2 协议库，可以像 SSH 客户端一样连接网络设备：

Paramiko 在自动化中的位置：

Python 脚本 ┌────────────────────────────────────┐ └────────────────────────────────────┘ ▼ ┌────────────────────────────────────┐ 应用层：Your Automation Script ┌──────────────────────────────┐ └──────────────────────────────┘ SSH 网络设备（华为/思科/华三/...） SSH Server 端口 22 Netmiko（高级封装，推荐） 或 Paramiko（底层库）

为什么直接学 Paramiko： ┌─ 理解 SSH 协议底层原理 ├─ Netmiko 底层基于 Paramiko ├─ 灵活控制 SSH 连接 └─ 特殊情况（交互式命令）需要 Paramiko

安装： python -m pip install paramiko

1.2 SSH 认证方式

Paramiko 支持的 SSH 认证方式：

  1. 密码认证（最常用）
  ┌─ 用户名 + 密码
  ├─ 简单直接

  2. 密钥认证（推荐生产环境）
  ┌─ RSA/ECDSA 密钥对
  ├─ 无需明文密码
  ├─ 更安全

  3. 键盘交互认证
  ┌─ 需要输入额外信息（如 OTP）
  └─ 较少使用

二、Paramiko 基础操作

2.1 连接与执行命令

#!/usr/bin/env python3
# basic_connect.py — Paramiko 基础连接示例

import paramiko
import time

def connect_and_run(host, username, password, commands):
    """
    连接 SSH 设备并执行命令列表
    """
    # 创建 SSH 客户端
    client = paramiko.SSHClient()

    # 自动接受未知主机的密钥
    client.set_missing_host_key_policy(
        paramiko.AutoAddPolicy()
    )

    try:
        # 连接设备
        print(f"正在连接 {host}...")
        client.connect(
            hostname=host,
            username=username,
            password=password,
            port=22,
            look_for_keys=False,    # 不找本地密钥
            allow_agent=False,      # 不使用 SSH agent
        )
        print(f"连接成功")

        # 打开交互式 shell
        shell = client.invoke_shell()
        time.sleep(1)  # 等待 shell 就绪

        # 发送命令
        output = ""
        for cmd in commands:
            print(f"执行命令: {cmd}")
            shell.send(cmd + "\n")
            time.sleep(1)  # 等待命令执行
            output += shell.recv(65535).decode("utf-8")

        return output

    except paramiko.AuthenticationException:
        print("认证失败：用户名或密码错误")
    except paramiko.SSHException as e:
        print(f"SSH 错误: {e}")
    except Exception as e:
        print(f"其他错误: {e}")
    finally:
        client.close()
        print("连接已关闭")

# 使用示例
if __name__ == "__main__":
    result = connect_and_run(
        host="192.168.1.1",
        username="admin",
        password="Huawei@123",
        commands=[
            "display version",
            "display interface brief",
            "display vlan",
        ]
    )
    print("=== 命令输出 ===")
    print(result)

2.2 使用 exec_command

# exec_command_example.py — 使用 exec_command

import paramiko

def exec_commands(host, username, password, commands):
    """
    使用 exec_command（非交互式，一次发送一条）
    注意：华为设备可能不支持 exec_command 的分页问题
    """
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            host, username=username,
            password=password,
            look_for_keys=False
        )

        for cmd in commands:
            # exec_command 是非交互式的
            stdin, stdout, stderr = client.exec_command(cmd)

            # 读取输出
            output = stdout.read().decode("utf-8")
            error = stderr.read().decode("utf-8")

            print(f"\n─── 命令: {cmd} ───")
            if output:
                print(output)
            if error:
                print(f"[错误] {error}")

    finally:
        client.close()

# 注意：华为设备 screen-length 可能导致输出被截断
# 建议先设置：screen-length 0 temporary
if __name__ == "__main__":
    exec_commands(
        host="192.168.1.1",
        username="admin",
        password="Huawei@123",
        commands=[
            "screen-length 0 temporary",  # 取消分页
            "display ip interface brief",
        ]
    )

三、处理设备交互

3.1 处理分页

# handle_pagination.py — 处理命令输出分页

import paramiko
import time

def send_command_no_page(host, username, password, command):
    """
    发送命令并处理分页输出
    """
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            host, username=username,
            password=password,
            look_for_keys=False
        )

        shell = client.invoke_shell()
        time.sleep(1)

        # 先取消分页
        shell.send("screen-length 0 temporary\n")
        time.sleep(1)
        shell.recv(65535)  # 清除输出缓存

        # 发送实际命令
        shell.send(f"{command}\n")
        time.sleep(2)

        # 循环读取直到输出结束
        output = ""
        while True:
            if shell.recv_ready():
                data = shell.recv(65535).decode("utf-8")
                output += data
                # 如果包含提示符，说明命令执行完毕
                if data.endswith(">") or data.endswith("]"):
                    break
            else:
                time.sleep(0.5)
                # 超时保护
                if len(output) > 0:
                    break

        return output

    finally:
        client.close()

# 测试
if __name__ == "__main__":
    output = send_command_no_page(
        host="192.168.1.1",
        username="admin",
        password="Huawei@123",
        command="display current-configuration"
    )
    print(output[:2000])  # 只打印前 2000 字符

3.2 配置模式

# config_mode.py — 进入系统视图执行配置

import paramiko
import time

def configure_device(host, username, password, config_commands):
    """
    进入系统视图执行配置命令
    """
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            host, username=username,
            password=password,
            look_for_keys=False
        )

        shell = client.invoke_shell()
        time.sleep(1)

        # 读取欢迎信息
        shell.recv(65535)

        # 进入系统视图
        shell.send("system-view\n")
        time.sleep(0.5)
        shell.recv(65535)

        # 执行配置命令
        for cmd in config_commands:
            print(f"配置: {cmd}")
            shell.send(f"{cmd}\n")
            time.sleep(0.5)
            output = shell.recv(65535).decode("utf-8")
            if "%" in output:  # 华为设备错误提示以 % 开头
                print(f"[警告] {output.strip()}")

        # 提交配置
        shell.send("commit\n")  # 如果是 VRP8+
        # 或者 shell.send("save\n")  # VRP5/VRP8
        time.sleep(1)
        print(shell.recv(65535).decode("utf-8"))

        # 退出系统视图
        shell.send("return\n")
        time.sleep(0.5)
        shell.recv(65535)

        print(f"配置完成")

    except Exception as e:
        print(f"配置失败: {e}")
    finally:
        client.close()

if __name__ == "__main__":
    configure_device(
        host="192.168.1.1",
        username="admin",
        password="Huawei@123",
        config_commands=[
            "vlan batch 100 200 300",
            "interface 10GE1/0/1",
            " port link-type trunk",
            " port trunk allow-pass vlan 100 200",
            " description To-Spine01",
            "quit",
        ]
    )

四、批量设备管理

4.1 多设备并发

# multi_device.py — 使用线程批量管理多设备

import paramiko
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def get_device_info(host, username, password):
    """
    获取单台设备基本信息
    """
    try:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(
            host, username=username,
            password=password,
            look_for_keys=False,
            timeout=10
        )

        shell = client.invoke_shell()
        time.sleep(1)
        shell.send("display version | include version\n")
        time.sleep(2)
        output = shell.recv(65535).decode("utf-8")

        client.close()

        # 提取版本信息
        version = "未知"
        for line in output.split("\n"):
            if "version" in line.lower() or "V200" in line:
                version = line.strip()
                break

        return {"host": host, "status": "成功", "version": version}

    except Exception as e:
        return {"host": host, "status": "失败", "error": str(e)}

def batch_collect(device_list):
    """
    批量收集设备信息（并发执行）
    """
    results = []
    with ThreadPoolExecutor(max_workers=10) as executor:
        # 提交所有任务
        future_map = {
            executor.submit(
                get_device_info, dev["ip"],
                dev["username"], dev["password"]
            ): dev["name"]
            for dev in device_list
        }

        # 收集结果
        for future in as_completed(future_map):
            device_name = future_map[future]
            try:
                result = future.result()
                results.append(result)
                print(f"{device_name}: {result['status']}")
            except Exception as e:
                results.append({
                    "name": device_name,
                    "status": "异常",
                    "error": str(e)
                })

    return results

if __name__ == "__main__":
    devices = [
        {"name": "Leaf-01", "ip": "192.168.1.11", "username": "admin", "password": "Huawei@123"},
        {"name": "Leaf-02", "ip": "192.168.1.12", "username": "admin", "password": "Huawei@123"},
        {"name": "Spine-01", "ip": "192.168.1.21", "username": "admin", "password": "Huawei@123"},
        {"name": "Spine-02", "ip": "192.168.1.22", "username": "admin", "password": "Huawei@123"},
    ]

    results = batch_collect(devices)
    for r in results:
        print(f"  {r['host']}: {r.get('version', r.get('error', '未知'))}")

五、异常处理与日志

# ssh_with_logging.py — 带日志和异常处理的 SSH 连接

import paramiko
import logging
import time

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("ssh_operations.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class SSHManager:
    """SSH 连接管理器"""

    def __init__(self, host, username, password, port=22, timeout=30):
        self.host = host
        self.username = username
        self.password = password
        self.port = port
        self.timeout = timeout
        self.client = None
        self.shell = None

    def connect(self):
        """建立 SSH 连接"""
        try:
            self.client = paramiko.SSHClient()
            self.client.set_missing_host_key_policy(
                paramiko.AutoAddPolicy()
            )
            self.client.connect(
                hostname=self.host,
                port=self.port,
                username=self.username,
                password=self.password,
                timeout=self.timeout,
                look_for_keys=False,
                allow_agent=False,
            )
            logger.info(f"已连接到 {self.host}")
            return True
        except paramiko.AuthenticationException:
            logger.error(f"{self.host} 认证失败")
            return False
        except paramiko.SSHException as e:
            logger.error(f"{self.host} SSH 错误: {e}")
            return False
        except Exception as e:
            logger.error(f"{self.host} 连接失败: {e}")
            return False

    def send_command(self, command, wait_time=2):
        """发送命令并获取输出"""
        if not self.client:
            logger.error("未连接")
            return None

        try:
            if not self.shell:
                self.shell = self.client.invoke_shell()
                time.sleep(1)
                self.shell.recv(65535)  # 清空缓存

            self.shell.send(command + "\n")
            time.sleep(wait_time)

            output = ""
            while self.shell.recv_ready():
                output += self.shell.recv(65535).decode("utf-8")

            logger.debug(f"命令: {command}")
            return output

        except Exception as e:
            logger.error(f"命令执行失败: {e}")
            return None

    def close(self):
        """关闭连接"""
        try:
            if self.shell:
                self.shell.close()
            if self.client:
                self.client.close()
            logger.info(f"{self.host} 连接已关闭")
        except Exception as e:
            logger.error(f"关闭连接失败: {e}")

# 使用示例
if __name__ == "__main__":
    ssh = SSHManager("192.168.1.1", "admin", "Huawei@123")
    if ssh.connect():
        output = ssh.send_command("display ip interface brief")
        if output:
            print(output)
        ssh.close()

六、Paramiko vs Netmiko

对比项 Paramiko Netmiko
抽象层 底层 SSH 库 高级封装
代码量 多，需自己处理细节 少，两行就行
灵活性 高，完全控制 中，适合标准操作
简易性 低 高
分页处理 需自己处理 自动处理
错误处理 需自己写 内置
多厂商 需自己适配 自动适配 30+ 厂商
推荐 学习原理 日常使用

总结

关键点 说明
Paramiko 作用 Python SSH 库，远程操作网络设备
连接方式 SSHClient + connect()
执行命令 exec_command（非交互）或 invoke_shell（交互）
分页处理 先 screen-length 0 temporary 取消分页
配置模式 system-view 进入配置模式
批量管理 ThreadPoolExecutor 并发执行
日志 logging 模块记录操作日志

思考

  1. Paramiko 连接 SSH 设备需要哪几个基本参数？
  2. exec_command 和 invoke_shell 有什么区别？
  3. 为什么要处理命令输出的分页？
  4. set_missing_host_key_policy(AutoAddPolicy()) 有什么作用？
  5. 如何用 Python 并发连接 20 台设备？
  6. 写一个脚本：连接交换机，获取接口状态，输出到 CSV 文件。

下篇预告：第304篇 - Netmiko 批量执行配置命令，Netmiko 是 Paramiko 的高级封装，让网络自动化更简单。