snmp.py

import logging
import re

from pysnmp.hlapi import nextCmd, SnmpEngine, CommunityData, \
        UdpTransportTarget, ContextData, ObjectType, ObjectIdentity
from pysnmp.smi import builder, compiler
# from pysnmp.smi import view, rfc1902


RFC1213_MIB_IFDESC = '1.3.6.1.2.1.2.2.1.2'


class SNMPWalkError(ConnectionError):
    pass


def _v6address_oid2str(dotted_decimal):
    hex_params = []
    for dec in re.split(r'\.', dotted_decimal):
        hex_params.append("%02x" % int(dec))
    return ":".join(hex_params)


def walk(agent_hostname, community, base_oid):  # pragma: no cover
    """
    https://stackoverflow.com/a/45001921
    http://snmplabs.com/pysnmp/docs/hlapi/asyncore/sync/manager/cmdgen/nextcmd.html
    http://snmplabs.com/pysnmp/faq/pass-custom-mib-to-manager.html
    https://github.com/etingof/pysnmp/blob/master/examples/v3arch/asyncore/manager/cmdgen/getnext-multiple-oids-and-resolve-with-mib.py
    http://snmplabs.com/pysnmp/examples/smi/manager/browsing-mib-tree.html

    :param agent_hostname:
    :param community:
    :param base_oid:
    :return:
    """

    logger = logging.getLogger(__name__)

    mibBuilder = builder.MibBuilder()
    # mibViewController = view.MibViewController(mibBuilder)
    compiler.addMibCompiler(
        mibBuilder,
        sources=['http://mibs.snmplabs.com/asn1/@mib@'])
    # Pre-load MIB modules we expect to work with
    mibBuilder.loadModules(
        'SNMPv2-MIB',
        'SNMP-COMMUNITY-MIB',
        'RFC1213-MIB')

    logger.debug("walking %s: %s" % (agent_hostname, base_oid))

    for (engineErrorIndication,
         pduErrorIndication,
         errorIndex,
         varBinds) in nextCmd(
            SnmpEngine(),
            CommunityData(community),
            UdpTransportTarget((agent_hostname, 161)),
            ContextData(),
            ObjectType(ObjectIdentity(base_oid)),
            lexicographicMode=False,
            lookupNames=True,
            lookupValues=True):

        # cf. http://snmplabs.com/
        #       pysnmp/examples/hlapi/asyncore/sync/contents.html
        if engineErrorIndication:
            raise SNMPWalkError(
                f'snmp response engine error indication: '
                f'{str(engineErrorIndication)} - {agent_hostname}')
        if pduErrorIndication:
            raise SNMPWalkError(
                'snmp response pdu error %r at %r' % (
                    pduErrorIndication,
                    errorIndex and varBinds[int(errorIndex) - 1][0] or '?'))
        if errorIndex != 0:
            raise SNMPWalkError(
                'sanity failure: errorIndex != 0, '
                'but no error indication')

        # varBinds = [
        #     rfc1902.ObjectType(rfc1902.ObjectIdentity(x[0]),x[1])
        #         .resolveWithMib(mibViewController)
        #     for x in varBinds]
        for oid, val in varBinds:
            result = {"oid": "." + str(oid), "value": val.prettyPrint()}
            logger.debug(result)
            yield result


def get_router_snmp_indexes(hostname, community):
    for ifc in walk(hostname, community, RFC1213_MIB_IFDESC):
        m = re.match(r'.*\.(\d+)$', ifc['oid'])
        assert m, f'sanity failure parsing oid: {ifc["oid"]}'
        yield {
            'name': ifc['value'],
            'index': int(m.group(1))
        }