netmiko-ssh-automation

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Netmiko SSH Automation

Netmiko SSH自动化

Use this skill when writing or reviewing Python automation that connects to network devices with Netmiko. Keep the default path read-only; config changes need a separate change window, peer review, and rollback plan.
当你编写或审查使用Netmiko连接网络设备的Python自动化脚本时,可以参考本技能。默认流程保持只读;配置变更需要单独的变更窗口、同行评审和回滚计划。

When to Use

适用场景

  • Collecting
    show
    command output across routers, switches, or firewalls.
  • Building a small audit script for interface, routing, or config evidence.
  • Adding timeouts and exception handling to network SSH scripts.
  • Parsing command output with TextFSM when a template exists.
  • Reviewing automation before it touches production devices.
  • 在路由器、交换机或防火墙之间收集
    show
    命令输出。
  • 编写用于接口、路由或配置验证的小型审计脚本。
  • 为网络SSH脚本添加超时和异常处理机制。
  • 在已有模板的情况下,使用TextFSM解析命令输出。
  • 在自动化脚本接触生产设备前进行审查。

Safety Defaults

安全默认规则

  • Start with read-only
    send_command()
    collection.
  • Keep inventory small and explicit; do not sweep whole address ranges.
  • Use environment variables, a vault, or
    getpass
    ; never hardcode credentials.
  • Set connection and read timeouts.
  • Limit concurrency so older devices are not overloaded.
  • Require an explicit operator flag before
    send_config_set()
    .
  • Do not call
    save_config()
    until the change has been verified and approved.
  • 从只读的
    send_command()
    数据收集开始。
  • 保持设备清单规模小且明确;不要扫描整个地址范围。
  • 使用环境变量、密钥管理工具或
    getpass
    ;绝对不要硬编码凭证。
  • 设置连接和读取超时。
  • 限制并发数,避免旧设备过载。
  • 在执行
    send_config_set()
    前需要显式的操作标志。
  • 只有在变更经过验证和批准后,才能调用
    save_config()

Read-Only Connection Pattern

只读连接模式

python
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")
Use placeholder addresses from documentation ranges in examples. Keep real inventory in an ignored local file or a secrets-managed system.
python
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")
示例中使用文档范围内的占位地址。将真实设备清单存储在本地忽略文件或密钥管理系统中。

Batch Collection

批量收集

python
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())
Keep
max_workers
low unless the device estate and AAA systems are known to handle higher connection volume.
python
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
数值较低。

Structured Parsing

结构化解析

Netmiko can ask TextFSM, TTP, or Genie to parse supported command output. Treat parser output as an optimization, not the only evidence path.
python
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)
If parsing drives a blocking decision, keep the raw command output alongside the parsed result so an operator can inspect mismatches.
Netmiko可以调用TextFSM、TTP或Genie来解析支持的命令输出。将解析器输出视为优化手段,而非唯一的验证路径。
python
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)
如果解析结果会影响阻塞性决策,请将原始命令输出与解析结果一同保存,以便操作人员检查不匹配的情况。

Guarded Config Pattern

受控配置模式

python
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.")
Saving the config is a separate approval step. In production, include a rollback snippet and capture before/after evidence in the change record.
python
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.")
保存配置是一个单独的审批步骤。在生产环境中,需包含回滚代码片段,并在变更记录中保存变更前后的验证证据。

Review Checklist

审查清单

  • Does the script identify an explicit inventory source?
  • Are credentials absent from source, logs, and exception messages?
  • Are
    conn_timeout
    ,
    auth_timeout
    , and command
    read_timeout
    set?
  • Are failures reported per device without stopping the whole batch?
  • Does the script avoid broad scans and unbounded concurrency?
  • Are config changes behind a dry-run or explicit operator flag?
  • Is
    save_config()
    separate from the initial push and tied to verification?
  • 脚本是否指定了明确的设备清单来源?
  • 凭证是否未出现在源代码、日志和异常信息中?
  • 是否设置了
    conn_timeout
    auth_timeout
    和命令
    read_timeout
  • 是否会按设备报告失败情况,而不会终止整个批量任务?
  • 脚本是否避免了大范围扫描和无限制并发?
  • 配置变更是否处于试运行模式或需要显式的操作标志?
  • save_config()
    是否与初始推送分离,并与验证环节绑定?

Anti-Patterns

反模式

  • Hardcoding passwords, enable secrets, or private keys in source.
  • Sending config commands as the default code path.
  • Running automation against a CIDR range instead of a reviewed inventory.
  • Logging full running configs to shared systems without sanitization.
  • Treating parser success as proof that the device state is correct.
  • 在源代码中硬编码密码、启用密钥或私钥。
  • 将发送配置命令作为默认代码流程。
  • 针对CIDR范围运行自动化,而非经过审查的设备清单。
  • 在未经过清理的情况下,将完整的运行配置记录到共享系统中。
  • 将解析器成功视为设备状态正确的证明。

See Also

相关技能

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