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, context
snmp_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()
7. 常见问题
Q1: 超时或无响应
- 检查目标设备的 SNMP 服务是否启用。
- 检查防火墙是否允许 UDP 161 端口通信。
Q2: OID 不存在
- 使用
snmpwalk
或 MIB 浏览器确认目标设备支持的 OID。
Q3: 权限不足
- 确保使用正确的社区名(如
private
用于写操作)。 - 对于 SNMPv3,检查用户认证和加密参数。
Q4: MIB 加载失败
- 确认 MIB 文件路径正确,或手动下载 MIB 文件。
通过以上步骤,可以快速上手 pysnmp
进行基本的 SNMP 操作。更详细的文档请参考 pysnmp 官方文档。