Agent-Side MIB Implementations

import sys
from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, context
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.proto.api import v2c

# Create SNMP engine
snmpEngine = engine.SnmpEngine()

# Transport setup

# UDP over IPv4
config.add_transport(
    snmpEngine, udp.DOMAIN_NAME, udp.UdpTransport().open_server_mode(("127.0.0.1", 161))
)

# SNMPv2c setup

# SecurityName <-> CommunityName mapping.
config.add_v1_system(snmpEngine, "my-area", "public")

# Allow read MIB access for this user / securityModels at VACM
config.add_vacm_user(snmpEngine, 2, "my-area", "noAuthNoPriv", (1, 3, 6, 5))

# Create an SNMP context
snmpContext = context.SnmpContext(snmpEngine)

# --- create custom Managed Object Instance ---

mibBuilder = snmpContext.get_mib_instrum().get_mib_builder()

MibScalar, MibScalarInstance = mibBuilder.import_symbols(
    "SNMPv2-SMI", "MibScalar", "MibScalarInstance"
)


class MyStaticMibScalarInstance(MibScalarInstance):
    # noinspection PyUnusedLocal,PyUnusedLocal
    def getValue(self, name, **context):
        return self.getSyntax().clone(
            f"Python {sys.version} running on a {sys.platform} platform"
        )


mibBuilder.export_symbols(
    "__MY_MIB",
    MibScalar((1, 3, 6, 5, 1), v2c.OctetString()),
    MyStaticMibScalarInstance((1, 3, 6, 5, 1), (0,), v2c.OctetString()),
)

# --- end of Managed Object Instance initialization ----

# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)

# Register an imaginary never-ending job to keep I/O dispatcher running forever
snmpEngine.transport_dispatcher.job_started(1)

# Run I/O dispatcher which would receive queries and send responses
try:
    snmpEngine.open_dispatcher()
except:
    snmpEngine.close_dispatcher()
    raise

Download script.

Implementing conceptual table

Listen and respond to SNMP GET/SET/GETNEXT/GETBULK queries with the following options:

• SNMPv2c

• with SNMP community “public”

• define a simple SNMP Table within a newly created EXAMPLE-MIB

• pre-populate SNMP Table with a single row of values

• allow read access only to the subtree where example SNMP Table resides

• over IPv4/UDP, listening at 127.0.0.1:161

The following Net-SNMP commands will populate and walk a table:

$ snmpset -v2c -c public 127.0.0.1 1.3.6.6.1.5.2.97.98.99 s “my value”

$ snmpset -v2c -c public 127.0.0.1 1.3.6.6.1.5.4.97.98.99 i 4

$ snmpwalk -v2c -c public 127.0.0.1 1.3.6

…while the following command will destroy the same row

$ snmpset -v2c -c public 127.0.0.1 1.3.6.6.1.5.4.97.98.99 i 6

$ snmpwalk -v2c -c public 127.0.0.1 1.3.6

from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, context
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.proto.api import v2c


# Create SNMP engine
snmpEngine = engine.SnmpEngine()

# Transport setup

# UDP over IPv4
config.add_transport(
    snmpEngine, udp.DOMAIN_NAME, udp.UdpTransport().open_server_mode(("127.0.0.1", 161))
)

# SNMPv2c setup

# SecurityName <-> CommunityName mapping.
config.add_v1_system(snmpEngine, "my-area", "public")

# Allow read MIB access for this user / securityModels at VACM
config.add_vacm_user(
    snmpEngine, 2, "my-area", "noAuthNoPriv", (1, 3, 6, 6), (1, 3, 6, 6)
)

# Create an SNMP context
snmpContext = context.SnmpContext(snmpEngine)

# --- define custom SNMP Table within a newly defined EXAMPLE-MIB ---

mibBuilder = snmpContext.get_mib_instrum().get_mib_builder()

(MibTable, MibTableRow, MibTableColumn, MibScalarInstance) = mibBuilder.import_symbols(
    "SNMPv2-SMI", "MibTable", "MibTableRow", "MibTableColumn", "MibScalarInstance"
)

(RowStatus,) = mibBuilder.import_symbols("SNMPv2-TC", "RowStatus")

mibBuilder.export_symbols(
    "__EXAMPLE-MIB",
    # table object
    exampleTable=MibTable((1, 3, 6, 6, 1)).setMaxAccess("read-create"),
    # table row object, also carries references to table indices
    exampleTableEntry=MibTableRow((1, 3, 6, 6, 1, 5))
    .setMaxAccess("read-create")
    .setIndexNames((0, "__EXAMPLE-MIB", "exampleTableColumn1")),
    # table column: string index
    exampleTableColumn1=MibTableColumn(
        (1, 3, 6, 6, 1, 5, 1), v2c.OctetString()
    ).setMaxAccess("read-create"),
    # table column: string value
    exampleTableColumn2=MibTableColumn(
        (1, 3, 6, 6, 1, 5, 2), v2c.OctetString()
    ).setMaxAccess("read-create"),
    # table column: integer value with default
    exampleTableColumn3=MibTableColumn(
        (1, 3, 6, 6, 1, 5, 3), v2c.Integer32(123)
    ).setMaxAccess("read-create"),
    # table column: row status
    exampleTableStatus=MibTableColumn(
        (1, 3, 6, 6, 1, 5, 4), RowStatus("notExists")
    ).setMaxAccess("read-create"),
)

# --- end of custom SNMP table definition, empty table now exists ---

# --- populate custom SNMP table with one row ---

(
    exampleTableEntry,
    exampleTableColumn2,
    exampleTableColumn3,
    exampleTableStatus,
) = mibBuilder.import_symbols(
    "__EXAMPLE-MIB",
    "exampleTableEntry",
    "exampleTableColumn2",
    "exampleTableColumn3",
    "exampleTableStatus",
)
rowInstanceId = exampleTableEntry.getInstIdFromIndices("example record one")
mibInstrumentation = snmpContext.get_mib_instrum()
mibInstrumentation.write_variables(
    (exampleTableColumn2.name + rowInstanceId, "my string value"),
    (exampleTableColumn3.name + rowInstanceId, 123456),
    (exampleTableStatus.name + rowInstanceId, "createAndGo"),
)

# --- end of SNMP table population ---

# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)

# Register an imaginary never-ending job to keep I/O dispatcher running forever
snmpEngine.transport_dispatcher.job_started(1)

# Run I/O dispatcher which would receive queries and send responses
try:
    snmpEngine.open_dispatcher()
except:
    snmpEngine.close_dispatcher()
    raise

Download script.

Serve non-default MIB tree

Listen and respond to SNMP GET/SET/GETNEXT/GETBULK queries with the following options:

• SNMPv3

• with USM username usr-md5-none

• using alternative set of Managed Objects addressed by contextEngineId: 8000000001020304, contextName: my-context

• allow access to SNMPv2-MIB objects (1.3.6.1.2.1)

• over IPv4/UDP, listening at 127.0.0.1:161

Either of the following Net-SNMP commands will walk this Agent:

$ snmpwalk -v3 -u usr-md5-none -l authNoPriv -A authkey1 -E 8000000001020304 -n my-context 127.0.0.1 .1.3.6

$ snmpwalk -v3 -u usr-md5-none -l authNoPriv -A authkey1 -E 8000000001020304 127.0.0.1 .1.3.6

from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, context
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.smi import instrum, builder
from pysnmp.proto.api import v2c

# Create SNMP engine
snmpEngine = engine.SnmpEngine()

# Transport setup

# UDP over IPv4
config.add_transport(
    snmpEngine, udp.DOMAIN_NAME, udp.UdpTransport().open_server_mode(("127.0.0.1", 161))
)

# SNMPv3/USM setup

# user: usr-md5-none, auth: MD5, priv NONE
config.add_v3_user(snmpEngine, "usr-md5-none", config.USM_AUTH_HMAC96_MD5, "authkey1")

# Allow full MIB access for each user at VACM
config.add_vacm_user(
    snmpEngine, 3, "usr-md5-none", "authNoPriv", (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1)
)

# Create an SNMP context with ContextEngineId = 8000000001020304
snmpContext = context.SnmpContext(
    snmpEngine, contextEngineId=v2c.OctetString(hexValue="8000000001020304")
)

# Create an [empty] set of Managed Objects (MibBuilder), pass it to
# Management Instrumentation Controller and register at SNMP Context
# under ContextName 'my-context'
snmpContext.register_context_name(
    v2c.OctetString("my-context"),  # Context Name
    instrum.MibInstrumController(builder.MibBuilder()),  # Managed Objects
)

# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)

# Register an imaginary never-ending job to keep I/O dispatcher running forever
snmpEngine.transport_dispatcher.job_started(1)

# Run I/O dispatcher which would receive queries and send responses
try:
    snmpEngine.open_dispatcher()
except:
    snmpEngine.close_dispatcher()
    raise

Download script.

Multiple MIB trees under distinct context names

Listen and respond to SNMP GET/SET/GETNEXT/GETBULK queries with the following options:

• SNMPv3

• with USM username usr-md5-none

• using two independent (empty) sets of Managed Objects addressed by SNMP context name context-a and context-b

• allow access to SNMPv2-MIB objects (1.3.6.1.2.1)

• over IPv4/UDP, listening at 127.0.0.1:161

Either of the following Net-SNMP commands will walk this Agent:

$ snmpwalk -v3 -u usr-md5-none -l authNoPriv -A authkey1 -n context-a 127.0.0.1 .1.3.6

$ snmpwalk -v3 -u usr-md5-none -l authNoPriv -A authkey1 -n context-b 127.0.0.1 .1.3.6

from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, context
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.smi import instrum, builder
from pysnmp.proto.api import v2c

# Create SNMP engine
snmpEngine = engine.SnmpEngine()

# Transport setup

# UDP over IPv4
config.add_transport(
    snmpEngine, udp.domainName, udp.UdpTransport().open_server_mode(("127.0.0.1", 161))
)

# SNMPv3/USM setup

# user: usr-md5-none, auth: MD5, priv NONE
config.add_v3_user(snmpEngine, "usr-md5-none", config.USM_AUTH_HMAC96_MD5, "authkey1")

# Allow full MIB access for each user at VACM
config.add_vacm_user(
    snmpEngine, 3, "usr-md5-none", "authNoPriv", (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1)
)

# Create an SNMP context with default ContextEngineId (same as SNMP engine ID)
snmpContext = context.SnmpContext(snmpEngine)

# Create multiple independent trees of MIB managed objects (empty so far)
mibTreeA = instrum.MibInstrumController(builder.MibBuilder())
mibTreeB = instrum.MibInstrumController(builder.MibBuilder())

# Register MIB trees at distinct SNMP Context names
snmpContext.register_context_name(v2c.OctetString("context-a"), mibTreeA)
snmpContext.register_context_name(v2c.OctetString("context-b"), mibTreeB)

# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)

# Register an imaginary never-ending job to keep I/O dispatcher running forever
snmpEngine.transport_dispatcher.job_started(1)

# Run I/O dispatcher which would receive queries and send responses
try:
    snmpEngine.open_dispatcher()

except:
    snmpEngine.close_dispatcher()
    raise

Download script.

Custom MIB Controller

Listen and respond to SNMP GET/SET/GETNEXT/GETBULK queries with the following options:

• SNMPv3

• with USM username usr-none-none

• using alternative set of Managed Objects addressed by contextName: my-context

• allow access to SNMPv2-MIB objects (1.3.6.1.2.1)

• over IPv4/UDP, listening at 127.0.0.1:161

The following Net-SNMP command will send GET request to this Agent:

$ snmpget -v3 -u usr-none-none -l noAuthNoPriv -n my-context -Ir 127.0.0.1 sysDescr.0

from pysnmp.entity import engine, config
from pysnmp.entity.rfc3413 import cmdrsp, context
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.smi import instrum
from pysnmp.proto.api import v2c

# Create SNMP engine
snmpEngine = engine.SnmpEngine()

# Transport setup

# UDP over IPv4
config.add_transport(
    snmpEngine, udp.DOMAIN_NAME, udp.UdpTransport().open_server_mode(("127.0.0.1", 161))
)

# SNMPv3/USM setup

# user: usr-none-none, auth: NONE, priv NONE
config.add_v3_user(snmpEngine, "usr-none-none")

# Allow full MIB access for each user at VACM
config.add_vacm_user(
    snmpEngine,
    3,
    "usr-none-none",
    "noAuthNoPriv",
    (1, 3, 6, 1, 2, 1),
    (1, 3, 6, 1, 2, 1),
)

# Create an SNMP context
snmpContext = context.SnmpContext(snmpEngine)


# Very basic Management Instrumentation Controller without
# any Managed Objects attached. It supports only GET's and
# always echos request var-binds in response.
class EchoMibInstrumController(instrum.AbstractMibInstrumController):
    def read_variables(self, *varBinds, **context):
        return [
            (ov[0], v2c.OctetString("You queried OID %s" % ov[0])) for ov in varBinds
        ]


# Create a custom Management Instrumentation Controller and register at
# SNMP Context under ContextName 'my-context'
snmpContext.register_context_name(
    v2c.OctetString("my-context"),  # Context Name
    EchoMibInstrumController(),  # Management Instrumentation
)

# Register GET&SET Applications at the SNMP engine for a custom SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)

# Register an imaginary never-ending job to keep I/O dispatcher running forever
snmpEngine.transport_dispatcher.job_started(1)

# Run I/O dispatcher which would receive queries and send responses
try:
    snmpEngine.open_dispatcher()
except:
    snmpEngine.close_dispatcher()
    raise

Download script.

See also: library reference.