Agent side¶
Implementing MIB objects¶
This script explains how SNMP Agent application could model real-world data as Managed Objects defined in MIB.
from pysnmp.smi import builder
# MIB Builder is normally pre-created by SNMP engine
mibBuilder = builder.MibBuilder()
#
# This may be done in a stand-alone file and then loaded up
# by SNMP Agent
#
# A base class for a custom Managed Object
MibScalarInstance, = mibBuilder.importSymbols(
'SNMPv2-SMI', 'MibScalarInstance'
)
# Managed object specification
sysLocation, = mibBuilder.importSymbols('SNMPv2-MIB', 'sysLocation')
# Custom Managed Object
class MySysLocationInstance(MibScalarInstance):
# noinspection PyUnusedLocal
def readGet(self, varBind, **context):
cbFun = context['cbFun']
# Just return a custom value
cbFun((varBind[0], self.syntax.clone('The Leaky Cauldron')), **context)
sysLocationInstance = MySysLocationInstance(
sysLocation.name, (0,), sysLocation.syntax
)
# Register Managed Object with a MIB tree
mibBuilder.exportSymbols(
# '__' prefixed MIB modules take precedence on indexing
'__MY-LOCATION-MIB', sysLocationInstance=sysLocationInstance
)
if __name__ == '__main__':
#
# This is what is done internally by Agent.
#
from pysnmp.smi import instrum, exval
mibInstrum = instrum.MibInstrumController(mibBuilder)
def cbFun(varBinds, **context):
for oid, val in varBinds:
if exval.endOfMib.isSameTypeWith(val):
context['app']['stop'] = True
print('%s = %s' % ('.'.join([str(x) for x in oid]), not val.isValue and 'N/A' or val.prettyPrint()))
context['app']['varBinds'] = varBinds
app_context = {
'varBinds': [((1, 3, 6), None)],
'stop': False
}
print('Remote manager read access to MIB instrumentation (table walk)')
while not app_context['stop']:
mibInstrum.readNextMibObjects(*app_context['varBinds'], cbFun=cbFun, app=app_context)
print('done')
Download
script.
Agent operations on MIB¶
This script explains how SNMP Agent application manipulates its MIB possibly triggered by SNMP Manager’s commands.
from pysnmp.smi import builder
from pysnmp.smi import instrum
from pysnmp.smi import exval
from pysnmp.smi import error
def walkMib():
def cbFun(varBinds, **context):
err = context.get('error')
if err:
print(err)
for oid, val in varBinds:
if exval.endOfMib.isSameTypeWith(val):
context['app']['stop'] = True
elif not (exval.noSuchInstance.isSameTypeWith(val) or
exval.noSuchObject.isSameTypeWith(val)):
print('%s = %s' % ('.'.join([str(x) for x in oid]),
not val.isValue and 'N/A' or val.prettyPrint()))
context['app']['varBinds'] = varBinds
app_context = {
'varBinds': [((1, 3, 6), None)],
'stop': False
}
print('Read whole MIB (table walk)')
while not app_context['stop']:
mibInstrum.readNextMibObjects(*app_context['varBinds'], cbFun=cbFun, app=app_context)
print('Loading MIB modules...')
mibBuilder = builder.MibBuilder().loadModules(
'SNMPv2-MIB', 'SNMP-FRAMEWORK-MIB', 'SNMP-COMMUNITY-MIB'
)
print('Building MIB tree...')
mibInstrum = instrum.MibInstrumController(mibBuilder)
walkMib()
print('Building table entry index from human-friendly representation...')
snmpCommunityEntry, = mibBuilder.importSymbols(
'SNMP-COMMUNITY-MIB', 'snmpCommunityEntry'
)
instanceId = snmpCommunityEntry.getInstIdFromIndices('my-router')
print('Create/update some of SNMP-COMMUNITY-MIB::snmpCommunityEntry table columns: ')
def cbFun(varBinds, **context):
err = context.get('error')
if err:
print(err)
for oid, val in varBinds:
print('%s = %s' % ('.'.join([str(x) for x in oid]), not val.isValue and 'N/A' or val.prettyPrint()))
mibInstrum.writeMibObjects(
(snmpCommunityEntry.name + (2,) + instanceId, 'mycomm'),
(snmpCommunityEntry.name + (3,) + instanceId, 'mynmsname'),
(snmpCommunityEntry.name + (7,) + instanceId, 'volatile'),
cbFun=cbFun
)
walkMib()
def cbFun(varBinds, **context):
err = context.get('error')
if err:
print(err)
for oid, val in varBinds:
print('%s = %s' % ('.'.join([str(x) for x in oid]), not val.isValue and 'N/A' or val.prettyPrint()))
print('Destroy SNMP-COMMUNITY-MIB::snmpCommunityEntry table row via RowStatus column: ')
mibInstrum.writeMibObjects(
(snmpCommunityEntry.name + (8,) + instanceId, 'destroy'),
cbFun=cbFun
)
walkMib()
def cbFun(varBinds, **context):
err = context.get('errors', None)
if err:
print(err)
for oid, val in varBinds:
print('%s = %s' % ('.'.join([str(x) for x in oid]), not val.isValue and 'N/A' or val.prettyPrint()))
print('Create SNMP-COMMUNITY-MIB::snmpCommunityEntry table row: ')
mibInstrum.writeMibObjects(
(snmpCommunityEntry.name + (1,) + instanceId, 'mycomm'),
(snmpCommunityEntry.name + (2,) + instanceId, 'mycomm'),
(snmpCommunityEntry.name + (3,) + instanceId, 'mysecname'),
(snmpCommunityEntry.name + (4,) + instanceId, 'abcdef'),
(snmpCommunityEntry.name + (5,) + instanceId, ''),
(snmpCommunityEntry.name + (6,) + instanceId, 'mytag'),
(snmpCommunityEntry.name + (7,) + instanceId, 'nonVolatile'),
(snmpCommunityEntry.name + (8,) + instanceId, 'createAndGo'),
cbFun=cbFun
)
walkMib()
print('Destroy SNMP-COMMUNITY-MIB::snmpCommunityEntry table row via RowStatus column: ')
mibInstrum.writeMibObjects(
(snmpCommunityEntry.name + (8,) + instanceId, 'destroy'),
cbFun=cbFun
)
walkMib()
print('Unloading MIB modules...'),
mibBuilder.unloadModules()
Download
script.
See also: library reference.