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

基于 pysnmp 的实际业务场景应用案例：网络设备监控与配置系统

news 来源：原创 2025/6/16 5:18:55

在这里插入图片描述

场景背景

某企业需对全网 Cisco 交换机实现：

状态监控：实时获取设备健康状态
自动配置：批量修改设备名称
故障排查：接口状态异常告警
安全审计：SNMPv3 加密通信

案例一：设备健康监控（GET 操作）

需求场景

每5分钟采集核心交换机的：

系统描述
运行时间
CPU 使用率
内存利用率

示例代码

from pysnmp.hlapi import *# 监控目标设备
devices = [{"ip": "192.168.1.1", "community": "public"},{"ip": "192.168.1.2", "community": "public"}
]# 监控OID列表（Cisco MIB）
oids = ["1.3.6.1.2.1.1.1.0",    # sysDescr"1.3.6.1.2.1.1.3.0",    # sysUpTime"1.3.6.1.4.1.9.9.109.1.1.1.1.6.1",  # cpmCPUTotal1min"1.3.6.1.4.1.9.9.48.1.1.1.6.1"     # ciscoMemoryPoolUsed
]for device in devices:print(f"\n=== 设备 {device['ip']} 状态 ===")for oid in oids:error_indication, error_status, error_index, var_binds = next(getCmd(SnmpEngine(),CommunityData(device['community']),UdpTransportTarget((device['ip'], 161)),ContextData(),ObjectType(ObjectIdentity(oid)))# 错误处理if error_indication:print(f"OID {oid} 查询失败: {error_indication}")continue# 解析结果for var_bind in var_binds:oid_str = var_bind[0].prettyPrint()  # 可读的OID名称value = var_bind[1]  # 值对象# 示例输出解析print(f"{oid_str} = ", end="")# 根据OID类型格式化输出if "1.1.1.0" in oid_str:  # 系统描述print(value.prettyPrint())elif "1.1.3.0" in oid_str:  # 运行时间ticks = int(value)hours = ticks // 360000  # 百分之一秒转小时print(f"{hours}小时")elif "109.1.1.1.1.6" in oid_str:  # CPU使用率print(f"{int(value)}%")elif "48.1.1.1.6" in oid_str:  # 内存使用print(f"{int(value)/1000000:.2f} MB")

实际输出示例

=== 设备 192.168.1.1 状态 ===
SNMPv2-MIB::sysDescr.0 = Cisco IOS Software, C3750 Software (C3750-IPSERVICESK9-M), Version 12.2(58)SE2, RELEASE SOFTWARE (fc1)
SNMPv2-MIB::sysUpTime.0 = 128小时
CISCO-PROCESS-MIB::cpmCPUTotal1min.1 = 24%
CISCO-MEMORY-POOL-MIB::ciscoMemoryPoolUsed.1 = 256.34 MB

var_bind 结构解析

# var_bind[0] 示例 (OID对象)
<ObjectName value object at 0x7f9a5c0b2120; tagSet=<TagSet object at 0x7f9a5c0b2080>, payload=[1.3.6.1.2.1.1.1.0]># var_bind[1] 示例 (值对象)
<OctetString value object at 0x7f9a5c0b21c0; tagSet=<TagSet object at 0x7f9a5c0b2080>, encoding=utf-8, payload=Cisco IOS...># 常用格式化方法：
str(var_bind[0])  # '1.3.6.1.2.1.1.1.0'
var_bind[0].prettyPrint()  # 'SNMPv2-MIB::sysDescr.0' (需加载MIB)
var_bind[1].prettyPrint()  # 人类可读的值

案例二：批量设备重命名（SET 操作）

需求场景

新采购的交换机需按规范命名：
<位置>-<设备类型>-<序列号>

设备命名规则：上海机房-A区-交换机-C2960X-123456

SET 操作目的

修改设备系统名称（sysName OID: 1.3.6.1.2.1.1.5.0），实现：

统一设备标识规范
便于资产管理
快速定位故障设备

示例代码

from pysnmp.hlapi import *# 设备列表及新名称
devices = [{"ip": "192.168.10.1", "new_name": "上海机房-A区-交换机-C2960X-123456"},{"ip": "192.168.10.2", "new_name": "上海机房-B区-交换机-C3750-789012"}
]for device in devices:print(f"\n重命名 {device['ip']} => {device['new_name']}")error_indication, error_status, error_index, var_binds = next(setCmd(SnmpEngine(),CommunityData('private', mpModel=1),  # 写权限社区名UdpTransportTarget((device['ip'], 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1.5.0'), OctetString(device['new_name'])))# 结果验证if error_indication:print(f"错误: {error_indication}")elif error_status:print(f"协议错误: {error_status} at {error_index}")else:print("设置成功！验证结果：")# 立即查询验证error_indication, _, _, var_binds = next(getCmd(SnmpEngine(),CommunityData('public'),UdpTransportTarget((device['ip'], 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1.5.0')))if not error_indication:print(f"当前设备名: {var_binds[0][1].prettyPrint()}")

实际输出

重命名 192.168.10.1 => 上海机房-A区-交换机-C2960X-123456
设置成功！验证结果：
当前设备名: 上海机房-A区-交换机-C2960X-123456

案例三：接口故障监控（WALK 操作）

需求场景

实时检测所有接口状态：

识别 down 状态接口
检测高错误率接口
自动生成故障报告

示例代码

from pysnmp.hlapi import *def monitor_interfaces(ip, community):print(f"\n扫描设备 {ip} 接口状态...")# 获取接口状态表 (ifOperStatus)status_oid = "1.3.6.1.2.1.2.2.1.8"# 获取接口错误计数 (ifInErrors)error_oid = "1.3.6.1.2.1.2.2.1.14"# 存储接口状态interface_status = {}# 扫描状态表for (error_indication, error_status, error_index,var_binds) in nextCmd(SnmpEngine(),CommunityData(community),UdpTransportTarget((ip, 161)),ContextData(),ObjectType(ObjectIdentity(status_oid)),lexicographicMode=False):if error_indication:print(f"接口扫描失败: {error_indication}")return# 解析接口索引和状态for var_bind in var_binds:oid_parts = str(var_bind[0]).split('.')if_index = oid_parts[-1]  # 最后一个数字是接口索引status = int(var_bind[1])interface_status[if_index] = {'status': 'up' if status == 1 else 'down'}# 扫描错误计数for (error_indication, error_status, error_index,var_binds) in nextCmd(SnmpEngine(),CommunityData(community),UdpTransportTarget((ip, 161)),ContextData(),ObjectType(ObjectIdentity(error_oid)),lexicographicMode=False):if error_indication:print(f"错误计数扫描失败: {error_indication}")return# 解析错误计数for var_bind in var_binds:oid_parts = str(var_bind[0]).split('.')if_index = oid_parts[-1]error_count = int(var_bind[1])# 添加到接口数据if if_index in interface_status:interface_status[if_index]['errors'] = error_count# 生成报告print(f"\n设备 {ip} 接口状态报告:")print("接口ID | 状态 | 错误计数")for if_index, data in interface_status.items():status = data.get('status', 'unknown')errors = data.get('errors', 'N/A')print(f"{if_index:6} | {status:4} | {errors}")# 故障检测逻辑if status == 'down':print(f"  !!! 接口 {if_index} 宕机 !!!")elif errors > 1000:print(f"  !!! 接口 {if_index} 高错误率: {errors} 个错误包 !!!")# 执行监控
monitor_interfaces("192.168.1.1", "public")

实际输出示例

扫描设备 192.168.1.1 接口状态...设备 192.168.1.1 接口状态报告:
接口ID | 状态 | 错误计数
1     | up   | 12
2     | up   | 5
3     | down | 0!!! 接口 3 宕机 !!!
4     | up   | 1203!!! 接口 4 高错误率: 1203 个错误包 !!!

案例四：SNMPv3 安全审计

需求场景

对财务部门网络设备实现：

AES-128 加密通信
用户身份认证
操作日志审计

示例代码

from pysnmp.hlapi import *# SNMPv3 安全参数
security_params = {"user": "audit_user","auth_key": "authentication_key123","priv_key": "encryption_key456","auth_proto": usmHMACSHAAuthProtocol,  # SHA-1认证"priv_proto": usmAesCfb128Protocol     # AES-128加密
}def secure_operation(ip, oid, operation="get", value=None):# 构建引擎snmp_engine = SnmpEngine()# 用户安全模型usm_data = UsmUserData(security_params["user"],authKey=security_params["auth_key"],privKey=security_params["priv_key"],authProtocol=security_params["auth_proto"],privProtocol=security_params["priv_proto"])# 执行操作if operation == "get":cmd_gen = getCmd(snmp_engine, usm_data,UdpTransportTarget((ip, 161)),ContextData(),ObjectType(ObjectIdentity(oid)))elif operation == "set" and value is not None:cmd_gen = setCmd(snmp_engine, usm_data,UdpTransportTarget((ip, 161)),ContextData(),ObjectType(ObjectIdentity(oid), value))else:print("无效操作")return# 执行并获取结果error_indication, error_status, error_index, var_binds = next(cmd_gen)# 审计日志with open("snmp_audit.log", "a") as log:log.write(f"{ip}|{operation}|{oid}|{error_indication or '成功'}\n")return var_binds# 安全获取系统信息
result = secure_operation("10.0.1.100", "1.3.6.1.2.1.1.1.0")
if result:print("安全获取结果:", result[0][1].prettyPrint())# 审计日志示例
# 10.0.1.100|get|1.3.6.1.2.1.1.1.0|成功

关键技术总结

操作类型	业务场景	关键 OID	pysnmp 实现
GET	状态监控	系统: `1.3.6.1.2.1.1.1.0` CPU: `1.3.6.1.4.1.9.9.109.1.1.1.1.6.1`	`getCmd()` + 值解析
SET	批量配置	设备名: `1.3.6.1.2.1.1.5.0`	`setCmd()` + `OctetString()`
WALK	接口监控	接口状态: `1.3.6.1.2.1.2.2.1.8` 错误计数: `1.3.6.1.2.1.2.2.1.14`	`nextCmd(lexicographicMode=False)`
SNMPv3	安全审计	任意敏感 OID	`UsmUserData()` + 加密协议