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