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

通过Python的第三方库Netmiko实现不同厂商网络设备的日常巡检及配置备份。一、设备列表文件JSON 文件1、 我们先看一个示例1拓扑2脚本import time from netmiko import ConnectHandler AR1 { host: 192.168.22.100, username: pytest, password: admin123, 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: admin123, # password 为用户密码device_type: huawei_vrp # device_type 为Netmiko所支持的设备类型Netmiko 当前可支持绝大多数厂商设备包括华为、H3C、TP-Link等通过支持的设备类型预设了一些指令。send_command() 用户执行查询、保存等命令AR1执行display versionAR2执行display ip interface brief3结果当有大量设备时脚本中就需要定义很多台设备信息每台设备都需要写一段连接执行的指令。下次再用时还得对所有信息进行修改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: HWpytest123 } }, { name: CiscoRouter, connection: { device_type: cisco_xe, host: 192.168.44.102, username: ciscopytest, password: Pythontest123 } }, { name: H3CAR, connection: { device_type: hp_comware, host: 192.168.44.101, username: h3cpytest, password: Pythontest123 } } ]二、脚本1、模块导入import os import json import logging import datetime from netmiko import ConnectHandler from netmiko.exceptions import NetMikoTimeoutException, NetMikoAuthenticationException2、获取脚本所在目录# 定义变量 SCRIPT_IDR获取脚本所在目录用于获取设备列表JSON文件同时巡检信息和配置备份也将在此目录保存 SCRIPT_DIR os.path.dirname(os.path.abspath(__file__)) # 也可指定自定义JOSN文件所在目录路径同样在脚本运行时也会在此目录生成巡检和配置备份目录 # SCRIPT_DIR rD:\BASE # 以Windows系统为例 # --- 自定义输出根目录脚本所在目录输出文件将保存在 ./backups 和 ./reports --- OUTPUT_BASE_DIR SCRIPT_DIR3、设备指令列表根据不同厂商设备提前设置好需要使用到的指令如需其他指令可在 ‘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, encodingutf-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, encodingutf-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, encodingutf-8) as f: f.write(f设备 {actual_hostname} ({host}) 巡检报告 - {datetime.datetime.now()}\n) f.write(fJSON 定义名称{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_factor2) 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, levellogging.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, encodingutf-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, encodingutf-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, encodingutf-8) as f: f.write(f设备 {actual_hostname} ({host}) 巡检报告 - {datetime.datetime.now()}\n) f.write(fJSON 定义名称{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_factor2) 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脚本顺利执行