BrokenPipe

Code and the other things

Writing an SNMP Agent With a Custom MIB Using Pysnmp

| Comments

In this post I’m going to go through the process of exposing a custom SNMP MIB using pysnmp. I assume you’re already familiar with SNMP concepts and how to write a MIB. This post will take you through the process of getting your MIB into a format pysnmp likes and then writing a little python program to serve up that MIB using pysnmp. The program will also be able to send a trap defined in the MIB.

I had a hard time finding good documentation or examples for how to do this. pysnmp has some nice documentation for their one-liner modules, but seems to be lacking full examples for the rest of their API. The code below works, but there may be a better way to do things. If you’re more familiar with pysnmp, leave a comment and let me know.

Let’s start with the prerequisites. You obviously need to install pysnmp. You can grab it with pip or easy_install. It’ll bring in a couple other modules with it. If you’re running Windows, you may have some trouble with PyCrypto, but you can find binaries here.

Once that’s installed, grab a copy of libsmi. This is used to convert the MIB into a format for pysnmp. You may also want to grab the standard snmp utilities to test our Agent. Your package manager will have them, or head over to net-snmp.

That’s it for pre-requisites. The next step is to convert our MIB file to a Python module that pysnmp can use. We’ll be using the MIB below. I assume it’s in a file called MY-MIB in the current working directory.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
MY-MIB DEFINITIONS ::= BEGIN

IMPORTS
        OBJECT-TYPE, Integer32, NOTIFICATION-TYPE, enterprises
                     FROM SNMPv2-SMI
;

myCompany       OBJECT IDENTIFIER ::= {enterprises 42}

testCount OBJECT-TYPE
    SYNTAX Integer32
    MAX-ACCESS read-only
    STATUS current
    DESCRIPTION "A sample count of something."
    ::= {myCompany 1}

testDescription OBJECT-TYPE
    SYNTAX OCTET STRING
    MAX-ACCESS read-only
    STATUS current
    DESCRIPTION "A description of something"
    ::= {myCompany 2}

testTrap NOTIFICATION-TYPE
    STATUS current
    DESCRIPTION "Test notification"
    ::= {myCompany 3}

END

libsmi and pysnmp should have installed some some stuff in /usr/local/bin, unless you told them otherwise. The important things here are smilint and build-pysnmp-mib. They should be somewhere on your PATH.

First, let’s make sure that our MIB is well formed. To do that run

1
2
$ smilint ./MY-MIB -s
./MY-MIB:29: [2] missing MODULE-IDENTITY clause in SMIv2 MIB

We’ll ignore that warning for this simple example. If you get an error (indicated by a [1]), you need to fix it before you proceed. Now we’re ready to convert this MIB to a Python module pysnmp can read. To do that run

1
2
$ build-pysnmp-mib -o MY-MIB.py MY-MIB
$

This will result in a new file called MY-MIB.py. If you want to query our agent, you’ll want to add a copy of the MIB to a place that net-snmp searches for MIBs. I added the MY-MIB file to ~/.snmp/mibs.

Ok, so our MIB is ready to use in pysnmp. Below I have the code for our agent. It’s a pretty big chunk of code, but we’ll go through the important bits. The comments in the code should help explain things as well.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
from pysnmp.entity import engine, config
from pysnmp import debug
from pysnmp.entity.rfc3413 import cmdrsp, context, ntforg
from pysnmp.carrier.asynsock.dgram import udp
from pysnmp.smi import builder

import threading
import collections
import time

#can be useful
#debug.setLogger(debug.Debug('all'))

MibObject = collections.namedtuple('MibObject', ['mibName',
                                   'objectType', 'valueFunc'])

class Mib(object):
    """Stores the data we want to serve. 
    """

    def __init__(self):
        self._lock = threading.RLock()
        self._test_count = 0

    def getTestDescription(self):
        return "My Description"

    def getTestCount(self):
        with self._lock:
            return self._test_count

    def setTestCount(self, value):
        with self._lock:
            self._test_count = value


def createVariable(SuperClass, getValue, *args):
    """This is going to create a instance variable that we can export. 
    getValue is a function to call to retreive the value of the scalar
    """
    class Var(SuperClass):
        def readGet(self, name, *args):
            return name, self.syntax.clone(getValue())
    return Var(*args)


class SNMPAgent(object):
    """Implements an Agent that serves the custom MIB and
    can send a trap.
    """

    def __init__(self, mibObjects):
        """
        mibObjects - a list of MibObject tuples that this agent
        will serve
        """

        #each SNMP-based application has an engine
        self._snmpEngine = engine.SnmpEngine()

        #open a UDP socket to listen for snmp requests
        config.addSocketTransport(self._snmpEngine, udp.domainName,
                                  udp.UdpTransport().openServerMode(('', 161)))

        #add a v2 user with the community string public
        config.addV1System(self._snmpEngine, "agent", "public")
        #let anyone accessing 'public' read anything in the subtree below,
        #which is the enterprises subtree that we defined our MIB to be in
        config.addVacmUser(self._snmpEngine, 2, "agent", "noAuthNoPriv",
                           readSubTree=(1,3,6,1,4,1))

        #each app has one or more contexts
        self._snmpContext = context.SnmpContext(self._snmpEngine)

        #the builder is used to load mibs. tell it to look in the
        #current directory for our new MIB. We'll also use it to
        #export our symbols later
        mibBuilder = self._snmpContext.getMibInstrum().getMibBuilder()
        mibSources = mibBuilder.getMibSources() + (builder.DirMibSource('.'),)
        mibBuilder.setMibSources(*mibSources)

        #our variables will subclass this since we only have scalar types
        #can't load this type directly, need to import it
        MibScalarInstance, = mibBuilder.importSymbols('SNMPv2-SMI',
                                                      'MibScalarInstance')
        #export our custom mib
        for mibObject in mibObjects:
            nextVar, = mibBuilder.importSymbols(mibObject.mibName,
                                                mibObject.objectType)
            instance = createVariable(MibScalarInstance,
                                      mibObject.valueFunc,
                                      nextVar.name, (0,),
                                      nextVar.syntax)
            #need to export as <var name>Instance
            instanceDict = {str(nextVar.name)+"Instance":instance}
            mibBuilder.exportSymbols(mibObject.mibName,
                                     **instanceDict)

        # tell pysnmp to respotd to get, getnext, and getbulk
        cmdrsp.GetCommandResponder(self._snmpEngine, self._snmpContext)
        cmdrsp.NextCommandResponder(self._snmpEngine, self._snmpContext)
        cmdrsp.BulkCommandResponder(self._snmpEngine, self._snmpContext)


    def setTrapReceiver(self, host, community):
        """Send traps to the host using community string community
        """
        config.addV1System(self._snmpEngine, 'nms-area', community)
        config.addVacmUser(self._snmpEngine, 2, 'nms-area', 'noAuthNoPriv',
                           notifySubTree=(1,3,6,1,4,1))
        config.addTargetParams(self._snmpEngine,
                               'nms-creds', 'nms-area', 'noAuthNoPriv', 1)
        config.addTargetAddr(self._snmpEngine, 'my-nms', udp.domainName,
                             (host, 162), 'nms-creds',
                             tagList='all-my-managers')
        #set last parameter to 'notification' to have it send
        #informs rather than unacknowledged traps
        config.addNotificationTarget(
            self._snmpEngine, 'test-notification', 'my-filter',
            'all-my-managers', 'trap')


    def sendTrap(self):
        print "Sending trap"
        ntfOrg = ntforg.NotificationOriginator(self._snmpContext)
        errorIndication = ntfOrg.sendNotification(
            self._snmpEngine,
            'test-notification',
            ('MY-MIB', 'testTrap'),
            ())


    def serve_forever(self):
        print "Starting agent"
        self._snmpEngine.transportDispatcher.jobStarted(1)
        try:
           self._snmpEngine.transportDispatcher.runDispatcher()
        except:
            self._snmpEngine.transportDispatcher.closeDispatcher()
            raise

class Worker(threading.Thread):
    """Just to demonstrate updating the MIB
    and sending traps
    """

    def __init__(self, agent, mib):
        threading.Thread.__init__(self)
        self._agent = agent
        self._mib = mib
        self.setDaemon(True)

    def run(self):
        while True:
            time.sleep(3)
            self._mib.setTestCount(mib.getTestCount()+1)
            self._agent.sendTrap()

if __name__ == '__main__':
    mib = Mib()
    objects = [MibObject('MY-MIB', 'testDescription', mib.getTestDescription),
               MibObject('MY-MIB', 'testCount', mib.getTestCount)]
    agent = SNMPAgent(objects)
    agent.setTrapReceiver('192.168.1.14', 'traps')
    Worker(agent, mib).start()
    try:
        agent.serve_forever()
    except KeyboardInterrupt:
        print "Shutting down"

I have MIB and Worker classes that aren’t terribly important, but are used to make the example complete. The class SNMPAgent is the important code here. On lines 59-80 we do some initial configuration. This agent supports SNMP v2c and exposes a community string ‘public’ that has read permissions on the subtree .1.3.6.1.4.1 (the enterprises subtree).

Next, on line 79 we tell pysnmp to check the current directory for our newly created MIB module.

Lines 84-97 export our MIB objects. Each mib object subclasses MibScalarInstance (this code doesn’t support tables). The createVariable function will return a subclass of MibScalarInstance that gets its value from a method call to the Mib object. The builder is then used to export the object.

Lines 100-102 tell pysnmp to handle get, getnext, and getbulk requests.

To be able to send traps, we need to tell pysnmp about a trap target. This is done in the setTrapReceiver method. The actual sending of traps is done by NotificationOriginator in the sendTrap method.

If we run this, the Worker thread should increment the testCount object and send a trap every three seconds. You probably want to change the IP address passed to setTrapReceiver on line 164. Start the agent up and we’ll test it out. With the agent started, run

1
2
3
4
5
$ snmpwalk -m MY-MIB -v 2c -c public localhost .1
MY-MIB::testCount.0 = INTEGER: 0
MY-MIB::testDescription.0 = STRING: "My Description"
MY-MIB::testDescription.0 = No more variables left in this MIB View (It is past the end of the MIB tree)
$

We were able to walk our MIB! In order to see the traps, make sure you edit snmptrapd.conf on your trap receiver to accept traps to the traps community string. For testing, you can add

1
authCommunity log traps

to your snmptrapd.conf file and run

1
$ snmptrapd -f -Lo

You should see the traps received on the standard output log.

That’s it. You’re serving a custom MIB with pysnmp.

Comments