Pysnmp使用指南

在这里插入图片描述

1. 简介

pysnmp 是一个纯 Python 实现的 SNMP（Simple Network Management Protocol）库，支持 SNMPv1、SNMPv2c 和 SNMPv3 协议。用于：

查询（GET）和修改（SET）网络设备的管理信息。
遍历（WALK）设备的 MIB（Management Information Base）树。
实现 SNMP Agent（服务器）或 Manager（客户端）。

2. 安装

pip install pysnmp pysnmp-mibs  # 安装核心库和 MIB 支持

3. 基本用法

3.1 SNMP GET 请求

from pysnmp.hlapi import *# 定义 SNMP 参数
community = 'public'  # 社区名（SNMPv1/v2c）
ip_address = '192.168.1.1'  # 目标设备 IP
port = 161  # 默认 SNMP 端口
oid = '1.3.6.1.2.1.1.1.0'  # OID（例如：系统描述）# 发起 GET 请求
error_indication, error_status, error_index, var_binds = next(getCmd(SnmpEngine(),CommunityData(community),UdpTransportTarget((ip_address, port), timeout=2, retries=2),ContextData(),ObjectType(ObjectIdentity(oid)))
)# 处理响应
if error_indication:print(f"Error: {error_indication}")
elif error_status:print(f"Error Status: {error_status}")
else:for var_bind in var_binds:print(f"{var_bind[0]} = {var_bind[1]}")

3.2 SNMP WALK 遍历

from pysnmp.hlapi import *error_indication, error_status, error_index, var_bind_table = next(nextCmd(SnmpEngine(),CommunityData('public'),UdpTransportTarget(('192.168.1.1', 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1')),  # 遍历系统组lexicographicMode=False)  # 按字典顺序遍历
)if error_indication:print(f"Error: {error_indication}")
else:for var_bind_row in var_bind_table:for var_bind in var_bind_row:print(f"{var_bind[0]} = {var_bind[1]}")

3.3 SNMP SET 操作

from pysnmp.hlapi import *error_indication, error_status, error_index, var_binds = next(setCmd(SnmpEngine(),CommunityData('private', mpModel=1),  # 使用写权限的社区名UdpTransportTarget(('192.168.1.1', 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1.5.0'), OctetString('NewDeviceName')))
)if error_indication:print(f"Error: {error_indication}")
elif error_status:print(f"SET Error: {error_status}")
else:print("SET 操作成功")

4. SNMPv3 配置

SNMPv3 支持认证和加密：

from pysnmp.hlapi import *user = 'user1'
auth_key = 'authkey123'
priv_key = 'privkey123'error_indication, error_status, error_index, var_binds = next(getCmd(SnmpEngine(),UsmUserData(user,authKey=auth_key,privKey=priv_key,authProtocol=usmHMACSHAAuthProtocol,  # 认证协议privProtocol=usmAesCfb128Protocol),    # 加密协议UdpTransportTarget(('192.168.1.1', 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')))
)

5. MIB 管理

使用 MIB 名称代替原始 OID：

from pysnmp.hlapi import *
from pysnmp.smi import compiler# 加载 MIB 文件
compiler.addMibCompiler(compiler.DirMibSource('/path/to/mibs'))
ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0).resolveWithMib()# 查询
error_indication, error_status, error_index, var_binds = next(getCmd(SnmpEngine(),CommunityData('public'),UdpTransportTarget(('192.168.1.1', 161)),ContextData(),ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)))
)

6. 高级主题

6.1 异步操作

使用 AsyncCommandGenerator 实现异步请求：

from pysnmp.hlapi.asyncio import *async def snmp_get():snmp_engine = SnmpEngine()error_indication, error_status, error_index, var_binds = await getCmd(snmp_engine,CommunityData('public'),UdpTransportTarget(('192.168.1.1', 161)),ContextData(),ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')))# 处理结果...import asyncio
asyncio.run(snmp_get())

6.2 SNMP Agent 实现

创建简单的 SNMP Agent：

from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, contextsnmp_engine = engine.SnmpEngine()
config.addV1System(snmp_engine, 'my-area', 'public')
config.addTransport(snmp_engine, udp.domainName, udp.UdpTransport().openServerMode(('0.0.0.0', 161)))context.SnmpContext(snmp_engine)
cmdrsp.GetCommandResponder(snmp_engine, context.SnmpContext(snmp_engine))snmp_engine.transportDispatcher.jobStarted(1)
try:snmp_engine.transportDispatcher.runDispatcher()
except KeyboardInterrupt:snmp_engine.transportDispatcher.closeDispatcher()