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
|
from pysnmp.entity import config
from pysnmp.entity.rfc3413 import ntforg, context, mibvar
from pysnmp.entity.rfc3413.oneliner import cmdgen
# Auth protocol
usmHMACMD5AuthProtocol = cmdgen.usmHMACMD5AuthProtocol
usmHMACSHAAuthProtocol = cmdgen.usmHMACSHAAuthProtocol
usmNoAuthProtocol = cmdgen.usmNoAuthProtocol
# Privacy protocol
usmDESPrivProtocol = cmdgen.usmDESPrivProtocol
usmNoPrivProtocol = cmdgen.usmNoPrivProtocol
# Credentials
CommunityData = cmdgen.CommunityData
UsmUserData = cmdgen.UsmUserData
# Transport
UdpTransportTarget = cmdgen.UdpTransportTarget
class AsynNotificationOriginator(cmdgen.AsynCommandGenerator):
def __init__(self, snmpEngine=None, snmpContext=None):
cmdgen.AsynCommandGenerator.__init__(self, snmpEngine)
self.snmpContext = snmpContext
self.__knownAuths = {}
def asyncSendNotification(
self, authData, transportTarget, notifyType,
notificationType, varBinds=None
):
tagList = 'notify-list'
addrName, paramsName = cmdgen.AsynCommandGenerator._configure(
self, authData, transportTarget, tagList
)
notifyName = '%s-name' % tagList
if not self.__knownAuths.has_key(authData):
config.addNotificationTarget(
self.snmpEngine,
notifyName,
paramsName,
tagList,
notifyType
)
config.addContext(
self.snmpEngine, ''
)
config.addTrapUser(
self.snmpEngine,
authData.securityModel,
authData.securityName,
authData.securityLevel,
(1,3,6)
)
if self.snmpContext is None:
self.snmpContext = context.SnmpContext(self.snmpEngine)
self.__knownAuths[authData] = 1
if varBinds:
__varBinds = []
for varName, varVal in varBinds:
name, oid = mibvar.mibNameToOid(
self.mibViewController, varName
)
__varBinds.append((name + oid, varVal))
else:
__varBinds = None
return ntforg.NotificationOriginator(self.snmpContext).sendNotification(
self.snmpEngine, notifyName, notificationType, __varBinds
)
class NotificationOriginator(AsynNotificationOriginator):
def sendNotification(
self, authData, transportTarget, notifyType,
notificationType, varBinds=None
):
errorIndication = self.asyncSendNotification(
authData, transportTarget, notifyType, notificationType, varBinds
)
if errorIndication:
return errorIndication
self.snmpEngine.transportDispatcher.runDispatcher()
|
