第304篇：Netmiko 批量执行配置命令

关键词

Netmiko、批量执行、配置变更、多厂商支持、send_command、send_config_set、设备驱动、自动化脚本

一、Netmiko 简介

1.1 什么是 Netmiko

Netmiko 是对 Paramiko 的 高级封装库，专门为网络设备 SSH 操作优化：

Netmiko 的优势：

安装：

python -m pip install netmiko

1.2 支持的网络设备

Netmiko 支持的主要厂商（90+ 驱动）：

  华为：   huawei、huawei_vrpv8、huawei_smartax
  思科：   cisco_ios、cisco_xr、cisco_nxos、cisco_asa
  华三：   hp_comware、hp_procurve
  锐捷：   ruijie_os
  中兴：   zte_zxros
  Juniper：juniper_junos
  Arista： arista_eos
  Fortinet：fortinet

  查看完整列表：
  from netmiko import ssh_dispatcher
  print(list(ssh_dispatcher.platforms.keys()))

二、Netmiko 基础操作

2.1 连接与执行命令

#!/usr/bin/env python3
# netmiko_basic.py — Netmiko 基础连接

from netmiko import ConnectHandler

# 设备信息
device = {
    "device_type": "huawei",       # 驱动类型
    "host": "192.168.1.1",
    "username": "admin",
    "password": "admin123",
    "port": 22,                     # 默认 22
    "timeout": 30,                  # 超时秒数
    "verbose": False,               # 日志输出
}

# 建立连接
conn = ConnectHandler(**device)
print(f"已连接到 {device['host']}")

# 执行 show 命令
output = conn.send_command("display interface brief")
print(output)

# 执行多条命令
output = conn.send_commands([
    "display ip interface brief",
    "display bgp peer",
])
print(output)

# 关闭连接
conn.disconnect()

2.2 发送配置命令

#!/usr/bin/env python3
# netmiko_config.py — 配置模式操作

from netmiko import ConnectHandler
from netmiko.exceptions import ConfigException

device = {
    "device_type": "huawei",
    "host": "192.168.1.1",
    "username": "admin",
    "password": "admin123",
}

conn = ConnectHandler(**device)

# 进入系统视图并执行配置
config_commands = [
    "sysname CORP-CORE-01",
    "interface GigabitEthernet0/0/1",
    "description TRUNK_TO_SW01",
    "port link-type trunk",
    "port trunk allow-pass vlan 10 20 30",
    "quit",
    "interface GigabitEthernet0/0/2",
    "description LINK_TO_FW01",
    "port link-type access",
    "port default vlan 100",
    "quit",
]

# send_config_set 自动进入/退出配置模式
output = conn.send_config_set(config_commands)
print("配置输出：")
print(output)

# 验证配置
result = conn.send_command("display current-configuration interface GigabitEthernet0/0/1")
print(result)

conn.disconnect()

三、批量多设备操作

3.1 批量执行命令

#!/usr/bin/env python3
# batch_show.py — 批量收集设备信息

from netmiko import ConnectHandler
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

# 设备清单
DEVICES = [
    {"device_type": "huawei", "host": "192.168.1.1",
     "username": "admin", "password": "admin123"},
    {"device_type": "huawei", "host": "192.168.1.2",
     "username": "admin", "password": "admin123"},
    {"device_type": "huawei", "host": "192.168.1.3",
     "username": "admin", "password": "admin123"},
    {"device_type": "cisco_ios", "host": "192.168.2.1",
     "username": "admin", "password": "admin123"},
]

def collect_device(dev):
    """单设备收集"""
    try:
        conn = ConnectHandler(**dev)
        conn.enable()  # 思科需要 enable
        output = conn.send_command("display version" if "huawei" in dev["device_type"] else "show version")
        conn.disconnect()
        return (dev["host"], True, output)
    except Exception as e:
        return (dev["host"], False, str(e))

# 并行收集（提高效率）
print("开始批量收集...")
start = time.time()

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(collect_device, dev): dev for dev in DEVICES}
    for future in as_completed(futures):
        host, ok, data = future.result()
        status = "✓" if ok else "✗"
        print(f"[{status}] {host}")
        if ok:
            # 只打印前 5 行
            lines = data.strip().split("\n")
            for line in lines[:5]:
                print(f"  {line}")

elapsed = time.time() - start
print(f"\n总耗时：{elapsed:.1f} 秒")
print(f"4 台设备串行需 ~20 秒，并行仅 {elapsed:.1f} 秒")

3.2 批量下发配置

#!/usr/bin/env python3
# batch_config.py — 批量配置变更

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
import datetime

# 要添加的新 VLAN
NEW_VLANS = [
    {"id": 100, "name": "VLAN100_IT"},
    {"id": 200, "name": "VLAN200_HR"},
    {"id": 300, "name": "VLAN300_FINANCE"},
]

# 设备列表（核心交换机）
SWITCHES = [
    {"device_type": "huawei", "host": "192.168.100.1",
     "username": "admin", "password": "admin123"},
    {"device_type": "huawei", "host": "192.168.100.2",
     "username": "admin", "password": "admin123"},
]

def deploy_vlans(sw):
    """在交换机上创建 VLAN"""
    conn = ConnectHandler(**sw)
    hostname = conn.send_command("display current-configuration | include sysname")
    hostname = hostname.replace("sysname ", "").strip()

    commands = []
    for vlan in NEW_VLANS:
        commands.append(f"vlan {vlan['id']}")
        commands.append(f"name {vlan['name']}")
        commands.append("quit")

    print(f"[{hostname}] 即将创建 {len(NEW_VLANS)} 个 VLAN...")
    output = conn.send_config_set(commands)
    conn.save()  # 保存配置
    conn.disconnect()
    return (hostname, True, f"VLAN {[v['id'] for v in NEW_VLANS]} 创建成功")

# 执行
with ThreadPoolExecutor(max_workers=2) as ex:
    futures = [ex.submit(deploy_vlans, sw) for sw in SWITCHES]
    for f in as_completed(futures):
        name, ok, msg = f.result()
        print(f"[{'✓' if ok else '✗'}] {name}: {msg}")

四、高级功能

4.1 分页与超时处理

#!/usr/bin/env python3
# netmiko_advanced.py — 高级功能

from netmiko import ConnectHandler
from netmiko.exceptions import (
    NetmikoTimeoutException,
    NetmikoAuthenticationException,
    ReadTimeout,
)

device = {
    "device_type": "huawei",
    "host": "192.168.1.1",
    "username": "admin",
    "password": "admin123",
}

try:
    conn = ConnectHandler(**device)
except NetmikoAuthenticationException:
    print("认证失败，请检查用户名密码")
    exit(1)
except NetmikoTimeoutException:
    print("连接超时，请检查连通性")
    exit(1)

# 1. 处理超长输出
output = conn.send_command(
    "display current-configuration",
    read_timeout=60,           # 增加超时（秒）
    max_loops=500,             # 最大循环次数
)
print(f"配置共 {len(output.splitlines())} 行")

# 2. 使用正则匹配
output = conn.send_command(
    "display ip interface brief",
    use_textfsm=False,         # 是否启用 TextFSM 解析
)
print(output)

# 3. 命令终止字符串
output = conn.send_command(
    "display bgp routing-table",
    expect_string=r"<[\w\-]+>",  # 自定义提示符
    read_timeout=120,
)

# 4. 命令回退
output = conn.send_command(
    "display this",            # 当不确定是否在配置模式
    cmd_verify=False,          # 跳过命令回显验证
)

# 5. 全局延迟
conn.global_delay_factor = 2  # 所有命令延时 ×2（设备慢时使用）

conn.disconnect()

4.2 配置回滚

#!/usr/bin/env python3
# netmiko_rollback.py — 配置变更与回滚

from netmiko import ConnectHandler
import datetime

device = {
    "device_type": "huawei",
    "host": "192.168.1.1",
    "username": "admin",
    "password": "admin123",
}

def backup_config(conn):
    """备份当前配置"""
    backup = conn.send_command("display current-configuration")
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"backup_{device['host']}_{timestamp}.txt"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(backup)
    print(f"配置已备份到 {filename}")
    return backup

def apply_config(conn, commands):
    """应用配置变更"""
    output = conn.send_config_set(commands)
    # 检查是否有错误
    if "%" in output or "Error" in output:
        print("配置出错，正在回滚...")
        return False
    conn.save()
    return True

def rollback_config(conn, backup):
    """按行回滚配置（简化版）"""
    # 注意：实际回滚需要 diff + reverse 逻辑
    conn.send_config_set(["rollback last-configuration"])
    conn.save()
    print("已回滚到上一个配置")

conn = ConnectHandler(**device)

# 1. 先备份
backup = backup_config(conn)

# 2. 应用变更
new_config = [
    "snmp-agent sys-info version v3",
    "snmp-agent group v3 readonlygroup privacy",
    "snmp-agent usm-user v3 readonlyuser readonlygroup",
]
success = apply_config(conn, new_config)

# 3. 验证
if success:
    result = conn.send_command("display snmp-agent sys-info version")
    print(f"SNMP 配置验证：{result}")
else:
    rollback_config(conn, backup)

conn.disconnect()

五、日志与异常处理

5.1 完整的生产级脚本

#!/usr/bin/env python3
# netmiko_prod.py — 生产级批量操作脚本

from netmiko import ConnectHandler
from netmiko.exceptions import (
    NetmikoTimeoutException,
    NetmikoAuthenticationException,
    ConfigException,
)
import logging
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

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

# 设备清单（可从文件/YAML 读取）
DEVICES = [
    {"host": "192.168.1.1", "device_type": "huawei"},
    {"host": "192.168.1.2", "device_type": "huawei"},
]

# 通用凭证（生产环境建议从环境变量读取）
USERNAME = "admin"
PASSWORD = "admin123"

def safe_connect(host, device_type):
    """安全连接"""
    try:
        conn = ConnectHandler(
            device_type=device_type,
            host=host,
            username=USERNAME,
            password=PASSWORD,
            timeout=30,
            banner_timeout=15,
        )
        log.info(f"连接成功: {host}")
        return conn
    except NetmikoAuthenticationException:
        log.error(f"认证失败: {host}")
    except NetmikoTimeoutException:
        log.error(f"连接超时: {host}")
    except Exception as e:
        log.error(f"未知错误 {host}: {e}")
    return None

def process_device(dev):
    """处理单设备"""
    conn = safe_connect(dev["host"], dev["device_type"])
    if not conn:
        return (dev["host"], False, "连接失败")

    try:
        # 收集信息
        hostname = conn.send_command("display current-configuration | include sysname")
        hostname = hostname.replace("sysname ", "").strip()
        log.info(f"设备名称: {hostname}")

        # 执行配置
        commands = [
            "ntp-service unicast-server 203.0.113.1",
            "clock timezone CST add 08:00:00",
        ]
        output = conn.send_config_set(commands)
        log.info(f"配置完成: {hostname}")

        # 验证
        verify = conn.send_command("display ntp-service status")
        conn.save()

        # 延迟断开（确保配置生效）
        time.sleep(2)
        conn.disconnect()

        return (dev["host"], True, f"{hostname} 配置成功")

    except ConfigException as e:
        log.error(f"配置错误 {dev['host']}: {e}")
        return (dev["host"], False, f"配置错误: {e}")
    except Exception as e:
        log.error(f"执行错误 {dev['host']}: {e}")
        return (dev["host"], False, f"执行错误: {e}")

# 主流程
log.info("=== 批量操作开始 ===")
start = time.time()
success_count = 0

with ThreadPoolExecutor(max_workers=5) as ex:
    futures = {ex.submit(process_device, dev): dev for dev in DEVICES}
    for future in as_completed(futures):
        host, ok, msg = future.result()
        if ok:
            success_count += 1
        log.info(f"[{'✓' if ok else '✗'}] {host}: {msg}")

elapsed = time.time() - start
log.info(f"=== 完成: {success_count}/{len(DEVICES)} 成功, 耗时 {elapsed:.1f} 秒 ===")

六、Netmiko 与 Paramiko 对比

6.1 使用对比

#!/usr/bin/env python3
# compare_vs.py — Paramiko vs Netmiko 对比

"""

Paramiko 实现（约 35 行）：
┌──────────────────────────────────────┐
│ import paramiko                      │
│ import time                          │
│                                      │
│ client = paramiko.SSHClient()        │
│ client.set_missing_host_key_policy(  │
│     paramiko.AutoAddPolicy())        │
│ client.connect(host, port,           │
│     username, password)              │
│                                      │
│ channel = client.invoke_shell()      │
│ channel.send("screen-length 0 temp") │
│ time.sleep(1)                        │
│ channel.send("display int brief\n")  │
│ time.sleep(2)                        │
│ output = channel.recv(65535)         │
│ print(output.decode())               │
│ client.close()                       │
└──────────────────────────────────────┘

Netmiko 实现（约 8 行）：
┌──────────────────────────────────────┐
│ from netmiko import ConnectHandler   │
│                                      │
│ device = {                           │
│   "device_type": "huawei",           │
│   "host": "192.168.1.1",             │
│   "username": "admin",               │
│   "password": "admin123",            │
│ }                                    │
│                                      │
│ conn = ConnectHandler(**device)      │
│ output = conn.send_command(          │
│     "display interface brief")       │
│ print(output)                        │
│ conn.disconnect()                    │
└──────────────────────────────────────┘

结论：日常操作优先使用 Netmiko，
特殊需求（交互式命令、原始 SSH 通道）用 Paramiko
"""

6.2 选型建议

何时用 Netmiko：
  ┌─ 90% 日常运维需求
  ├─ 多厂商设备管理
  ├─ 批量命令执行
  ├─ 配置变更下发
  └─ 监控数据采集

何时用 Paramiko：
  ┌─ 需要交互式 CLI（如 dialog）
  ├─ 需要原始 SSH 通道
  ├─ 自定义协议扩展
  └─ 学习 SSH 协议原理

推荐路径：
  Paramiko（理解原理）→ Netmiko（高效工作）

七、最佳实践总结

Netmiko 使用原则：

  1. 连接管理
  ┌─ 用 ThreadPoolExecutor 实现并行
  ├─ 设置合理的 timeout（默认 30s）
  ├─ 用完调用 disconnect()
  └─ 捕获 NetmikoTimeoutException

  2. 命令执行
  ┌─ 查询命令用 send_command
  ├─ 配置命令用 send_config_set
  ├─ 长输出设置 read_timeout
  └─ 慢设备用 global_delay_factor

  3. 错误处理
  ┌─ 配置前备份当前配置
  ├─ 配置后验证关键参数
  ├─ 失败时记录完整日志
  └─ 准备回滚方案

  4. 安全实践
  ┌─ 凭证从环境变量读取（非硬编码）
  ├─ 使用 SFTP 替代 TFTP
  ├─ 记录变更审计日志
  └─ 变更操作添加确认步骤

下篇预告：第305篇 — NAPALM 统一网络设备管理接口，将介绍如何通过 NAPALM 用一套 API 管理多厂商网络设备，实现配置的 Getter/Replacer/Committer 模式。