netmiko-ssh-automation

netmiko-ssh-automation

热门

安全的 Python Netmiko 实战范例,涵盖只读信息采集、限流批量 SSH 连接、TextFSM 结构化解析、带防护机制的配置变更、超时设置以及网络自动化错误处理。

24万Star
3.6万Fork
更新于 2026/8/2
SKILL.md
只读
名称
netmiko-ssh-automation
描述

安全的 Python Netmiko 实战范例,涵盖只读信息采集、限流批量 SSH 连接、TextFSM 结构化解析、带防护机制的配置变更、超时设置以及网络自动化错误处理。

Netmiko SSH Automation

在编写或评审基于 Netmiko 连接网络设备的 Python 自动化脚本时使用此 Skill。请保持默认执行路径为“只读”模式;配置变更必须具备独立的变更窗口、同行评审(Peer Review)以及回滚方案。

适用场景

  • 跨路由器、交换机或防火墙采集 show 命令输出。
  • 编写小型审计脚本,用于收集接口、路由或配置凭证。
  • 为网络 SSH 脚本添加超时设置与异常处理逻辑。
  • 在有对应模板时,使用 TextFSM 解析命令输出。
  • 在自动化脚本上线生产设备前进行代码审查。

安全默认规范

  • 优先使用只读的 send_command() 进行数据采集。
  • 保持 Device Inventory(设备清单)精简且明确,切勿对整段 IP 地址进行无差别网段扫描。
  • 使用环境变量、密钥管理工具(Vault)或 getpass 获取凭证,严禁硬编码密码。
  • 必须显式设置连接超时与读取超时。
  • 限制并发数量,避免压垮老旧设备。
  • 在执行 send_config_set() 前,必须要求显式的操作员标志(Operator Flag)。
  • 在变更未经过验证和批准前,切勿调用 save_config()

只读连接范式

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

device = {
    "device_type": "cisco_ios",
    "host": "192.0.2.10",
    "username": os.environ.get("NETMIKO_USERNAME") or input("Username: "),
    "password": os.environ.get("NETMIKO_PASSWORD") or getpass("Password: "),
    "secret": os.environ.get("NETMIKO_ENABLE_SECRET"),
    "conn_timeout": 10,
    "auth_timeout": 20,
    "banner_timeout": 15,
    "read_timeout_override": 30,
}

try:
    with ConnectHandler(**device) as conn:
        if device.get("secret") and not conn.check_enable_mode():
            conn.enable()
        output = conn.send_command("show ip interface brief", read_timeout=30)
        print(output)
except NetmikoAuthenticationException:
    print("Authentication failed")
except NetmikoTimeoutException:
    print("SSH connection timed out")
except ReadTimeout:
    print("Command read timed out")

示例中请统一使用文档专用保留网段的占位 IP 地址。实际的设备清单应存放在被 gitignore 的本地文件或受密钥管理系统护航的环境中。

批量采集

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any

def collect_show(device: dict[str, Any], command: str) -> dict[str, Any]:
    host = device["host"]
    try:
        with ConnectHandler(**device) as conn:
            output = conn.send_command(command, read_timeout=45)
        return {"host": host, "ok": True, "output": output}
    except (NetmikoAuthenticationException, NetmikoTimeoutException, ReadTimeout) as exc:
        return {"host": host, "ok": False, "error": type(exc).__name__}

results = []
with ThreadPoolExecutor(max_workers=8) as pool:
    futures = [pool.submit(collect_show, device, "show version") for device in devices]
    for future in as_completed(futures):
        results.append(future.result())

除非已确认设备规模及 AAA 认证系统能够承受更高的并发连接,否则请保持 max_workers 在较低水平。

结构化解析

Netmiko 支持调用 TextFSM、TTP 或 Genie 来解析支持的命令输出。请将解析器的输出视为一种优化手段,而非唯一的凭证存证路径。

with ConnectHandler(**device) as conn:
    parsed = conn.send_command(
        "show ip interface brief",
        use_textfsm=True,
        raise_parsing_error=False,
        read_timeout=30,
    )

if isinstance(parsed, str):
    print("No parser template matched; store raw output for review")
else:
    for row in parsed:
        print(row)

如果解析结果直接影响关键阻断逻辑,请务必在保留解析结果的同时留存原始命令输出,以便运维人员比对核查。

带防护机制的配置范式

import os

commands = [
    "interface GigabitEthernet0/1",
    "description CHANGE-1234 UPLINK-TO-CORE",
]

apply_changes = os.environ.get("APPLY_NETWORK_CHANGES") == "1"

if not apply_changes:
    print("Dry run only. Candidate commands:")
    print("\n".join(commands))
else:
    with ConnectHandler(**device) as conn:
        conn.enable()
        before = conn.send_command("show running-config interface GigabitEthernet0/1")
        output = conn.send_config_set(commands)
        after = conn.send_command("show running-config interface GigabitEthernet0/1")
        print(before)
        print(output)
        print(after)
        print("Verify behavior before saving startup config.")

保存配置(Save Config)属于单独的审批步骤。在生产环境中,请在变更记录中附带回滚命令片段,并完整保存变更前的快照与变更后的凭证。

评审清单

  • 脚本是否指定了明确的设备清单来源?
  • 源代码、日志及异常信息中是否已完全抹去凭据信息?
  • 是否已设置 conn_timeoutauth_timeout 以及命令的 read_timeout
  • 单台设备失败时,是否能按设备上报异常且不影响整个批次的继续执行?
  • 脚本是否避免了无差别的网段扫描和无限制的并发连接?
  • 配置变更是否受控于 Dry-Run 模式或显式的操作员开关标志?
  • save_config() 是否独立于初始下发动作,并且严格与验证步骤绑定?

反模式(应避免的做法)

  • 在源码中硬编码密码、enable 口令或私钥。
  • 将下发配置命令设为默认执行路径。
  • 直接针对整个 CIDR 网段执行自动化,而非使用经评审的设备清单。
  • 未经脱敏处理就将完整运行配置(running-config)日志输出到共享系统中。
  • 将解析器解析成功盲目等同于设备状态完全正确。

参见

  • Skill: cisco-ios-patterns
  • Skill: network-config-validation
  • Skill: network-interface-health