
关于
安全的 Python Netmiko 模式,涵盖只读采集、有界批量 SSH、TextFSM 解析、受保护的配置变更、超时处理和网络自动化错误处理。
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.")
保存配置是单独的审批步骤。在生产中,包含回滚代码片段并在变更记录中捕获变更前后的证据。
审查清单
- 脚本是否标识了明确的设备清单来源?
- 凭据是否来自环境变量或保险库?
- 是否设置了连接和读取超时?
- 并发是否有界?
- 配置变更是否受干运行标志保护?
- 是否捕获了变更前后的证据?
- 是否有回滚计划?
限制
- 仅在任务明确匹配上述描述范围时使用此技能。
- 不要将输出视为环境特定验证、测试或专家审查的替代品。
- 如果缺少所需输入、权限、安全边界或成功标准,请停下来要求澄清。
兼容工具
Claude CodeCursor
标签
运维部署

