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

关键词：Netmiko、批量配置、多厂商兼容、配置模板、批量变更、幂等性

一、引言

第303篇中，我们使用Paramiko直接管理SSH连接和Shell交互。然而，当需要对接华为、思科、Juniper等多品牌设备时，每个厂商的CLI提示符、配置模式、命令语法各不相同，手工编写适配代码的工作量巨大。

Netmiko在Paramiko之上构建了一层设备抽象层，内置了100+厂商的平台驱动（Driver），自动识别提示符、自动进入配置模式、自动处理分页输出。这使得网络工程师可以用同一套代码管理异构设备，将批量配置的效率提升到一个新的高度。

二、Netmiko 安装与设备类型识别

2.1 安装

pip install netmiko

2.2 设备平台类型速查

Netmiko通过device_type参数选择对应的驱动。常用厂商映射：

from netmiko import ConnectHandler

# 核心连接参数结构
device_params = {
    "device_type": "huawei",        # 必须指定
    "host": "192.168.1.1",
    "username": "admin",
    "password": "H@wa123",
    "port": 22,                     # 默认22
    "secret": "H@wa123",            # enable密码（如需要）
    "timeout": 30,                  # 连接超时（秒）
    "session_timeout": 30,          # 单次操作超时（秒）
    "auth_timeout": 10,             # 认证超时（秒）
    "banner_timeout": 15,           # Banner超时（秒）
}

# 建立连接
net_connect = ConnectHandler(**device_params)

# 执行单条显示命令
output = net_connect.send_command("display version")
print(output)

# 关闭连接
net_connect.disconnect()

三、发送命令的三种方式

3.1 send_command：执行单条命令（只读）

from netmiko import ConnectHandler

net_connect = ConnectHandler(**device_params)

# 基本用法
output = net_connect.send_command("display ip interface brief")

# 使用expect_string等待特定字符串出现
output = net_connect.send_command(
    "ping 10.0.0.1",
    expect_string=r"#.*$",      # 等待回到命令行提示符
    delay_factor=2              # 延迟倍率（慢设备增大此值）
)

# 长输出：使用read_timeout延长读取时间
output = net_connect.send_command(
    "display ip routing-table",
    read_timeout=120             # 等待120秒读取完整输出
)

net_connect.disconnect()

3.2 send_config_set：下发配置命令

from netmiko import ConnectHandler

net_connect = ConnectHandler(**device_params)

# 方式1：传入命令列表
config_commands = [
    "vlan 100",
    "description SERVER_VLAN",
    "quit",
    "interface GigabitEthernet0/0/10",
    "port link-type access",
    "port default vlan 100",
]

result = net_connect.send_config_set(config_commands)
print(result)

# 方式2：传入多行字符串
config_text = """
vlan 200
 description BACKUP_VLAN
quit
interface GigabitEthernet0/0/11
 port link-type access
 port default vlan 200
"""

result = net_connect.send_config_set(config_text.strip().splitlines())
print(result)

net_connect.disconnect()

3.3 save_config：保存配置

# 下发配置后立即保存
net_connect.send_config_set(config_commands)
net_connect.save_config()   # 自动执行厂商对应的保存命令
# 华为: save force
# 思科: copy running-config startup-config
# Juniper: commit

四、批量设备配置实战

4.1 标准批量配置流程

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

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s'
)

# 设备清单
devices = [
    {"host": "192.168.1.1", "username": "admin", "password": "H@wa123"},
    {"host": "192.168.1.2", "username": "admin", "password": "H@wa123"},
    {"host": "192.168.1.3", "username": "admin", "password": "H@wa123"},
    {"host": "192.168.1.4", "username": "admin", "password": "H@wa123"},
    {"host": "192.168.1.5", "username": "admin", "password": "H@wa123"},
]

# 待下发配置
new_vlans = [
    "vlan 300",
    "description HR_VLAN",
    "quit",
    "vlan 301",
    "description FINANCE_VLAN",
    "quit",
]

def deploy_config(device):
    """在单台设备上部署配置"""
    device_params = {
        "device_type": "huawei",
        **device
    }

    try:
        logging.info(f"连接设备: {device['host']}")
        net_connect = ConnectHandler(**device_params)

        # 步骤1：执行前验证——检查VLAN是否已存在
        existing = net_connect.send_command("display vlan 300")
        if "VlanId300" in existing:
            logging.info(f"  VLAN 300已存在，跳过配置")
            net_connect.disconnect()
            return {"host": device["host"], "status": "skipped", "reason": "vlan_exists"}

        # 步骤2：下发配置
        logging.info(f"  下发配置到: {device['host']}")
        result = net_connect.send_config_set(new_vlans)

        # 步骤3：保存配置
        net_connect.save_config()
        logging.info(f"  配置已保存: {device['host']}")

        # 步骤4：验证配置
        verify = net_connect.send_command("display vlan 300")
        if "VlanId300" in verify:
            logging.info(f"  验证通过: {device['host']}")
            return {"host": device["host"], "status": "success"}
        else:
            logging.warning(f"  验证失败: {device['host']}")
            return {"host": device["host"], "status": "failed", "reason": "verify_failed"}

    except Exception as e:
        logging.error(f"  异常: {device['host']} - {e}")
        return {"host": device["host"], "status": "error", "reason": str(e)}
    finally:
        try:
            net_connect.disconnect()
        except:
            pass


# 使用线程池并发执行（最多3台并行）
results = []
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = {executor.submit(deploy_config, dev): dev for dev in devices}

    for future in as_completed(futures):
        result = future.result()
        results.append(result)

# 汇总报告
print("\n" + "=" * 50)
print("批量部署报告")
print("=" * 50)
success = sum(1 for r in results if r["status"] == "success")
failed = sum(1 for r in results if r["status"] == "failed")
errors = sum(1 for r in results if r["status"] == "error")
skipped = sum(1 for r in results if r["status"] == "skipped")

print(f"成功: {success}")
print(f"跳过: {skipped}")
print(f"失败: {failed}")
print(f"异常: {errors}")

4.2 从CSV读取设备清单

import csv

def load_devices_from_csv(csv_path):
    """从CSV文件加载设备清单"""
    devices = []
    with open(csv_path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            devices.append({
                "host": row["ip"],
                "username": row.get("username", "admin"),
                "password": row.get("password", "default"),
                "device_type": row.get("device_type", "huawei"),
            })
    return devices

# devices.csv 格式:
# ip,username,password,device_type
# 192.168.1.1,admin,H@wa123,huawei
# 192.168.1.2,admin,C1sc0!,cisco_ios
# 192.168.1.3,admin,Jun1per!,juniper_junos

devices = load_devices_from_csv("devices.csv")

五、配置回滚与幂等性

5.1 变更前备份 + 自动回滚

from netmiko import ConnectHandler
from datetime import datetime
import os

class SafeConfigDeployer:
    """安全配置部署器：支持备份与回滚"""

    def __init__(self, device_params):
        self.device_params = device_params
        self.net_connect = None
        self.backup_file = None

    def connect(self):
        self.net_connect = ConnectHandler(**self.device_params)

    def backup_config(self, backup_dir="./backups"):
        """备份当前运行配置"""
        os.makedirs(backup_dir, exist_ok=True)

        hostname = self.net_connect.find_prompt().strip()
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.backup_file = f"{backup_dir}/{hostname}_{ts}_rollback.cfg"

        config = self.net_connect.send_command("display current-configuration")
        with open(self.backup_file, "w", encoding="utf-8") as f:
            f.write(f"# 备份时间: {ts}\n")
            f.write(f"# 设备: {hostname}\n")
            f.write(config)

        logging.info(f"配置已备份: {self.backup_file}")
        return self.backup_file

    def deploy_and_verify(self, config_commands, verify_commands):
        """部署配置并验证"""
        # 下发配置
        self.net_connect.send_config_set(config_commands)
        self.net_connect.save_config()

        # 验证
        for cmd, expected in verify_commands:
            output = self.net_connect.send_command(cmd)
            if expected not in output:
                raise VerificationError(
                    f"验证失败: '{cmd}' 输出中未找到 '{expected}'"
                )

    def rollback(self):
        """回滚到备份配置"""
        if not self.backup_file:
            raise RuntimeError("无备份文件，无法回滚")

        logging.warning("正在回滚配置...")

        # 读取备份文件中的配置
        with open(self.backup_file, "r", encoding="utf-8") as f:
            backup_config = f.read()

        # 通过Shell交互模式回滚（华为VRP）
        shell = self.net_connect.config_mode()
        # 此处简化处理，实际需解析配置并反向操作
        self.net_connect.exit_config_mode()

        logging.info("回滚完成")

    def disconnect(self):
        if self.net_connect:
            self.net_connect.disconnect()


# 使用示例
deployer = SafeConfigDeployer({
    "device_type": "huawei",
    "host": "192.168.1.1",
    "username": "admin",
    "password": "H@wa123"
})

try:
    deployer.connect()
    deployer.backup_config()

    deployer.deploy_and_verify(
        config_commands=["vlan 400", "description TEST", "quit"],
        verify_commands=[
            ("display vlan 400", "VlanId400"),
            ("display current-configuration | include TEST", "TEST"),
        ]
    )
    logging.info("部署成功")

except Exception as e:
    logging.error(f"部署失败: {e}，执行回滚")
    deployer.rollback()
finally:
    deployer.disconnect()

5.2 幂等性检查器

class IdempotentChecker:
    """配置幂等性检查器"""

    def __init__(self, net_connect):
        self.nc = net_connect

    def check_vlan_exists(self, vlan_id):
        """检查VLAN是否已存在"""
        output = self.nc.send_command(f"display vlan {vlan_id}")
        return f"VlanId{vlan_id}" in output

    def check_interface_description(self, interface, description):
        """检查接口描述是否为期望值"""
        output = self.nc.send_command(f"display current-configuration interface {interface}")
        return description in output

    def check_bgp_neighbor(self, neighbor_ip):
        """检查BGP邻居是否已配置"""
        output = self.nc.send_command("display bgp peer")
        return neighbor_ip in output


# 使用示例：只在需要时才下发配置
checker = IdempotentChecker(net_connect)

if not checker.check_vlan_exists(400):
    net_connect.send_config_set(["vlan 400", "description TEST_VLAN", "quit"])
    logging.info("VLAN 400已创建")
else:
    logging.info("VLAN 400已存在，跳过")

六、故障案例：Netmiko delay_factor 引发的超时

6.1 故障现象

批量配置脚本在连接老旧的CE6850交换机时频繁超时：

netmiko.exceptions.NetmikoTimeoutException:
Timed-out reading channel, data not available.

6.2 排查过程

步骤1：确认设备本身正常

# 手动SSH登录，命令执行正常，无延迟
ssh admin@10.10.10.1
<CE6850> display version         # 立即返回

步骤2：降低线程并发数 将max_workers从10降至2，超时仍然发生。

步骤3：分析Netmiko日志 开启调试日志：

import logging
logging.getLogger("paramiko").setLevel(logging.DEBUG)
logging.getLogger("netmiko").setLevel(logging.DEBUG)

日志显示：命令已发送，但Netmiko未能识别到命令返回后的提示符。

6.3 根因定位

Netmiko通过正则表达式匹配设备提示符来判断命令是否执行完毕。老旧VRP版本的提示符格式为：

<CE6850>

但设备在CPU负载高时，提示符前会出现多余空格或延迟，导致匹配失败。

6.4 解决方案

from netmiko import ConnectHandler

net_connect = ConnectHandler(
    device_type="huawei",
    host="192.168.1.1",
    username="admin",
    password="H@wa123",
    # 关键参数调整：
    delay_factor=2,             # 增大延迟倍率（默认1）
    global_delay_factor=2,      # 全局延迟倍率
    fast_cli=False,             # 禁用快速CLI（老旧设备可能不兼容）
    session_timeout=60,         # 增加会话超时
)

# 对特定慢命令使用额外的delay_factor
output = net_connect.send_command(
    "display ip routing-table",
    delay_factor=3,
    read_timeout=120
)

七、Netmiko 高级技巧

7.1 自定义设备驱动

当内置驱动不匹配时，可基于现有驱动扩展：

from netmiko.cisco.cisco_ios import CiscoIosBase

class CustomHuaweiDriver(CiscoIosBase):
    """基于CiscoIOS驱动的自定义华为驱动"""

    def session_preparation(self):
        self.ansi_escape_codes = True
        self._test_channel_read(pattern=r"[<\[]\w+[>\]]")
        self.set_base_prompt()

    def config_mode(self, config_command="system-view"):
        return super().config_mode(config_command=config_command)

    def exit_config_mode(self):
        return super().exit_config_mode()

    def save_config(self, cmd="save force"):
        return super().save_config(cmd=cmd)

# 注册自定义驱动
from netmiko import Dispatch
Dispatch["custom_huawei"] = CustomHuaweiDriver

# 使用
net_connect = ConnectHandler(
    device_type="custom_huawei",
    host="192.168.1.1",
    username="admin",
    password="H@wa123"
)

7.2 配置差异检查

from netmiko import ConnectHandler

net_connect = ConnectHandler(**device_params)

# 获取运行配置
running = net_connect.send_command("display current-configuration")

# 获取启动配置
startup = net_connect.send_command("display saved-configuration")

# 比对
running_lines = set(running.splitlines())
startup_lines = set(startup.splitlines())

new_in_running = running_lines - startup_lines
removed_from_startup = startup_lines - running_lines

if new_in_running:
    print(f"运行配置中有 {len(new_in_running)} 行未保存的变更")
    for line in list(new_in_running)[:10]:
        print(f"  + {line}")

if not new_in_running and not removed_from_startup:
    print("运行配置与启动配置一致")

net_connect.disconnect()

八、总结

思考题

1. Netmiko如何处理不同厂商的配置模式差异？如果目标设备是Nokia SR OS，send_config_set下发的第一条命令和思科IOS有何不同？
2. 为什么批量配置时需要"幂等性检查"？如果一个脚本每天定时执行一次，幂等性缺失会导致什么后果？
3. 设计一个批量BGP邻居配置方案：要求支持灰度发布（先10%设备、再50%、最后100%），每批次完成后自动检查BGP状态，异常则中止后续批次。

下篇预告

第305篇：NAPALM——统一网络设备管理接口。屏蔽厂商差异，用同一套API管理华为、思科、Juniper设备的配置、状态和路由信息。