asn1crypto_snmp.py

#!/usr/bin/python

from asn1crypto.core import (
        Choice,
        Sequence,
        SequenceOf,
        Integer,
        ObjectIdentifier,
        OctetString,
        Any,
        )


class Integer32(Integer):
    pass

class ErrStatus(Integer):
    _map = {
            0: 'noError',
            # FIXME: add rest
            }

class IpAddress(OctetString):
    class_ = 1
    tag = 0

class Counter32(Integer):
    class_ = 1
    tag = 1

class Gauge32(Integer):
    class_ = 1
    tag = 2

class TimeTicks(Integer):
    class_ = 1
    tag = 3

class Counter64(Integer):
    class_ = 1
    tag = 6

class VarbindValue(Choice):
    _alternatives = [
            ('int', Integer),
            ('str', OctetString),
            ('oid', ObjectIdentifier),
            ('ip4', IpAddress),
            ('counter', Counter32),
            ('ticks', TimeTicks),
            ]

class Varbind(Sequence):
    _fields = [
            ('name', ObjectIdentifier),
            ('value', VarbindValue),
            ]

class Varbinds(SequenceOf):
    _child_spec = Varbind

class PDU(Sequence):
    _fields = [
            ('request_id', Integer32),
            ('error_status', ErrStatus),
            ('error_inded', Integer),
            ('varbinds', Varbinds),
            ]

class SnmpPDU(Choice):
    _alternatives = [
            ('get-pdu', PDU, {'tag_type': 'implicit', 'tag': 0}),
            ('getnext-pdu', PDU, {'tag_type': 'implicit', 'tag': 1}),
            ('response-pdu', PDU, {'tag_type': 'implicit', 'tag': 2}),
            ('set-pdu', PDU, {'tag_type': 'implicit', 'tag': 3}),
            ]

class Message(Sequence):
    _fields = [
            ('version', Integer),
            ('community', OctetString),
            ('pdu', SnmpPDU),
            ]

snmptest.py

#!/usr/bin/env python

import timeit

# libsnmp
import libsnmp.rfc1905

# pysnmp
from pyasn1.codec.ber import encoder, decoder
from pysnmp.proto import api

pysnmp_v1 = api.protoModules[api.protoVersion1]
pysnmp_v2 = api.protoModules[api.protoVersion2c]

# asn1crypto
from asn1crypto_snmp import Message as asn1cMessage

pdu = """3081e60201010409636f6d6d756e697479a381d50204648e7c1c0201000201003081c6300c06072b060187670101020101300e06072b0601876702010403663030300f06072b0601876703014004c0a80101301106072b06018767040106062a03833c842b300c06072b060187670102020101300e06072b0601876702020403663030300f06072b0601876703024004c0a80101301106072b06018767040206062a03833c842b300c06072b060187670103020101300e06072b0601876702030403663030300f06072b0601876703034004c0a80101301106072b06018767040306062a03833c842b""".decode('hex')

def test_libsnmp(data=pdu):
    msg = libsnmp.rfc1905.Message().decode(data)
    err = msg.data.errorStatus.value
    vb = msg.data.varBindList.value

def test_pysnmp(data=pdu, apiver=pysnmp_v2, asn1spec=pysnmp_v2.Message()):
    rsp, msg = decoder.decode(data, asn1Spec=asn1spec)
    pdu = apiver.apiMessage.getPDU(rsp)
    err = apiver.apiPDU.getErrorStatus(pdu)
    vb = apiver.apiPDU.getVarBinds(pdu)

def test_asn1crypto(data=pdu):
    msg = asn1cMessage.load(data)
    ntv = msg.native

def main(n=50000):
    rv = timeit.timeit(test_libsnmp, number=n)
    print "timeit for libsnmp {0} iterations per-call {1:3f}ms".format(n, 1000*rv/n)

    rv = timeit.timeit(test_pysnmp, number=n)
    print "timeit for pysnmp {0} iterations per-call {1:3f}ms".format(n, 1000*rv/n)

    rv = timeit.timeit(test_asn1crypto, number=n)
    print "timeit for asn1crypto {0} iterations per-call {1:3f}ms".format(n, 1000*rv/n)

if __name__=='__main__':
    main()