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_timeout、auth_timeout以及命令的read_timeout? - 单台设备失败时,是否能按设备上报异常且不影响整个批次的继续执行?
- 脚本是否避免了无差别的网段扫描和无限制的并发连接?
- 配置变更是否受控于 Dry-Run 模式或显式的操作员开关标志?
save_config()是否独立于初始下发动作,并且严格与验证步骤绑定?
反模式(应避免的做法)
- 在源码中硬编码密码、enable 口令或私钥。
- 将下发配置命令设为默认执行路径。
- 直接针对整个 CIDR 网段执行自动化,而非使用经评审的设备清单。
- 未经脱敏处理就将完整运行配置(running-config)日志输出到共享系统中。
- 将解析器解析成功盲目等同于设备状态完全正确。
参见
- Skill:
cisco-ios-patterns - Skill:
network-config-validation - Skill:
network-interface-health






