netmiko-ssh-automation

Netmiko SSH 自动化

在编写或审查使用 Netmiko 连接网络设备的 Python 自动化脚本时使用此技能。保持默认路径为只读；配置变更需要单独的变更窗口、同行评审和回滚计划。

使用场景

• 在路由器、交换机或防火墙上收集 show 命令的输出。
• 为接口、路由或配置证据编写小型审计脚本。
• 为网络 SSH 脚本添加超时和异常处理。
• 当存在模板时，使用 TextFSM 解析命令输出。
• 在自动化脚本接触生产设备之前进行审查。

安全默认设置

• 从只读的 send_command() 收集开始。
• 保持库存清单小而明确；不要扫描整个地址范围。
• 使用环境变量、密码库或 getpass；切勿硬编码凭证。
• 设置连接和读取超时。
• 限制并发度，以避免过载老旧设备。
• 在 send_config_set() 之前需要明确的运算符标志。
• 在变更得到验证和批准之前，不要调用 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("用户名: "),
    "password": os.environ.get("NETMIKO_PASSWORD") or getpass("密码: "),
    "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("认证失败")
except NetmikoTimeoutException:
    print("SSH 连接超时")
except ReadTimeout:
    print("命令读取超时")

在示例中使用文档范围内的占位地址。将真实库存清单保存在被忽略的本地文件或秘密管理系统中。

批量采集

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("没有匹配的解析器模板；保存原始输出以供审查")
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("仅检查模式。候选命令:")
    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("在保存启动配置前验证行为。")

保存配置是一个独立的审批步骤。在生产环境中，包含回滚代码片段，并在变更记录中捕获变更前后的证据。

审查清单

• 脚本是否指明了明确的库存源？
• 凭证是否未出现在源代码、日志和异常消息中？
• 是否设置了 conn_timeout、auth_timeout 和命令级别 read_timeout？
• 故障是否按设备报告，而不是停止整个批次？
• 脚本是否避免了广泛扫描和无限制的并发？
• 配置变更是否位于仅检查模式或明确的运算符标志之后？
• save_config() 是否与初始推送分开，并与验证关联？

反模式

• 在源代码中硬编码密码、启用密码或私钥。
• 将发送配置命令作为默认代码路径。
• 对 CIDR 范围运行自动化，而不是经过审查的库存清单。
• 将完整的运行配置记录到未进行脱敏处理的共享系统。
• 将解析成功视为设备状态正确的证据。

另请参阅

• 技能: cisco-ios-patterns
• 技能: network-config-validation
• 技能: network-interface-health