当前位置: 首页 > news >正文

Python:Netmiko实现网络设备巡检及配置备份

通过Python的第三方库Netmiko实现不同厂商网络设备的日常巡检及配置备份。

一、设备列表文件:JSON 文件

1、 我们先看一个示例

(1)拓扑

(2)脚本

import time from netmiko import ConnectHandler AR1 = { "host": "192.168.22.100", "username": "pytest", "password": "admin@123", "device_type": "huawei_vrp" } AR2 = { "host": "192.168.22.101", "username": "admin", "password": "pytest", "device_type": "huawei_vrp" } AR1_connect = ConnectHandler(**AR1) print("已成功登录设备 - " + AR1['host']) time.sleep(2) AR1_output = AR1_connect.send_command('display version') print(AR1_output) AR1_connect.disconnect() AR2_connect = ConnectHandler(**AR2) print("已成功登录设备 - " + AR2['host']) time.sleep(2) AR2_output = AR2_connect.send_command('display ip interface brief') print(AR2_output) AR2_connect.disconnect()

从上面的脚本可以看到,我们定义了两台路由器设备,并通过字典的形式保存了设备的信息。设备信息中:

"host": "192.168.22.100", # host 为设备的IP地址
"username": "pytest", # username 为SSH登录设备时使用的用户名
"password": "admin@123", # password 为用户密码
"device_type": "huawei_vrp" # device_type 为Netmiko所支持的设备类型

Netmiko 当前可支持绝大多数厂商设备,包括华为、H3C、TP-Link等,通过支持的设备类型,预设了一些指令。

send_command() :用户执行查询、保存等命令

AR1执行:display version

AR2执行:display ip interface brief

(3)结果


当有大量设备时,脚本中就需要定义很多台设备信息,每台设备都需要写一段连接执行的指令。下次再用时,还得对所有信息进行修改

2、设备列表文件:JSON文件

通过JSON文件,可以将大量的设备信息保存在一个文件当中,通过脚本来读取JSON文件中的设备信息。

设备列表采用JSON文件保存:

{
"name": "Layer3Switch-1", # 为设备名称,可自定义也可使用设备当前名称
"connection": {
"device_type": "huawei",
"host": "192.168.11.11",
"username": "python",
"password": "123"
}
}

cisco设备登录后需进入特权模式,所以JSON文件中会多一个enable_password的密码项。

JSON 文件示例

[ { "name": "HuaweiAR", "connection": { "device_type": "huawei", "host": "192.168.44.100", "username": "huaweipytest", "password": "HWpytest@123" } }, { "name": "CiscoRouter", "connection": { "device_type": "cisco_xe", "host": "192.168.44.102", "username": "ciscopytest", "password": "Pythontest@123" } }, { "name": "H3CAR", "connection": { "device_type": "hp_comware", "host": "192.168.44.101", "username": "h3cpytest", "password": "Pythontest@123" } } ]

二、脚本

1、模块导入

import os import json import logging import datetime from netmiko import ConnectHandler from netmiko.exceptions import NetMikoTimeoutException, NetMikoAuthenticationException

2、获取脚本所在目录

# 定义变量 SCRIPT_IDR,获取脚本所在目录,用于获取设备列表JSON文件,同时巡检信息和配置备份也将在此目录保存 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # 也可指定自定义JOSN文件所在目录路径,同样在脚本运行时也会在此目录生成巡检和配置备份目录 # SCRIPT_DIR = r"D:\BASE" # 以Windows系统为例 # --- 自定义输出根目录(脚本所在目录,输出文件将保存在 ./backups 和 ./reports) --- OUTPUT_BASE_DIR = SCRIPT_DIR

3、设备指令列表

根据不同厂商设备,提前设置好需要使用到的指令,如需其他指令可在 ‘inspection_commands’中添加

这里总结了华为、华三、Cisco,如有其他厂商设备可追加

# --- 厂商命令映射表 --- VENDOR_COMMANDS = { 'huawei': { 'device_type': 'huawei', 'disable_paging': 'screen-length 0 temporary', 'backup_command': 'display current-configuration', 'inspection_commands': [ 'display device', 'display cpu-usage', 'display memory-usage', 'display ip interface brief', 'display logbuffer' ] }, 'h3c': { 'device_type': 'hp_comware', 'disable_paging': 'screen-length disable', 'backup_command': 'display current-configuration', 'inspection_commands': [ 'display device', 'display cpu-usage', 'display memory', 'display ip interface brief', 'display logbuffer' ] }, 'cisco': { 'device_type': 'cisco_ios', 'disable_paging': 'terminal length 0', 'backup_command': 'show running-config', 'inspection_commands': [ 'show version', 'show processes cpu', 'show processes memory', 'show ip interface brief', 'show logging' ] } }

4、创建函数 load_devices_from_json(),用于从JSON文件中读取设备信息,并重新对设备信息进行格式化,并将所有设备的信息以列表的形式存放在变量 devices 中

Netmiko库可直接使用 JSON 文件中的设备信息。为了确保正确性定义了一个单独的函数出来。

如果 JSON 文件中有些设备的信息定义错误可通过函数获取到哪些设备的信息有误,以便发现错误进行修改。

def load_devices_from_json(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: raw_devices = json.load(f) devices = [] for item in raw_devices: device_name = item.get('name', 'Unknown') conn_info = item.get('connection', {}) device = { 'name': device_name, 'device_type': conn_info.get('device_type'), 'host': conn_info.get('host'), 'username': conn_info.get('username'), 'password': conn_info.get('password'), 'port': conn_info.get('port', 22), 'enable_password': conn_info.get('enable_password', ''), } if device['host'] and device['username'] and device['password']: devices.append(device) else: logging.warning(f"设备 {device_name} 缺少必要连接信息,已跳过。") logging.info(f"成功从 {filepath} 加载了 {len(devices)} 台设备。") return devices except FileNotFoundError: logging.error(f"错误:设备清单文件 {filepath} 未找到。") return [] except json.JSONDecodeError as e: logging.error(f"错误:{filepath} 文件 JSON 格式无效:{e}") return []

5、接下来就是巡检和配置备份部分,在这里分别定义了巡检和配置备份的函数。将JSON文件中预设置的指令进行格式化,并将结果进行保存。在这里我们通过“device_type”来判断设备的厂商来执行相应指令。

# --- 备份单台设备配置 --- def backup_device_config(conn, device_info, vendor, vendor_cmd): host = device_info['host'] device_name = device_info.get('name', host) try: if vendor == 'cisco': hostname_output = conn.send_command('show running-config | include hostname') else: hostname_output = conn.send_command('display current-configuration | include sysname') actual_hostname = hostname_output.strip().split()[-1] except Exception: actual_hostname = device_name logging.info(f"--- 正在备份设备 {device_name} ({host}) 的配置 ---") backup_output = conn.send_command(vendor_cmd['backup_command']) backup_dir = os.path.join(OUTPUT_BASE_DIR, "backups", vendor, str(datetime.date.today())) ensure_dir(backup_dir) filename = f"{backup_dir}/{actual_hostname}_{host}_{datetime.date.today()}.cfg" with open(filename, 'w', encoding='utf-8') as f: f.write(backup_output) logging.info(f"✓ 设备 {device_name} 的配置已备份至:{filename}") # --- 执行单台设备巡检 --- def inspect_device(conn, device_info, vendor, vendor_cmd): host = device_info['host'] device_name = device_info.get('name', host) try: if vendor == 'cisco': hostname_output = conn.send_command('show running-config | include hostname') else: hostname_output = conn.send_command('display current-configuration | include sysname') actual_hostname = hostname_output.strip().split()[-1] except Exception: actual_hostname = device_name report_dir = os.path.join(OUTPUT_BASE_DIR, "reports", vendor, str(datetime.date.today())) ensure_dir(report_dir) report_file = f"{report_dir}/{actual_hostname}_{host}_inspection.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write(f"设备 {actual_hostname} ({host}) 巡检报告 - {datetime.datetime.now()}\n") f.write(f"JSON 定义名称:{device_name}\n") f.write("=" * 60 + "\n\n") logging.info(f"--- 正在巡检设备 {device_name} ({host}) ---") for cmd in vendor_cmd['inspection_commands']: f.write(f">>> 执行命令: {cmd}\n") try: output = conn.send_command(cmd, delay_factor=2) f.write(output) except Exception as e: f.write(f"!!! 命令执行失败: {str(e)}\n") f.write("\n" + "-" * 40 + "\n\n") logging.info(f"✓ 设备 {device_name} 的巡检报告已生成至:{report_file}")

6、最后是SSH登录设备。定义 process_device()函数来进行远程登录,同样需要通过“device_type”来确定设备的厂商执行不同的登录需求。

def process_device(device_info): host = device_info['host'] device_name = device_info.get('name', host) device_type = device_info['device_type'] vendor = None for v, cmd_set in VENDOR_COMMANDS.items(): if cmd_set['device_type'] == device_type: vendor = v break if vendor is None: logging.error(f"设备 {device_name} ({host}) 的设备类型 {device_type} 不受支持,已跳过。") return vendor_cmd = VENDOR_COMMANDS[vendor] netmiko_device = { 'device_type': device_type, 'host': host, 'username': device_info['username'], 'password': device_info['password'], 'port': device_info.get('port', 22), 'enable_password': device_info.get('enable_password', ''), 'conn_timeout': 60, 'auth_timeout': 30, } try: logging.info(f"正在连接设备 {device_name} ({host})...") with ConnectHandler(**netmiko_device) as conn: if device_info.get('enable_password'): conn.enable() conn.send_command_timing(vendor_cmd['disable_paging']) backup_device_config(conn, device_info, vendor, vendor_cmd) inspect_device(conn, device_info, vendor, vendor_cmd) except NetMikoTimeoutException: logging.error(f"✗ 连接设备 {device_name} ({host}) 超时。") except NetMikoAuthenticationException: logging.error(f"✗ 设备 {device_name} ({host}) 认证失败。") except Exception as e: logging.error(f"✗ 处理设备 {device_name} ({host}) 时发生未知错误: {str(e)}")

7、所有的准备工作完成后,便是将前期所有的操作进行整合,完成想要达到的目的。完整代码如下:

import os import json import logging import datetime from netmiko import ConnectHandler from netmiko.exceptions import NetMikoTimeoutException, NetMikoAuthenticationException # --- 获取脚本所在目录 --- SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) # --- 配置日志 --- logging.basicConfig( format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO ) # --- 自定义输出根目录(脚本所在目录,输出文件将保存在 ./backups 和 ./reports) --- OUTPUT_BASE_DIR = SCRIPT_DIR # --- 厂商命令映射表 --- VENDOR_COMMANDS = { 'huawei': { 'device_type': 'huawei', 'disable_paging': 'screen-length 0 temporary', 'backup_command': 'display current-configuration', 'inspection_commands': [ 'display device', 'display cpu-usage', 'display memory-usage', 'display ip interface brief', 'display logbuffer' ] }, 'h3c': { 'device_type': 'hp_comware', 'disable_paging': 'screen-length disable', 'backup_command': 'display current-configuration', 'inspection_commands': [ 'display device', 'display cpu-usage', 'display memory', 'display ip interface brief', 'display logbuffer' ] }, 'cisco': { 'device_type': 'cisco_ios', 'disable_paging': 'terminal length 0', 'backup_command': 'show running-config', 'inspection_commands': [ 'show version', 'show processes cpu', 'show processes memory', 'show ip interface brief', 'show logging' ] } } # --- 辅助函数:创建目录 --- def ensure_dir(directory): if not os.path.exists(directory): os.makedirs(directory) # --- 从 JSON 文件加载设备列表 --- def load_devices_from_json(filepath): try: with open(filepath, 'r', encoding='utf-8') as f: raw_devices = json.load(f) devices = [] for item in raw_devices: device_name = item.get('name', 'Unknown') conn_info = item.get('connection', {}) device = { 'name': device_name, 'device_type': conn_info.get('device_type'), 'host': conn_info.get('host'), 'username': conn_info.get('username'), 'password': conn_info.get('password'), 'port': conn_info.get('port', 22), 'enable_password': conn_info.get('enable_password', ''), } if device['host'] and device['username'] and device['password']: devices.append(device) else: logging.warning(f"设备 {device_name} 缺少必要连接信息,已跳过。") logging.info(f"成功从 {filepath} 加载了 {len(devices)} 台设备。") return devices except FileNotFoundError: logging.error(f"错误:设备清单文件 {filepath} 未找到。") return [] except json.JSONDecodeError as e: logging.error(f"错误:{filepath} 文件 JSON 格式无效:{e}") return [] # --- 备份单台设备配置 --- def backup_device_config(conn, device_info, vendor, vendor_cmd): host = device_info['host'] device_name = device_info.get('name', host) try: if vendor == 'cisco': hostname_output = conn.send_command('show running-config | include hostname') else: hostname_output = conn.send_command('display current-configuration | include sysname') actual_hostname = hostname_output.strip().split()[-1] except Exception: actual_hostname = device_name logging.info(f"--- 正在备份设备 {device_name} ({host}) 的配置 ---") backup_output = conn.send_command(vendor_cmd['backup_command']) backup_dir = os.path.join(OUTPUT_BASE_DIR, "backups", vendor, str(datetime.date.today())) ensure_dir(backup_dir) filename = f"{backup_dir}/{actual_hostname}_{host}_{datetime.date.today()}.cfg" with open(filename, 'w', encoding='utf-8') as f: f.write(backup_output) logging.info(f"✓ 设备 {device_name} 的配置已备份至:{filename}") # --- 执行单台设备巡检 --- def inspect_device(conn, device_info, vendor, vendor_cmd): host = device_info['host'] device_name = device_info.get('name', host) try: if vendor == 'cisco': hostname_output = conn.send_command('show running-config | include hostname') else: hostname_output = conn.send_command('display current-configuration | include sysname') actual_hostname = hostname_output.strip().split()[-1] except Exception: actual_hostname = device_name report_dir = os.path.join(OUTPUT_BASE_DIR, "reports", vendor, str(datetime.date.today())) ensure_dir(report_dir) report_file = f"{report_dir}/{actual_hostname}_{host}_inspection.txt" with open(report_file, 'w', encoding='utf-8') as f: f.write(f"设备 {actual_hostname} ({host}) 巡检报告 - {datetime.datetime.now()}\n") f.write(f"JSON 定义名称:{device_name}\n") f.write("=" * 60 + "\n\n") logging.info(f"--- 正在巡检设备 {device_name} ({host}) ---") for cmd in vendor_cmd['inspection_commands']: f.write(f">>> 执行命令: {cmd}\n") try: output = conn.send_command(cmd, delay_factor=2) f.write(output) except Exception as e: f.write(f"!!! 命令执行失败: {str(e)}\n") f.write("\n" + "-" * 40 + "\n\n") logging.info(f"✓ 设备 {device_name} 的巡检报告已生成至:{report_file}") # --- 处理单台设备的主流程 --- def process_device(device_info): host = device_info['host'] device_name = device_info.get('name', host) device_type = device_info['device_type'] vendor = None for v, cmd_set in VENDOR_COMMANDS.items(): if cmd_set['device_type'] == device_type: vendor = v break if vendor is None: logging.error(f"设备 {device_name} ({host}) 的设备类型 {device_type} 不受支持,已跳过。") return vendor_cmd = VENDOR_COMMANDS[vendor] netmiko_device = { 'device_type': device_type, 'host': host, 'username': device_info['username'], 'password': device_info['password'], 'port': device_info.get('port', 22), 'enable_password': device_info.get('enable_password', ''), 'conn_timeout': 60, 'auth_timeout': 30, } try: logging.info(f"正在连接设备 {device_name} ({host})...") with ConnectHandler(**netmiko_device) as conn: if device_info.get('enable_password'): conn.enable() conn.send_command_timing(vendor_cmd['disable_paging']) backup_device_config(conn, device_info, vendor, vendor_cmd) inspect_device(conn, device_info, vendor, vendor_cmd) except NetMikoTimeoutException: logging.error(f"✗ 连接设备 {device_name} ({host}) 超时。") except NetMikoAuthenticationException: logging.error(f"✗ 设备 {device_name} ({host}) 认证失败。") except Exception as e: logging.error(f"✗ 处理设备 {device_name} ({host}) 时发生未知错误: {str(e)}") # --- 主程序入口 --- if __name__ == "__main__": logging.info("===== 开始执行设备巡检与备份任务 =====") # 使用脚本所在目录下的 devices.json devices_file = os.path.join(SCRIPT_DIR, 'devices.json') devices = load_devices_from_json(devices_file) if not devices: logging.error("设备列表为空,脚本终止。") exit(1) for dev in devices: process_device(dev) logging.info("===== 所有任务执行完毕 =====")

脚本使用:

设备列表文件和脚本放置在同一目录

运行后,会在当前目录(脚本所在目录)生成backups目录和reports目录分别存放配置文件和设备巡检运行状态文件。

三、结果

由于前面没有完成脚本的测试。今天在进行脚本测试时,遇到了一个问题:

在脚本执行登录Cisco设备时未成功。

当时认为是SSH 算法的问题,但将所有的算法经过测试后还是失败了。

后来通过获取了一下我所使用的netmiko版本支持的Cisco设备类型

from netmiko.ssh_dispatcher import CLASS_MAPPER_BASE # 获取支持的设备列表 supported_devices = list(CLASS_MAPPER_BASE.keys()) print("Netmiko支持的设备列表:") for device_type in sorted(supported_devices): print(f'- {device_type}')
Netmiko支持的设备列表: - a10 - accedian - adtran_os - adva_fsp150f2 - adva_fsp150f3 - alcatel_aos - alcatel_sros - allied_telesis_awplus - apresia_aeos - arista_eos - arris_cer - aruba_os - aruba_osswitch - aruba_procurve - audiocode_66 - audiocode_72 - audiocode_shell - avaya_ers - avaya_vsp - broadcom_icos - brocade_fastiron - brocade_fos - brocade_netiron - brocade_nos - brocade_vdx - brocade_vyos - calix_b6 - casa_cmts - cdot_cros - centec_os - checkpoint_gaia - ciena_saos - cisco_asa - cisco_ftd - cisco_ios - cisco_nxos - cisco_s200 - cisco_s300 - cisco_tp - cisco_viptela - cisco_wlc - cisco_xe - cisco_xr - cloudgenix_ion - coriant - dell_dnos9 - dell_force10 - dell_isilon - dell_os10 - dell_os6 - dell_os9 - dell_powerconnect - dell_sonic - dlink_ds - eltex - eltex_esr - endace - enterasys - ericsson_ipos - ericsson_mltn63 - ericsson_mltn66 - extreme - extreme_ers - extreme_exos - extreme_netiron - extreme_nos - extreme_slx - extreme_tierra - extreme_vdx - extreme_vsp - extreme_wing - f5_linux - f5_ltm - f5_tmsh - flexvnf - fortinet - generic - generic_termserver - hillstone_stoneos - hp_comware - hp_procurve - huawei - huawei_olt - huawei_smartax - huawei_vrp - huawei_vrpv8 - ipinfusion_ocnos - juniper - juniper_junos - juniper_screenos - keymile - keymile_nos - linux - mellanox - mellanox_mlnxos - mikrotik_routeros - mikrotik_switchos - mrv_lx - mrv_optiswitch - netapp_cdot - netgear_prosafe - netscaler - nokia_srl - nokia_sros - oneaccess_oneos - ovs_linux - paloalto_panos - pluribus - quanta_mesh - rad_etx - raisecom_roap - ruckus_fastiron - ruijie_os - sixwind_os - sophos_sfos - supermicro_smis - teldat_cit - tplink_jetstream - ubiquiti_edge - ubiquiti_edgerouter - ubiquiti_edgeswitch - ubiquiti_unifiswitch - vyatta_vyos - vyos - watchguard_fireware - yamaha - zte_zxros - zyxel_os

然后查看了一下我所使用的模拟器的设备版本

Router#show version Cisco IOS XE Software, Version 17.03.01a Cisco IOS Software [Amsterdam], Virtual XE Software (X86_64_LINUX_IOSD-UNIVERSALK9-M), Version 17.3.1a, RELEASE SOFTWARE (fc3) Technical Support: http://www.cisco.com/techsupport Copyright (c) 1986-2020 by Cisco Systems, Inc. Compiled Tue 11-Aug-20 23:56 by mcpre Cisco IOS-XE software, Copyright (c) 2005-2020 by cisco Systems, Inc. All rights reserved. Certain components of Cisco IOS-XE software are licensed under the GNU General Public License ("GPL") Version 2.0. The software code licensed under GPL Version 2.0 is free software that comes with ABSOLUTELY NO WARRANTY. You can redistribute and/or modify such GPL code under the terms of GPL Version 2.0. For more details, see the documentation or "License Notice" file accompanying the IOS-XE software, or the applicable URL provided on the flyer accompanying the IOS-XE software. ROM: IOS-XE ROMMON

于是就将JSON文件中Cisco设备的 "device_type" 修改为 "cisco_xe",脚本顺利执行!

http://www.jsqmd.com/news/767873/

相关文章:

  • 大厂AI布局启示录:小白也能抓住高薪机遇,一起学大模型!
  • Windows 11/Win10本地磁盘告急?试试用SSHFS把云服务器挂成“无限外挂硬盘”
  • slidev-agent-skill:为AI智能体赋能,自动化创建Slidev演示文稿
  • Armv8-A virtualization 笔记 (一)
  • RepoAgent:基于大语言模型的智能代码仓库分析与自动化文档生成
  • 【逻辑回归从原理到实战:正则化、参数调优与过拟合处理】
  • 网络安全之GRE
  • 基于 Simulink 的数字控制延时补偿与稳定性分析深度实战教程
  • Java调用海康SDK的NET_DVR_STDXMLConfig接口,手把手教你获取设备信息(附完整代码)
  • 迭代器模式是行为型设计模式的一种,其核心思想是提供一种方法顺序访问一个聚合对象中的各个元素
  • 开源三指机械爪OpenClaw:从Arduino控制到ROS集成的完整实现指南
  • 英语全局通用・元音弱读规律
  • 赛博“听诊器”:手把手教你用Windows命令给电脑做体检
  • Promise/A+ 02
  • 【数据库操作全指南:从表创建到高级查询】
  • LyricsX:让Mac音乐体验更完美的智能歌词同步神器 [特殊字符]
  • 服务器重启后 Docker Compose 容器如何自动恢复运行
  • 用立创EDA复刻蓝桥杯省赛真题电路:手把手搭建一个简易电压采集与显示系统(2022模拟题2)
  • DeepSeek-V4-pro 接入 Claude Code 教程
  • 三步轻松备份QQ空间说说历史记录:GetQzonehistory完整指南
  • Docker 27 医疗容器认证实操手册:从镜像签名、SBOM生成到FDA 21 CFR Part 11审计就绪,一步不踩坑
  • 软件评测师基础知识专项刷题:软件工程
  • C语言选择结构自用讲解
  • 03-二叉树——从递归遍历到非递归实现
  • 别再只盯着CAN了!手把手教你用CAN FD收发器搞定汽车ECU的8Mbps高速通信
  • 2026年质量好的江苏熔模铸造推荐品牌厂家 - 行业平台推荐
  • HTML 与 ISO-8859-1 编码
  • 2026新疆小包团定制旅行社推荐:纯玩无购物/口碑靠谱旅行社榜单排行 - 栗子测评
  • 专业干货:AI教材写作全攻略,低查重技巧与优质工具大揭秘!
  • AwesomeQt:最小的Qt6系列迷你版本教程发布!