【测试】python 模拟snmp-se
如下脚本，模拟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
相关文章