---

name: netmiko-ssh-automation description: 安全的 Python Netmiko 模式，用于只读采集、有界批量 SSH、TextFSM 解析、受保护的配置变更、超时和网络自动化错误处理。 origin: community

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("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")

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

批量采集

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.")

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

审查清单

• 脚本是否标识了明确的设备清单来源？
• 凭据是否来自环境变量或保险库？
• 是否设置了连接和读取超时？
• 并发是否有界？
• 配置变更是否受干运行标志保护？
• 是否捕获了变更前后的证据？
• 是否有回滚计划？

限制

• 仅在任务明确匹配上述描述范围时使用此技能。
• 不要将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少所需输入、权限、安全边界或成功标准，请停下来要求澄清。