Asynchronous SNMP (twisted, v3arch)

Twisted is quite old but still widely used I/O framework. With Twisted, your code will mostly live in isolated functions, but unlike as it is with callback-based design, with Twisted work-in-progress is represented by a Deferred class instance effectively carrying state and context of running operation. Your callback functions will be attached to these Deferred objects and invoked as Deferred is done.

Based on Twisted infrastructure, individual asynchronous functions could be chained to run sequentially or in parallel.

In most examples approximate analogues of well known Net-SNMP snmp* tools command line options are shown. That may help those readers who, by chance are familiar with Net-SNMP tools, better understanding what example code doe

Here’s a quick example on a simple SNMP GET by high-level API:

  • with SNMPv1, community ‘public’
  • over IPv4/UDP
  • to an Agent at demo.snmplabs.com:161
  • for two instances of SNMPv2-MIB::sysDescr.0 MIB object,
  • based on Twisted I/O framework
from twisted.internet.task import react
from pysnmp.hlapi.v3arch.twisted import *


def success(args, hostname):
    errorStatus, errorIndex, varBinds = args

    if errorStatus:
        print('%s: %s at %s' % (hostname,
                                errorStatus.prettyPrint(),
                                errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))

    else:
        for varBind in varBinds:
            print(' = '.join([x.prettyPrint() for x in varBind]))


def failure(errorIndication, hostname):
    print('%s failure: %s' % (hostname, errorIndication))


# noinspection PyUnusedLocal
def getSysDescr(reactor, hostname):
    deferred = getCmd(
        SnmpEngine(),
        CommunityData('public', mpModel=0),
        UdpTransportTarget((hostname, 161)),
        ContextData(),
        ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0))
    )

    deferred.addCallback(success, hostname).addErrback(failure, hostname)

    return deferred


react(getSysDescr, ['demo.snmplabs.com'])

To make use of SNMPv3 and USM, the following code performs a series of SNMP GETNEXT operations effectively fetching a table of SNMP variables from SNMP Agent:

  • with SNMPv3, user ‘usr-none-none’, no authentication, no privacy
  • over IPv4/UDP
  • to an Agent at demo.snmplabs.com:161
  • for all OIDs past SNMPv2-MIB::system
  • run till end-of-mib condition is reported by Agent
  • based on Twisted I/O framework
from twisted.internet.task import react
from pysnmp.hlapi.v3arch.twisted import *


def success(args, reactor, snmpEngine):
    errorStatus, errorIndex, varBindTable = args

    if errorStatus:
        print('%s: %s at %s' % (hostname,
                                errorStatus.prettyPrint(),
                                errorIndex and varBindTable[0][int(errorIndex) - 1][0] or '?'))

    else:
        for varBindRow in varBindTable:
            for varBind in varBindRow:
                print(' = '.join([x.prettyPrint() for x in varBind]))

        if not isEndOfMib(varBindTable[-1]):
            return getbulk(reactor, snmpEngine, *varBindTable[-1])


def failure(errorIndication):
    print(errorIndication)


def getbulk(reactor, snmpEngine, varBinds):
    deferred = bulkCmd(
        snmpEngine,
        UsmUserData('usr-none-none'),
        UdpTransportTarget(('demo.snmplabs.com', 161)),
        ContextData(),
        0, 50,
        varBinds
    )

    deferred.addCallback(success, reactor, snmpEngine).addErrback(failure)

    return deferred


react(getbulk, [SnmpEngine(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'system'))])

More examples on Command Generator API usage follow.

Sending SNMP TRAP’s and INFORM’s is as easy with PySNMP library. The following code sends SNMP TRAP:

  • SNMPv1
  • with community name ‘public’
  • over IPv4/UDP
  • send TRAP notification
  • with Generic Trap #1 (warmStart) and Specific Trap 0
  • with default Uptime
  • with default Agent Address
  • with Enterprise OID 1.3.6.1.4.1.20408.4.1.1.2
  • include managed object information ‘1.3.6.1.2.1.1.1.0’ = ‘my system’
from twisted.internet.task import react
from pysnmp.hlapi.v3arch.twisted import *


def success(args, hostname):
    errorStatus, errorIndex, varBinds = args

    if errorStatus:
        print('%s: %s at %s' % (
            hostname,
            errorStatus.prettyPrint(),
            errorIndex and varBinds[int(errorIndex) - 1][0] or '?'
        )
              )

    else:
        for varBind in varBinds:
            print(' = '.join([x.prettyPrint() for x in varBind]))


def failure(errorIndication, hostname):
    print('%s failure: %s' % (hostname, errorIndication))


# noinspection PyUnusedLocal
def run(reactor, hostname):

    deferred = sendNotification(
        SnmpEngine(),
        CommunityData('public', mpModel=0),
        UdpTransportTarget((hostname, 162)),
        ContextData(),
        'trap',
        NotificationType(
            ObjectIdentity('1.3.6.1.6.3.1.1.5.2')
        ).loadMibs(
            'SNMPv2-MIB'
        ).addVarBinds(
            ('1.3.6.1.6.3.1.1.4.3.0', '1.3.6.1.4.1.20408.4.1.1.2'),
            ('1.3.6.1.2.1.1.1.0', OctetString('my system'))
        )
    )

    deferred.addCallback(success, hostname).addErrback(failure, hostname)

    return deferred


react(run, ['demo.snmplabs.com'])

More examples on Notification Originator API usage follow.

More sophisticated or less popular SNMP operations can still be performed with PySNMP through its Native API to Standard SNMP Applications.