MIB object calls REST API

In this configuration, SNMP responder serves a scalar MIB object backed by a REST API.

You could test this configuration by running:

$ snmpget -v2c -c public 127.0.0.1:1161 SNMPv2-MIB::sysName.0

SNMP Command Responder is configured to:

• listen on UDP socket at localhost

• form a MIB tree out of a few objects of the SNMPv2-MIB module

• respond to SNMPv2c queries

• serve all queries against the configured MIB tree

#
# SNMP Command Responder configuration file
#

config-version: 1
program-name: snmpresponder

snmp-credentials-group {
  snmp-transport-domain: 1.3.6.1.6.1.1.100
  snmp-bind-address: 127.0.0.1:1161

  snmp-engine-id: 0x0102030405070809

  snmp-community-name: public
  snmp-security-name: public
  snmp-security-model: 2
  snmp-security-level: 1

  snmp-credentials-id: snmp-credentials
}

context-group {
  snmp-context-engine-id-pattern: .*?
  snmp-context-name-pattern: .*?

  snmp-context-id: any-context
}

content-group {
  snmp-pdu-type-pattern: .*?
  snmp-pdu-oid-prefix-pattern-list: .*?

  snmp-content-id: any-content
}

peers-group {
  snmp-transport-domain: 1.3.6.1.6.1.1.100
  snmp-bind-address-pattern-list: .*?
  snmp-peer-address-pattern-list: .*?

  snmp-peer-id: 100
}

managed-objects-group {
  mib-text-search-path-list: http://mibs.pysnmp.com/asn1/
  mib-code-modules-pattern-list: ${config-dir}/managed-objects/.*py[co]?

  mib-tree-id: managed-objects-1
}

routing-map {
  matching-snmp-context-id-list: any-context
  matching-snmp-content-id-list: any-content

  matching-snmp-credentials-id-list: snmp-credentials
  matching-snmp-peer-id-list: 100

  using-mib-tree-id: managed-objects-1
}

Download configuration file.

The only implemented managed object SNMPv2-MIB::sysName.0:

• gathers its value from a REST API call

• REST API call is done asynchronously, from separate thread(s)

• only SNMP read operations are implemented

• write operation are allowed, but has no effect

"""SNMP MIB module (SNMPv2-MIB) expressed in pysnmp data model.

This Python module is designed to be imported and executed by the
pysnmp library.

See https://www.pysnmp.com/pysnmp for further information.

Notes
-----
ASN.1 source file:///usr/share/snmp/mibs/SNMPv2-MIB.txt
Produced by pysmi-0.4.0 at Sun Jan 13 09:39:06 2019
On host igarlic platform Darwin version 17.7.0 by user ilya
Using Python version 3.6.0 (v3.6.0:41df79263a11, Dec 22 2016, 17:23:13)
"""
import concurrent.futures
import urllib.request
import json

if 'mibBuilder' not in globals():
    import sys

    sys.stderr.write(__doc__)
    sys.exit(1)


MibScalarInstance, = mibBuilder.importSymbols(
    'SNMPv2-SMI',
    'MibScalarInstance'
)

# Import Managed Objects to base Managed Objects Instances on

(sysName,) = mibBuilder.importSymbols(
    "SNMPv2-MIB",
    "sysName"
)


# Persistent threaded executor

executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)


def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return json.loads(conn.read())


# MIB Managed Objects in the order of their OIDs

class SysnameObjectInstance(MibScalarInstance):

    REDFISH_SYSTEM_URL = 'http://demo.pysnmp.com/redfish/v1/Systems/437XR1138R2'

    def readTest(self, varBind, **context):
        # Just confirm that this MIB object instance is available
        cbFun = context['cbFun']
        cbFun(varBind, **context)

    def _callRestApi(self, varBind, **context):
        cbFun = context['cbFun']

        name, value = varBind

        future = executor.submit(load_url, self.REDFISH_SYSTEM_URL, 5)

        def done_callback(future):
            rsp = future.result()

            value = self.syntax.clone(rsp.get('HostName', ''))

            cbFun((name, value), **context)

        future.add_done_callback(done_callback)

    def readGet(self, varBind, **context):
        self._callRestApi(varBind, **context)

    def readTestNext(self, varBind, **context):
        name, value = varBind

        if name >= self.name:
            # This object does not qualify as "next*, pass the call
            MibScalarInstance.readTestNext(self, varBind, **context)

        else:
            # Confirm this object is available and report its OID
            cbFun = context['cbFun']
            cbFun((self.name, value), **context)

    def readGetNext(self, varBind, **context):
        name, value = varBind

        if name >= self.name:
            # This object does not qualify as "next*, pass the call
            MibScalarInstance.readGetNext(self, varBind, **context)

        else:
            self._callRestApi((self.name, value), **context)


_sysName = SysnameObjectInstance(
     sysName.name,
     (0,),
     sysName.syntax
)

# Export Managed Objects Instances to the MIB builder

mibBuilder.exportSymbols(
    "__SNMPv2-MIB",
    **{"sysName": _sysName}
)

Download MIB implementation.

For more information on MIB implementation refer to the MIB implementation chapter in the documentation.