交换机自动化巡检工具(基于Paramiko)
转载自:https://mp.weixin.qq.com/s/P5hkCBmJmGxMjM3FT5NbwA
前段时间《Python 的交换机批量的巡检脚本》是基于Paramiko开发,经过多位大佬的指点,我这次升级采用了Netmiko框架,可实现多厂商设备兼容。
核心功能
针对企业网络中大量交换机的运维需求,开发此自动化巡检工具。通过标准化检查流程,提升网络设备维护效率,降低人工操作失误风险。
✅ 多厂商设备支持(Cisco IOS/Huawei/H3C)
✅ 关键配置自动采集(版本信息/接口状态/运行配置)
✅ 智能日志记录与错误追踪
✅ 批量设备巡检支持
技术亮点
模块选型对比
-
• Paramiko(旧项目方案):基础SSH协议库,需自行处理设备交互逻辑(需手动维护会话状态/适配不同厂商指令/实现错误重试机制),适合深度定制场景
-
• Netmiko(当前方案):基于Paramiko二次开发的网络设备自动化框架,主要增强功能包括:
-
• 多厂商协议适配(内置Cisco、Huawei、H3C等更多设备)
• 会话生命周期管理(自动处理SSH连接超时与重连机制)
• 智能交互优化(自动处理分页显示、命令提示符匹配、异常恢复)
• 标准化输出处理(自动清理ANSI转义字符与多余空行)
• 批处理执行引擎(支持命令队列的原子化执行与结果收集)
代码架构解析
核心模块说明
- 设备连接管理:
# 使用ConnectHandler统一连接入口
with netmiko.ConnectHandler(
device_type=switch['device_type'],
ip=switch['ip'],
# ... 认证参数
) as net_connect:
- 设备类型适配:
# 通过device_type自动匹配厂商协议
if switch['device_type'] in ['cisco_ios']:
net_connect.enable() # Cisco专属特权模式
- 批处理执行引擎:
# 标准化命令执行流程
for command in commands:
output = net_connect.send_command(command)
logging.info(f'### Output for {command}...')
- • 采用Netmiko实现多厂商设备兼容
- • 异常处理机制覆盖常见网络故障场景
- • 可视化实时日志输出
- • 配置信息加密存储
使用说明
环境要求
pip install netmiko==4.2.0
配置文件示例(switches.txt)
# 格式:IP,用户名,登录密码,enable密码,设备类型
192.168.1.1,admin,password123,enablepass,cisco_ios
10.0.0.1,huawei_admin,Huawei@123,,huawei
运行命令
python switch_inspection.py
典型应用场景
- 日常网络设备健康检查
- 变更配置前基准状态备份
- 故障排查时快速获取设备信息
- 合规性审计数据采集
设备类型支持
厂商 | 设备类型标识 | 适用场景 |
---|---|---|
Cisco | cisco_ios | IOS系统交换机(Catalyst系列) |
Cisco | cisco_nxos | NX-OS系统交换机(Nexus系列) |
Huawei | huawei | VRP系统交换机(S系列) |
H3C | h3c | Comware V7系统交换机 |
Juniper | juniper_junos | Junos系统交换机(EX系列) |
注:具体设备类型请参考Netmiko官方文档
注意事项
⚠️ 建议在隔离环境测试后再正式使用
⚠️ 配置文件需设置访问权限
⚠️ 确保执行主机与设备网络可达
完整代码
import netmiko
import logging
import re
logging.basicConfig(filename='switch_inspection.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 定义巡检交换机的函数
definspect_switch(switch):
try:
# 建立与交换机的连接
with netmiko.ConnectHandler(
device_type=switch['device_type'],
ip=switch['ip'],
username=switch['username'],
password=switch['password'],
secret=switch['enable_password']
) as net_connect:
# 记录成功连接的日志
logging.info(f'成功连接到 {switch["ip"]}')
# 如果是 Cisco IOS 设备,进入 enable 模式
if switch['device_type'] in ['cisco_ios']:
net_connect.enable()
# 记录进入 enable 模式的日志
logging.info(f'已进入 {switch["ip"]} 的 enable 模式')
else:
# 记录非 Cisco IOS 设备跳过进入 enable 模式的日志
logging.info(f'{switch["device_type"]} 设备不需要进入 enable 模式,跳过此步骤。')
# 定义要执行的命令列表
commands = [
'show version',
'show interfaces status',
'show running-config'
]
# 遍历命令列表并执行
for command in commands:
output = net_connect.send_command(command)
# 记录命令输出的日志
logging.info(f'### Output for {command} on {switch["ip"]} ###\n{output}')
# 打印命令输出
print(f'### Output for {command} on {switch["ip"]} ###')
print(output)
# 记录断开连接的日志
logging.info(f'已断开与 {switch["ip"]} 的连接')
except netmiko.NetMikoTimeoutException as e:
# 记录连接超时的错误日志
logging.error(f'连接到 {switch["ip"]} 超时: {e}')
# 打印连接超时的错误信息
print(f'Error connecting to {switch["ip"]}: 连接超时')
except netmiko.NetMikoAuthenticationException as e:
# 记录认证失败的错误日志
logging.error(f'连接到 {switch["ip"]} 认证失败: {e}')
# 打印认证失败的错误信息
print(f'Error connecting to {switch["ip"]}: 认证失败')
except Exception as e:
# 记录未知错误的日志
logging.error(f'连接到 {switch["ip"]} 发生未知错误: {e}')
# 打印未知错误的信息
print(f'Error connecting to {switch["ip"]}: {e}')
defread_switches_from_txt(file_path):
switches = []
withopen(file_path, 'r', encoding='utf-8') as f:
for line in f:
if line.strip() andnot line.startswith('#'):
ip, username, password, enable_password, device_type = line.strip().split(',')
switches.append({
'ip': ip,
'username': username,
'password': password,
'enable_password': enable_password,
'device_type': device_type
})
return switches
# 主程序入口
if __name__ == "__main__":
# 定义交换机信息文件的路径
file_path = "switches.txt"
# 从文件中读取交换机信息
switches = read_switches_from_txt(file_path)
# 遍历交换机列表并进行巡检
for switch in switches:
inspect_switch(switch)
本作品采用 知识共享署名-相同方式共享 4.0 国际许可协议 进行许可。