【测试】python 模拟snmp-se

发布时间：2019-08-28 09:06:24编辑：auto阅读（1154）

如下脚本，模拟windows和linux上的snmp-agent，返回数据给snmp请求者。


直接上代码吧：

# -*- coding: utf-8 -*-
import binascii, struct
import socket
import time
from threading import Thread

'''
windows OID
1.3.6.1.2.1.25.2.3.1.6.1 [ObjectIdentifier] //硬盘 
1.3.6.1.2.1.25.2.3.1.6.2 [ObjectIdentifier] //硬盘 
1.3.6.1.2.1.25.2.3.1.6.3 [ObjectIdentifier] //硬盘 
1.3.6.1.2.1.25.2.3.1.6.4 [ObjectIdentifier] //硬盘 
1.3.6.1.2.1.25.2.3.1.6.5 [ObjectIdentifier] //光盘 
1.3.6.1.2.1.25.2.3.1.6.6 [ObjectIdentifier] //光盘 
1.3.6.1.2.1.25.2.3.1.6.7 [ObjectIdentifier] //虚拟内存 
1.3.6.1.2.1.25.2.3.1.6.8 [ObjectIdentifier] //物理内存
OID号不固定，当只有一块硬盘，一个光驱时，物理内存占用OID为1.2.5（2.1占用一般0，不是真实硬盘，不知道是什么意思，真正硬盘占用从2.2开始）
'''

#a 是一个真实的请求内容，def test_parse（）函数可以将他解析出来,并打印出密码，请求类型，请求ID，OID对象
a = '''0x30, 0x82, 0x01, 0x09, 0x02, 0x01, 0x01, 0x04, 
0x06, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0xa0, 
0x81, 0xfb, 0x02, 0x04, 0x4a, 0xbb, 0x2b, 0xac, 
0x02, 0x01, 0x00, 0x02, 0x01, 0x00, 0x30, 0x81, 
0xec, 0x30, 0x0d, 0x06, 0x09, 0x2b, 0x06, 0x01, 
0x02, 0x01, 0x19, 0x02, 0x02, 0x00, 0x05, 0x00, 
0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 0x01, 0x02, 
0x01, 0x19, 0x03, 0x03, 0x01, 0x02, 0x01, 0x05, 
0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 0x01, 
0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 0x05, 0x01, 
0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 
0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 0x05, 
0x02, 0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 
0x06, 0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 
0x05, 0x03, 0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 
0x2b, 0x06, 0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 
0x01, 0x05, 0x04, 0x05, 0x00, 0x30, 0x0f, 0x06, 
0x0b, 0x2b, 0x06, 0x01, 0x02, 0x01, 0x19, 0x02, 
0x03, 0x01, 0x06, 0x01, 0x05, 0x00, 0x30, 0x0f, 
0x06, 0x0b, 0x2b, 0x06, 0x01, 0x02, 0x01, 0x19, 
0x02, 0x03, 0x01, 0x06, 0x02, 0x05, 0x00, 0x30, 
0x0f, 0x06, 0x0b, 0x2b, 0x06, 0x01, 0x02, 0x01, 
0x19, 0x02, 0x03, 0x01, 0x06, 0x03, 0x05, 0x00, 
0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 0x01, 0x02, 
0x01, 0x19, 0x02, 0x03, 0x01, 0x06, 0x04, 0x05, 
0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 0x01, 
0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 0x06, 0x05, 
0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 0x06, 
0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 0x06, 
0x06, 0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 0x2b, 
0x06, 0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 0x01, 
0x06, 0x07, 0x05, 0x00, 0x30, 0x0f, 0x06, 0x0b, 
0x2b, 0x06, 0x01, 0x02, 0x01, 0x19, 0x02, 0x03, 
0x01, 0x06, 0x08, 0x05, 0x00'''

class snmpReqParse:
    struct_type = {0x02:'INTEGER_TYPE',
                   0x30:'SQUENCE_TYPE',
                   0x04:'OCTET_TYPE',
                   0X05:'NULL_TYPE', 
                   0x06:'OBJID_TYPE', 
                   0xa0:'GET_REQ_TYPE', 
                   0xa5:'GET_BULKREQ_TYPE', 
                   0Xa2:'RESPONSE_TYPE'}
    def __init__(self, request):
        self.request = request
        self.objidList = []
        self.requestID = None
        self.getReqType = 0xa0
        self.publicPasswd = ''
        self.parse()
    def parse(self):
        currentPos = self.parse_header1()
        if not currentPos:
            print "parse SNMP header failed!"
            return
        currentPos = self.parse_header2(currentPos)
        if not currentPos:
            print "parse SNMP header failed!"
            return
        self.parse_obj(currentPos)    
    def parse_obj(self, pos):
        #开始解析obj对象了    
        currentPos = pos
        stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]
        currentPos += 1
        currentPos = self.struct_parse(currentPos)[0] 
        while len(self.request[currentPos:]) != 0:
            stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]
            currentPos += 1
            currentPos = self.struct_parse(currentPos)[0]
            stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]
            if stype != 0x06:
                print "wrong type coding, must be 0x06"
                break
            currentPos += 1
            currentPos, stringLen = self.struct_parse(currentPos)
            objid = struct.unpack(stringLen*'B', self.request[currentPos:currentPos+stringLen])
            objid1stStr = str(divmod(objid[0], 40)[0]) + '.' + str(divmod(objid[0], 40)[1])
            #当obj中每一段中有值大于127时，则后面一个数应该和前面一个数拼起来，例如0x8F和0X65，应该是0x8F*128+65
            templist = []
            temp1 = 0
            for i in objid[1:]:
                if not temp1:
                    if i<=127:
                        templist.append(str(i))
                    else:
                        temp1 = i
                else:
                    templist.append(str((temp1-128)*128+i))
                    temp1 = 0
            objidStr = objid1stStr + '.' + '.'.join(templist)
            self.objidList.append(objidStr)
            currentPos = currentPos+stringLen
            #跳过null type
            currentPos += 2
        #print self.objidList   
    def parse_header2(self, pos):
        #从团体属性密钥结束后，解析到OBject开始
        currentPos = pos
        stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]
        self.getReqType = stype
        currentPos += 1
        currentPos = self.struct_parse(currentPos)[0]
        stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]    
        if stype == 0x02:
            startPos = currentPos + 1
            currentPos, stringLen = self.struct_parse(startPos)
            self.requestID = self.request[currentPos:currentPos+stringLen]
            currentPos += stringLen
            #直接跳过error-statues和error-index
            currentPos += 6
            return currentPos
        else:
            print 'wrong type coding ,should be 0x02!'
            return
    def parse_header1(self):
        #start parse，从开始到得到团体属性密钥
        #currentPos为读取数据流的位置
        stype = struct.unpack('B', self.request[0:1])[0]
        if stype == 0x30:
            currentPos = self.struct_parse(1)[0]
            stype = struct.unpack('B', self.request[currentPos:currentPos+1])[0]
            if stype == 0x02:
                startPos = currentPos+1
                currentPos = self.struct_parse(startPos)[0]
                if currentPos - startPos == 1:
                    #不做判断了，直接跳到解析团体关键字
                    startPos = currentPos + 2
                    currentPos, stringLen = self.struct_parse(startPos)
                    self.publicPasswd = self.request[startPos+1:startPos+stringLen+1]
                    currentPos = startPos+stringLen+1 
                    return currentPos
                else:
                    print "wrong SNMP version ，must be v2c!"
            else:
                print "wrong type coding: %s,should be 0x02"%stype
        else:
            print "wrong type coding: %s, should be 0x30 "%stype
    def struct_parse(self, startpos):
        #结构解析，startpos的位置已经是长度字段了，解析出来，最后返回这个结构结尾的pos
        length = struct.unpack('B', self.request[startpos:startpos+1])[0]
        if length == 0x81:
            endpos = startpos + 2
            slen = struct.unpack('B', self.request[startpos+1:endpos])[0]
        elif length == 0x82:
            endpos = startpos + 3
            slen = struct.unpack('!H', self.request[startpos+1:endpos])[0]
        elif length <= 127:
            endpos = startpos + 1
            slen = length
        else:
            print "wrong length coding: "%length
            return
        if len(self.request[endpos:]) < slen:
            print "string is not enough long!"
            return
        return endpos, slen

class snmpResponse:
    OID = {'windows':['1.3.6.1.2.1.25.2.2.0',  #内存总大小(mem_count)，integar类型
                  '1.3.6.1.2.1.25.3.3.1.2.1',  #CPU,integar类型
                  '1.3.6.1.2.1.25.2.3.1.5.1',  #值为0，integar类型
                  '1.3.6.1.2.1.25.2.3.1.5.2',  #硬盘总大小，*4096单位为字节数，integar类型
                  '1.3.6.1.2.1.25.2.3.1.5.3',  #光驱总大小，实际为光盘容量，没有光盘时为0，integar类型
                  '1.3.6.1.2.1.25.2.3.1.5.4',  #暂时不清楚是什么，默认我给它赋值0x009a5f，integar类型
                  '1.3.6.1.2.1.25.2.3.1.6.1',  #值为0，integar类型
                  '1.3.6.1.2.1.25.2.3.1.6.2',  #硬盘占用的大小，integar类型
                  '1.3.6.1.2.1.25.2.3.1.6.3',  #光驱大小，和5.3一致
                  '1.3.6.1.2.1.25.2.3.1.6.4',  #虚拟内存大小，一般不关注，integar类型
                  '1.3.6.1.2.1.25.2.3.1.6.5'], #物理内存使用大小（mem_use），integar类型
       #内存使用率计算方法：(mem_use*65535)/（mem_count*1024）
       'linux':['1.3.6.1.4.1.2021.11.11.0',  #CPU剩余，如40
                '1.3.6.1.4.1.2021.4.5.0',  #内存总大小，integar类型
                '1.3.6.1.4.1.2021.4.6.0',  #内存剩余大小，integar类型
                '1.3.6.1.4.1.2021.9.1.9.1']} #硬盘使用百分比，如60, integar类型
    def __init__(self):
        self.getReqType = 0x00
        self.reqPublicPasswd = ''
        self.reqObjList = []
        self.OIDdic = {}
        self.getResType = 0xa2
        self.respondID = 0
        self.error_index = 0
        self.error_status = 0
        self.simDev = { 'publicPasswd':'public', 
                        'platform':'windows', 'CPULOAD':45,
                        'totalMem':1024000, 'usedMem':102400,  #mem单位为kB
                        'totalDisk':512000000, 'usedDisk':100000000}
    def check_request(self):
        if self.getReqType != 0xa0:
            print "this tool can only support get-request"
            return
        if self.reqObjList:
            if self.reqObjList[0] in self.OID['windows']:
                self.simDev['platform'] = 'windows'
                for i in range(0, len(self.reqObjList)):
                    if self.reqObjList[i] not in self.OID['windows']:
                        self.error_index = i + 1
                        self.error_status = 0x02
                        return
                useMemVal = (self.simDev['usedMem']*1024)/65535  
                self.OIDdic = {'1.3.6.1.2.1.25.2.2.0':self.simDev['totalMem'], #内存总大小(mem_count)，integar类型
                               '1.3.6.1.2.1.25.3.3.1.2.1':self.simDev['CPULOAD'],  #CPU,integar类型
                              '1.3.6.1.2.1.25.2.3.1.5.1':0,  #值为0，integar类型
                              '1.3.6.1.2.1.25.2.3.1.5.2':self.simDev['totalDisk']/2048,  #硬盘总大小，*4096单位为字节数，integar类型
                              '1.3.6.1.2.1.25.2.3.1.5.3':0,  #光驱总大小，实际为光盘容量，没有光盘时为0，integar类型
                              '1.3.6.1.2.1.25.2.3.1.5.4':39519,  #暂时不清楚是什么，默认我给它赋值0x009a5f，integar类型
                              '1.3.6.1.2.1.25.2.3.1.6.1':0,  #值为0，integar类型
                              '1.3.6.1.2.1.25.2.3.1.6.2':self.simDev['usedDisk']/2048,  #硬盘占用的大小，integar类型
                              '1.3.6.1.2.1.25.2.3.1.6.3':0,  #光驱大小，和5.3一致
                              '1.3.6.1.2.1.25.2.3.1.6.4':10835,  #虚拟内存大小，一般不关注，integar类型
                              '1.3.6.1.2.1.25.2.3.1.6.5':useMemVal}
            elif self.reqObjList[0] in self.OID['linux']:
                self.simDev['platform'] = 'linux'
                for i in self.reqObjList:
                    if i not in self.OID['linux']:
                        print 'can not support the OID:', i
                        return
                freeCPU = 100 - self.simDev['CPULOAD']
                usedDisk = (self.simDev['usedDisk']*100)/self.simDev['totalDisk']
                freeMemVal = self.simDev['totalMem'] - self.simDev['usedMem']
                self.OIDdic = {'1.3.6.1.4.1.2021.11.11.0':freeCPU,  #CPU剩余
                               '1.3.6.1.4.1.2021.4.5.0':self.simDev['totalMem'],  #内存总大小，integar类型
                               '1.3.6.1.4.1.2021.4.6.0':freeMemVal,  #内存剩余大小，integar类型
                               '1.3.6.1.4.1.2021.9.1.9.1':usedDisk}   #硬盘使用百分比，如60, integar类型
            else:
                print "the objlist will not be supported!"
                return
        else:
            print "error: objlist is Null!" 
            return
    def build_objbuff(self):
        objbuff = ''    
        if self.error_index:
                for i in self.reqObjList:
                    objbuff += self.coding_obj_and_value(i)
        else:
            for i in self.reqObjList:
                objbuff += self.coding_obj_and_value(i, self.OIDdic[i])
        if not objbuff:
            return
        objbuff = '\x30' + self.coding_of_length(len(objbuff)) + objbuff
        return objbuff        
    def build_snmpbuff(self):
        self.check_request()
        objBuff = self.build_objbuff()
        if not objBuff:
            return
        snmpbuff = ''
        if self.simDev['publicPasswd'] != self.reqPublicPasswd:
            print "public keyword is wrong!"
            return
        else:
            snmpbuff = '\x02' + struct.pack('B', len(self.respondID)) + self.respondID \
                       + '\x02\x01' + struct.pack('B', self.error_status) \
                       + '\x02\x01' + struct.pack('B', self.error_index)
            snmpbuff += objBuff
            snmpbuff = '\xa2' + self.coding_of_length(len(snmpbuff)) + snmpbuff
            snmpbuff = '\x02\x01\x01\x04' + struct.pack('B', len(self.simDev['publicPasswd'])) + self.simDev['publicPasswd'] + snmpbuff
            snmpbuff = '\x30' + self.coding_of_length(len(snmpbuff)) + snmpbuff
            return snmpbuff
    def coding_obj_and_value(self, obj, val=None):
        temps = ''
        temps += self.coding_obj(obj)
        temps = struct.pack('2B', 0x06, len(temps)) + temps
        if val == None:
            temps += '\x05\x00'
        elif str(type(val)) == '<type \'str\'>':
            temps += '\x04' + struct.pack('B', len(val)) + val
        elif str(type(val)) in ['<type \'int\'>', '<type \'long\'>']:
            if val<=2**7:
                temps += '\x02\x01' + struct.pack('B', val)
            elif 2**7<val<=2**15:
                temps += '\x02\x02' + struct.pack('!H', val)
            elif 2**15<val<=2**23:
                temps += '\x02\x03' + struct.pack('!I', val)[1:]
            elif 2**23<val<=2**31:
                temps += '\x02\x04' + struct.pack('!I', val)  
            else:
                print "value is too big!"
                return
        temps = '\x30' + struct.pack('B', len(temps)) + temps
        return temps 
    def coding_of_length(self, slen):
        #SNMP中域长度的编码方式。此函数直接返回长度字段的编码结果
        if slen<=127:
            return struct.pack('B', slen)
        if 127<slen<256:
            return struct.pack('2B', 0x81, slen)
        if 256<=slen<256**2:
            return struct.pack('!BH', 0x82, slen)            
    def coding_obj(self, obj):
        #将1.3.6.X的OID打包成字节流
        objlist = obj.split('.')
        objtemplist = []
        objfir1byte = int(objlist[0])*40 + int(objlist[1])
        objtemplist.append(struct.pack('B', objfir1byte))
        for i in objlist[2:]:
            if int(i)>127:
                a, b = divmod(int(i), 128)
                objtemplist.append(struct.pack('2B', a+128, b))
            else:
                objtemplist.append(struct.pack('B', int(i)))
        return ''.join(objtemplist)
    
    
def SNMP_server(serverIP = '', passWd = 'public', SNMPPORT = 161):
    usock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    usock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    usock.bind((serverIP, SNMPPORT))
    while 1:
        message, clientaddr = usock.recvfrom(8192)
        req = snmpReqParse(message)
        res = snmpResponse()
        res.reqPublicPasswd = req.publicPasswd
        res.simDev = read_config()[serverIP]
        res.getReqType = req.getReqType
        res.respondID = req.requestID
        res.reqObjList = req.objidList
        resbuff = res.build_snmpbuff()
        if resbuff:
            usock.sendto(resbuff, clientaddr)
            memUsage = res.simDev['usedMem']*100/res.simDev['totalMem']
            diskUsage = res.simDev['usedDisk']*100/res.simDev['totalDisk']
            if not res.error_index:
                print '########### %s ###########\n\
<%s>\n\
>>you request oid are %s\n\
>>server CPU load percentage is : %s\n\
>>server mem used percentage is : %s\n\
>>server disk used percentage is: %s'%(serverIP, time.ctime(), res.reqObjList,
                      res.simDev['CPULOAD'], memUsage, diskUsage)
            
def test_parse():
    global a
    b = a.replace(', 0x', r'\x')
    c = '\\' + b.replace(', \n0x', r'\x')[1:]
    d = c.replace(r'\x', '')
    request = binascii.a2b_hex(d)
    test = snmpReqParse(request)
    print test.simDev('publicPasswd')
    print hex(test.getReqType)
    print test.requestID
    for i in test.objidList:
        print i
def test_response():
    global a
    b = a.replace(', 0x', r'\x')
    c = '\\' + b.replace(', \n0x', r'\x')[1:]
    d = c.replace(r'\x', '')
    request = binascii.a2b_hex(d)
    test = snmpReqParse(request)
    res = snmpResponse('public')
    res.reqPublicPasswd = test.simDev('publicPasswd')
    res.getReqType = test.getReqType
    res.respondID = test.requestID
    res.reqObjList = test.objidList
    resbuff = res.build_snmpbuff()
    print repr(resbuff)
def read_config():
    import ConfigParser
    cf = ConfigParser.ConfigParser()
    cf.read('snmpServer.ini')
    simSers = {}
    for sec in cf.sections():
        simSers[sec] = {'publicPasswd':cf.get(sec, 'password'),
                       'CPULOAD': cf.getint(sec, 'cpuload'),
                       'totalMem':cf.getint(sec, 'totalMem'), 
                       'usedMem':cf.getint(sec, 'usedMem'),
                       'totalDisk':cf.getint(sec, 'totalDisk'), 
                       'usedDisk':cf.getint(sec, 'usedDisk')}
    return simSers
   
def main():
    sersDic = read_config()
    for s in sersDic.keys():
        print repr(s), 'is listening!'
        t = Thread(target=SNMP_server, args=(s, ))
        t.start()
        time.sleep(1)
       
if __name__=='__main__':
    #test_parse()
    #test_response()
    #serverIP = '172.16.1.102'
    #SNMP_server(serverIP)
    print 'start'
    main()
    raw_input('')




以上服务启动依赖配置文件，配置文件中指定模拟的linux或者windows服务器，可以填写多组，文件存为snmpServer.ini

[192.168.10.102]
password = public
cpuload = 11
totalMem = 1024000
usedMem = 102400
totalDisk = 512000000
usedDisk = 100000000

[172.16.1.102]
password = public
cpuload = 30
totalMem = 1024000
usedMem = 204800
totalDisk = 512000000
usedDisk = 300000000

关键字：