本文整理汇总了Python中pysnmp.entity.config.addVacmUser函数的典型用法代码示例。如果您正苦于以下问题:Python addVacmUser函数的具体用法?Python addVacmUser怎么用?Python addVacmUser使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了addVacmUser函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, objects):
self._snmpEngine = engine.SnmpEngine()
config.addSocketTransport( self._snmpEngine, udp.domainName, udp.UdpTransport().openServerMode((_addr, _port)))
config.addV3User(self._snmpEngine,_account,config.usmHMACMD5AuthProtocol,_auth_key,config.usmDESPrivProtocol,_priv_key)
config.addVacmUser(self._snmpEngine, 3, _account, "authPriv",(1,3,6,1,4,1), (1,3,6,1,4,1))
self._snmpContext = context.SnmpContext(self._snmpEngine)
#builder create
mibBuilder = self._snmpContext.getMibInstrum().getMibBuilder()
mibSources = mibBuilder.getMibSources() + (builder.DirMibSource('.'),)+(builder.DirMibSource(filepath),)
mibBuilder.setMibSources(*mibSources)
MibScalarInstance, = mibBuilder.importSymbols('SNMPv2-SMI','MibScalarInstance')
for mibObject in objects:
nextVar, = mibBuilder.importSymbols(mibObject.mibName,
mibObject.objectType)
instance = createVariable(MibScalarInstance, mibObject.valueGetFunc, nextVar.name, (0,), nextVar.syntax)
#need to export as <var name>Instance
instanceDict = {str(nextVar.name)+"Instance":instance}
mibBuilder.exportSymbols(mibObject.mibName, **instanceDict)
cmdrsp.GetCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.SetCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.NextCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.BulkCommandResponder(self._snmpEngine, self._snmpContext)
开发者ID:ggs134,项目名称:pysnmp,代码行数:31,代码来源:agent.py
示例2: __init__
def __init__(self, mibObjects, sqlObject, _rootDir, server_options):
"""
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(('', int(server_options['port']))))
#add a v2 user with the community string public
config.addV1System(self._snmpEngine, "agent", server_options['community'])
#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, int(server_options['version']), "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(os.path.join(_rootDir, 'lib_mib_py')),)
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)
if mibObject.objMib.flag:
#je suis une table
for client in sqlObject.getClientsId():
instance = createVariable(MibScalarInstance, mibObject.objMib, mibObject.valueFunc, nextVar.name,(client['ClientId'],), nextVar.syntax)
listName = list(nextVar.name)
listName.append(client['ClientId'] )
newName = tuple(listName)
instanceDict = {str(newName)+"Instance":instance}
mibBuilder.exportSymbols(mibObject.mibName, **instanceDict)
else :
instance = createVariable(MibScalarInstance, mibObject.objMib, mibObject.valueFunc, nextVar.name,(0,), nextVar.syntax)
#class ,class with fonc , nom de la fonction , oid , type d'oid
#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)
开发者ID:voileux,项目名称:bacula.snmpd,代码行数:59,代码来源:server.py
示例3: configure
def configure(self, snmpEngine, authData, transportTarget, notifyType,
contextName, **options):
cache = self._getCache(snmpEngine)
notifyName = None
# Create matching transport tags if not given by user. Not good!
if not transportTarget.tagList:
transportTarget.tagList = str(
hash((authData.securityName, transportTarget.transportAddr))
)
if isinstance(authData, CommunityData) and not authData.tag:
authData.tag = transportTarget.tagList.split()[0]
addrName, paramsName = self._cmdGenLcdCfg.configure(
snmpEngine, authData, transportTarget, contextName, **options)
tagList = transportTarget.tagList.split()
if not tagList:
tagList = ['']
for tag in tagList:
notifyNameKey = paramsName, tag, notifyType
if notifyNameKey in cache['name']:
notifyName, paramsName, useCount = cache['name'][notifyNameKey]
cache['name'][notifyNameKey] = notifyName, paramsName, useCount + 1
else:
notifyName = 'n%s' % self.nextID()
config.addNotificationTarget(
snmpEngine, notifyName, paramsName, tag, notifyType)
cache['name'][notifyNameKey] = notifyName, paramsName, 1
authDataKey = authData.securityName, authData.securityModel, authData.securityLevel, contextName
if authDataKey in cache['auth']:
authDataX, subTree, useCount = cache['auth'][authDataKey]
cache['auth'][authDataKey] = authDataX, subTree, useCount + 1
else:
subTree = (1, 3, 6)
config.addVacmUser(
snmpEngine,authData.securityModel, authData.securityName,
authData.securityLevel, (), (), subTree,
contextName=contextName)
cache['auth'][authDataKey] = authData, subTree, 1
return notifyName
开发者ID:etingof,项目名称:pysnmp,代码行数:57,代码来源:lcd.py
示例4: __init__
def __init__(self, host, port, rcommunity):
self.snmpEngine = engine.SnmpEngine()
config.addSocketTransport(self.snmpEngine, udp.domainName, udp.UdpTransport().openServerMode((host, port)))
config.addV1System(self.snmpEngine, 'my-area', rcommunity)
config.addVacmUser(self.snmpEngine, 2, 'my-area', 'noAuthNoPriv', (1, 3, 6))
self.snmpContext = context.SnmpContext(self.snmpEngine)
self.mibBuilder = self.snmpContext.getMibInstrum().getMibBuilder()
self.MibScalar, self.MibScalarInstance = self.mibBuilder.importSymbols('SNMPv2-SMI', 'MibScalar', 'MibScalarInstance')
cmdrsp.GetCommandResponder(self.snmpEngine, self.snmpContext)
cmdrsp.NextCommandResponder(self.snmpEngine, self.snmpContext)
cmdrsp.BulkCommandResponder(self.snmpEngine, self.snmpContext)
开发者ID:gabejohnson,项目名称:snmp-simulator,代码行数:11,代码来源:simulator.py
示例5: _add_user_permission
def _add_user_permission(self, OID):
"""
Add user permission to OID - readOnly
"""
OID = tuple(int(x) for x in OID.split("."))
if self.users:
for user in self.users:
print("add user permission %s %s " % (str(user), str(OID)))
config.addVacmUser(self.snmpEngine, 3, str(user[0]), str(user[3]), OID)
else:
# Allow full MIB access for this user / securityModels at VACM
config.addVacmUser(self.snmpEngine, 1, "my-read-area", "noAuthNoPriv", OID)
开发者ID:jeroenmaelbrancke,项目名称:openvstorage,代码行数:12,代码来源:server.py
示例6: setTrapReceiver
def setTrapReceiver(self, host, 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')
开发者ID:vincentwang901130,项目名称:gr-llsr,代码行数:15,代码来源:llsrSnmpAgent.py
示例7: __init__
def __init__(self, host, port, log_queue):
self.log_queue = log_queue
# Create SNMP engine
self.snmpEngine = engine.SnmpEngine()
# Transport setup
udp_sock = gevent.socket.socket(gevent.socket.AF_INET, gevent.socket.SOCK_DGRAM)
udp_sock.setsockopt(gevent.socket.SOL_SOCKET, gevent.socket.SO_BROADCAST, 1)
udp_sock.bind((host, port))
# UDP over IPv4
self.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp_sock
)
#SNMPv1
config.addV1System(self.snmpEngine, 'public-read', 'public')
# SNMPv3/USM setup
# user: usr-md5-des, auth: MD5, priv DES
config.addV3User(
self.snmpEngine, 'usr-md5-des',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usmDESPrivProtocol, 'privkey1'
)
# user: usr-sha-none, auth: SHA, priv NONE
config.addV3User(
self.snmpEngine, 'usr-sha-none',
config.usmHMACSHAAuthProtocol, 'authkey1'
)
# user: usr-sha-aes128, auth: SHA, priv AES/128
config.addV3User(
self.snmpEngine, 'usr-sha-aes128',
config.usmHMACSHAAuthProtocol, 'authkey1',
config.usmAesCfb128Protocol, 'privkey1'
)
# Allow full MIB access for each user at VACM
config.addVacmUser(self.snmpEngine, 1, 'public-read', 'noAuthNoPriv',
(1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-md5-des', 'authPriv',
(1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-sha-none', 'authNoPriv',
(1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-sha-aes128', 'authPriv',
(1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
# Get default SNMP context this SNMP engine serves
snmpContext = context.SnmpContext(self.snmpEngine)
# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(self.snmpEngine, snmpContext)
开发者ID:Root-nix,项目名称:conpot,代码行数:56,代码来源:snmp_command_responder.py
示例8: initTarget
def initTarget(host='127.0.0.1', port=162, community='LIC_OSS'):
#global snmpEngine, snmpContext, ntfOrg
# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()
# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', community)
# Specify security settings per SecurityName (SNMPv2c -> 1)
config.addTargetParams(snmpEngine, 'my-creds', 'my-area', 'noAuthNoPriv', 1)
# Setup transport endpoint and bind it with security settings yielding
# a target name
config.addSocketTransport(
snmpEngine,
udp.domainName,
udp.UdpSocketTransport().openClientMode()
)
config.addTargetAddr(
snmpEngine, 'my-nms',
udp.domainName, (host, port),
'my-creds',
tagList='all-my-managers'
)
# Specify what kind of notification should be sent (TRAP or INFORM),
# to what targets (chosen by tag) and what filter should apply to
# the set of targets (selected by tag)
config.addNotificationTarget(
snmpEngine, 'my-notification', 'my-filter', 'all-my-managers', 'trap'
)
# Allow NOTIFY access to Agent's MIB by this SNMP model (2), securityLevel
# and SecurityName
config.addContext(snmpEngine, '')
config.addVacmUser(snmpEngine, 2, 'my-area', 'noAuthNoPriv', (), (), (1,3,6))
# *** SNMP engine configuration is complete by this line ***
# Create default SNMP context where contextEngineId == SnmpEngineId
snmpContext = context.SnmpContext(snmpEngine)
# Create Notification Originator App instance.
ntfOrg = ntforg.NotificationOriginator(snmpContext)
return snmpEngine, ntfOrg
开发者ID:lowitty,项目名称:alarmfiles,代码行数:45,代码来源:SnmpTraps.py
示例9: run
def run(self):
snmpEngine = engine.SnmpEngine()
config.addSocketTransport(
snmpEngine,
udp.domainName,
udp.UdpTransport().openServerMode(('127.0.0.1',
self.__listening_port))
)
config.addV1System(
snmpEngine, 'my-area', 'public', contextName='my-context')
config.addVacmUser(snmpEngine=snmpEngine,
securityModel=2,
securityName='my-area',
securityLevel='noAuthNoPriv',
readSubTree=SNMPAgentResponder.OID_PREFIX,
writeSubTree=(),
notifySubTree=())
snmpContext = context.SnmpContext(snmpEngine)
snmpContext.registerContextName(
v2c.OctetString('my-context'), # Context Name
self.__responder # Management Instrumentation
)
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
snmpEngine.transportDispatcher.jobStarted(1)
self.__barrier.wait()
# TODO with statement here!
try:
snmpEngine.transportDispatcher.runDispatcher()
except:
snmpEngine.transportDispatcher.closeDispatcher()
raise
开发者ID:eugpermar,项目名称:rb_monitor,代码行数:39,代码来源:snmp_agent.py
示例10: __init__
def __init__(self, mibObjects):
# Each SNMP-based application has an engine
self._snmpEngine = engine.SnmpEngine()
# Open a UDP socket to listen for snmp requests (requset sudo command)
config.addSocketTransport(self._snmpEngine,
udp.domainName,
udp.UdpTransport().openServerMode(('', 161)))
config.addV1System(self._snmpEngine, 'agent', 'public')
# add a v2 user with the community string public
config.addVacmUser(self._snmpEngine, 2, 'agent', 'noAuthNoPriv',
readSubTree=(1, 3, 6, 1, 4, 1),
writeSubTree=(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
(MibTable, MibTableRow, MibTableColumn,
MibScalarInstance) = mibBuilder.importSymbols('SNMPv2-SMI',
'MibTable',
'MibTableRow',
'MibTableColumn',
'MibScalarInstance')
# import and maintain Table
maintaintable = maintainTableThread(0, mibObjects, mibBuilder,
MibScalarInstance)
maintaintable.start()
# tell pysnmp to respotd to get, getnext, and getbulk
cmdrsp.GetCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.SetCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.NextCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.BulkCommandResponder(self._snmpEngine, self._snmpContext)
开发者ID:vincentwang901130,项目名称:gr-llsr,代码行数:38,代码来源:llsrSnmpAgent.py
示例11: __init__
def __init__(self, mibObjects):
self._snmpEngine = engine.SnmpEngine()
config.addSocketTransport(self._snmpEngine, udp.domainName, udp.UdpTransport().openServerMode(('',165)))
config.addV1System(self._snmpEngine,"my-read-area","public")
config.addV1System(self._snmpEngine,"my-write-area","private")
config.addVacmUser(self._snmpEngine, 2,"my-read-area",'noAuthNoPriv',readSubTree=(1,3,6,1,4,1))
config.addVacmUser(self._snmpEngine, 2,"my-write-area",'noAuthNoPriv',readSubTree=(1,3,6,1,4,1), writeSubTree=(1,3,6,1,4,1))
self._snmpContext = context.SnmpContext(self._snmpEngine)
mibBuilder = self._snmpContext.getMibInstrum().getMibBuilder()
mibSources = mibBuilder.getMibSources()+(builder.DirMibSource('.'),)+(builder.DirMibSource('./pysnmp_mibs'),)
mibBuilder.setMibSources(*mibSources)
MibScalarInstance, = mibBuilder.importSymbols('SNMPv2-SMI','MibScalarInstance')
for mibObject in mibObjects:
nextVar, = mibBuilder.importSymbols(mibObject.mibName, mibObject.objectType)
instance = createVariable(MibScalarInstance, mibObject.valueGetFunc, mibObject.valueSetFunc, nextVar.name, (0,), nextVar.syntax)
instanceDict ={ str(nextVar.name)+"Instance":instance }
mibBuilder.exportSymbols(mibObject.mibName, **instanceDict)
cmdrsp.GetCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.NextCommandResponder(self._snmpEngine, self._snmpContext)
cmdrsp.BulkCommandResponder(self._snmpEngine, self._snmpContext)
开发者ID:ggs134,项目名称:pysnmp,代码行数:23,代码来源:pysa.py
示例12: sent
udp.domainName, ('127.0.0.1', 162),
'my-creds',
tagList='all-my-managers'
)
# Specify what kind of notification should be sent (TRAP or INFORM),
# to what targets (chosen by tag) and what filter should apply to
# the set of targets (selected by tag)
config.addNotificationTarget(
snmpEngine, 'my-notification', 'my-filter', 'all-my-managers', 'trap'
)
# Allow NOTIFY access to Agent's MIB by this SNMP model (1), securityLevel
# and SecurityName
config.addContext(snmpEngine, '')
config.addVacmUser(snmpEngine, 1, 'my-area', 'noAuthNoPriv', (), (), (1,3,6))
@trollius.coroutine
def snmpOperation(snmpEngine, target, snmpContext, contextName,
notificationName, instanceIndex, additionalVarBinds):
future = ntforg.NotificationOriginator().sendVarBinds(
snmpEngine,
target,
snmpContext,
contextName,
notificationName,
instanceIndex,
additionalVarBinds
)
# We know we are sending TRAP which will never produce any response.
开发者ID:ww9rivers,项目名称:pysnmp,代码行数:31,代码来源:trap-v1.py
示例13: control
snmpEngine, 'usr-md5-3des',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usm3DESEDEPrivProtocol, 'privkey1'
)
#
# Access control (VACM) setup
#
# Configure VACM from the scratch
# default context
config.addContext(snmpEngine, '')
# allow full MIB access for each user
config.addVacmUser(snmpEngine, 1, 'my-area', 'noAuthNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 2, 'my-area', 'noAuthNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-des', 'authPriv', (1,3,6), (1,3,6))
#
# CommandResponder could serve multiple independent MIB trees
# selected by ContextName parameter. The default ContextName is
# an empty string, this is where SNMP engine's LCD also lives.
#
snmpContext = context.SnmpContext(snmpEngine)
# Register SNMP Applications at the SNMP engine
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)
开发者ID:carmackjia,项目名称:pysnmp,代码行数:31,代码来源:cmdrsp.py
示例14:
#
# Install default Agent configuration
# Install default Agent configuration
#config.setInitialVacmParameters(snmpEngine)
#
# Apply initial VACM configuration to this user
#config.addVacmGroup(snmpEngine, "initial", 3, "usr-md5-des")
# Alternatively, configure VACM from the scratch
# default context
config.addContext(snmpEngine, '')
# allow full MIB access for each user
config.addVacmUser(snmpEngine, 1, 'my-area', 'noAuthNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 2, 'my-area', 'noAuthNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-des', 'authPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-none-none', 'noAuthNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-none', 'authNoPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-sha-aes128', 'authPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-aes256', 'authPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-aes192', 'authPriv', (1,3,6), (1,3,6))
config.addVacmUser(snmpEngine, 3, 'usr-md5-3des', 'authPriv', (1,3,6), (1,3,6))
#
# CommandResponder could serve multiple independent MIB trees
# selected by ContextName parameter. The default ContextName is
# an empty string, this is where SNMP engine's LCD also lives.
#
snmpContext = context.SnmpContext(snmpEngine)
开发者ID:carmackjia,项目名称:pysnmp,代码行数:31,代码来源:cmdrsp.py
示例15:
# UDP over IPv4
config.addTransport(
snmpEngine,
udp.domainName,
udp.UdpTwistedTransport().openServerMode(('127.0.0.1', 161))
)
# SNMPv1 setup
# SecurityName <-> CommunityName mapping.
# Here we configure two distinct CommunityName's to control read and write
# operations.
config.addV1System(snmpEngine, 'my-read-area', 'public')
config.addV1System(snmpEngine, 'my-write-area', 'private')
# Allow full MIB access for this user / securityModels at VACM
config.addVacmUser(snmpEngine, 1, 'my-read-area', 'noAuthNoPriv', (1, 3, 6, 1, 2, 1))
config.addVacmUser(snmpEngine, 1, 'my-write-area', 'noAuthNoPriv', (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
# Get default SNMP context this SNMP engine serves
snmpContext = context.SnmpContext(snmpEngine)
# Register SNMP Applications at the SNMP engine for particular SNMP context
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
# Run Twisted main loop
reactor.run()
开发者ID:bbmorten,项目名称:pysnmp,代码行数:29,代码来源:multiple-snmp-communities.py
示例16: __init__
def __init__(self, host, port, mibpaths):
self.oid_mapping = {}
self.databus_mediator = DatabusMediator(self.oid_mapping)
# mapping between OID and databus keys
# Create SNMP engine
self.snmpEngine = engine.SnmpEngine()
# path to custom mibs
mibBuilder = self.snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder
mibSources = mibBuilder.getMibSources()
for mibpath in mibpaths:
mibSources += (builder.DirMibSource(mibpath),)
mibBuilder.setMibSources(*mibSources)
# Transport setup
udp_sock = gevent.socket.socket(gevent.socket.AF_INET, gevent.socket.SOCK_DGRAM)
udp_sock.setsockopt(gevent.socket.SOL_SOCKET, gevent.socket.SO_BROADCAST, 1)
udp_sock.bind((host, port))
self.server_port = udp_sock.getsockname()[1]
# UDP over IPv4
self.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp_sock
)
# SNMPv1
config.addV1System(self.snmpEngine, 'public-read', 'public')
# SNMPv3/USM setup
# user: usr-md5-des, auth: MD5, priv DES
config.addV3User(
self.snmpEngine, 'usr-md5-des',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usmDESPrivProtocol, 'privkey1'
)
# user: usr-sha-none, auth: SHA, priv NONE
config.addV3User(
self.snmpEngine, 'usr-sha-none',
config.usmHMACSHAAuthProtocol, 'authkey1'
)
# user: usr-sha-aes128, auth: SHA, priv AES/128
config.addV3User(
self.snmpEngine, 'usr-sha-aes128',
config.usmHMACSHAAuthProtocol, 'authkey1',
config.usmAesCfb128Protocol, 'privkey1'
)
# Allow full MIB access for each user at VACM
config.addVacmUser(self.snmpEngine, 1, 'public-read', 'noAuthNoPriv',
readSubTree=(1, 3, 6, 1, 2, 1), writeSubTree=(1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 2, 'public-read', 'noAuthNoPriv',
readSubTree=(1, 3, 6, 1, 2, 1), writeSubTree=(1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-md5-des', 'authPriv',
readSubTree=(1, 3, 6, 1, 2, 1), writeSubTree=(1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-sha-none', 'authNoPriv',
readSubTree=(1, 3, 6, 1, 2, 1), writeSubTree=(1, 3, 6, 1, 2, 1))
config.addVacmUser(self.snmpEngine, 3, 'usr-sha-aes128', 'authPriv',
readSubTree=(1, 3, 6, 1, 2, 1), writeSubTree=(1, 3, 6, 1, 2, 1))
# Get default SNMP context this SNMP engine serves
snmpContext = context.SnmpContext(self.snmpEngine)
# Register SNMP Applications at the SNMP engine for particular SNMP context
self.resp_app_get = conpot_cmdrsp.c_GetCommandResponder(self.snmpEngine, snmpContext, self.databus_mediator, host, port)
self.resp_app_set = conpot_cmdrsp.c_SetCommandResponder(self.snmpEngine, snmpContext, self.databus_mediator, host, port)
self.resp_app_next = conpot_cmdrsp.c_NextCommandResponder(self.snmpEngine, snmpContext, self.databus_mediator, host, port)
self.resp_app_bulk = conpot_cmdrsp.c_BulkCommandResponder(self.snmpEngine, snmpContext, self.databus_mediator, host, port)
开发者ID:agnivesh,项目名称:conpot,代码行数:71,代码来源:command_responder.py
示例17: __init__
def __init__(self, mibPath, temperatureValue, snmpRelays, criticalStatus=True):
from types import ListType, TupleType,StringTypes
from re import compile,search
from socket import gethostbyname
extractPaths=compile(r'[,:]')
checkIP=compile(r'(\d{1,3}\.){3}\d{1,3}')
# Create SNMP engine instance
self.snmpEngine = engine.SnmpEngine()
if not temperatureValue:
raise ValueError, 'A temperature must be provided'
self.temperature=temperatureValue
#print "============>mibPath type: %s" %type(mibPath)
if type(mibPath) in StringTypes:
mibPathTuple=tuple(extractPaths.split(mibPath))
elif type(mibPath) in (ListType, TupleType):
mibPathTuple=tuple(mibPath)
else:
mibPathTuple=('/usr/local/share/snmp/python/',)
mibBuilder = self.snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder
#print mibPathTuple
mibSources = mibBuilder.getMibPath() + mibPathTuple
mibBuilder.setMibPath(*mibSources)
mibBuilder.loadModules( 'USC-IGFAE-MIB' )
if type(snmpRelays) in StringTypes:
snmpRelays=snmpRelays.split(',')
elif not type(snmpRelays) in (ListType,TupleType):
raise TypeError, 'The list of SNMP relays must be a string or a list or tuple of strings'
(temperatureCritical, temperatureOK, self.roomTemp) = mibBuilder.importSymbols('USC-IGFAE-MIB','temperatureCritical', 'temperatureOK', 'roomTemp' )
# SecurityName <-> CommunityName mapping
config.addV1System(self.snmpEngine, 'Arduino', 'ups')
# Specify security settings per SecurityName (SNMPv2c -> 1)
config.addTargetParams(self.snmpEngine, 'creds', 'Arduino', 'noAuthNoPriv', 0)
# Setup transport endpoint and bind it with security settings yielding
# a target name
config.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp.UdpSocketTransport().openClientMode()
)
index=0
for machine in snmpRelays:
index=index+1
if not checkIP.match(machine):
try:
machine=gethostbyname(machine)
except:
continue
#print "==============>SNMP relay IP: %s" % machine
config.addTargetAddr(
self.snmpEngine, 'NMS%s' % index,
udp.domainName, (machine, 162),
'creds',
tagList='managers'
)
# Specify what kind of notification should be sent (TRAP or INFORM),
# to what targets (chosen by tag) and what filter should apply to
# the set of targets (selected by tag)
config.addNotificationTarget(
self.snmpEngine, 'sendShutdownTrap', 'my-filter', 'managers', 'trap'
)
# Allow NOTIFY access to Agent's MIB by this SNMP model (2), securityLevel
# and SecurityName
config.addContext(self.snmpEngine, '')
config.addVacmUser(self.snmpEngine, 1, 'Arduino', 'noAuthNoPriv',
(), (), (1,3,6))
# *** SNMP engine configuration is complete by this line ***
# Create default SNMP context where contextEngineId == SnmpEngineId
snmpContext = context.SnmpContext(self.snmpEngine)
if criticalStatus:
self.trap=temperatureCritical
else:
self.trap=temperatureOK
# Create Notification Originator App instance.
ntforg.NotificationOriginator.__init__(self,snmpContext)
开发者ID:vfalbor,项目名称:nagios-arduino,代码行数:95,代码来源:Send_Trap.py
示例18:
config.addV3User(
snmpEngine, 'test-user',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usmDESPrivProtocol, 'privkey1'
# config.usmAesCfb128Protocol, 'privkey1'
)
# Install default Agent configuration
#config.setInitialVacmParameters(snmpEngine)
#
# Apply initial VACM configuration to this user
#config.addVacmGroup(snmpEngine, "initial", 3, "test-user")
# Alternatively, configure VACM from the scratch
config.addContext(snmpEngine, '')
config.addVacmUser(snmpEngine, 1, 'test-agent', 'noAuthNoPriv',
(1,3,6), (1,3,6)) # v1
config.addVacmUser(snmpEngine, 2, 'test-agent', 'noAuthNoPriv',
(1,3,6), (1,3,6)) # v2c
config.addVacmUser(snmpEngine, 3, 'test-user', 'authPriv',
(1,3,6), (1,3,6)) # v3
# SNMP context
snmpContext = context.SnmpContext(snmpEngine)
# Apps registration
cmdrsp.GetCommandResponder(snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(snmpEngine, snmpContext)
snmpEngine.transportDispatcher.jobStarted(1) # this job would never finish
snmpEngine.transportDispatcher.runDispatcher()
开发者ID:chrisliom,项目名称:pysnmp,代码行数:32,代码来源:cmdrsp.py
示例19:
config.USM_PRIV_CBC56_DES, 'privkey1'
)
# user: usr-sha-none, auth: SHA, priv NONE
config.addV3User(
snmpEngine, 'usr-sha-none',
config.USM_AUTH_HMAC96_SHA, 'authkey1'
)
# user: usr-sha-none, auth: SHA, priv AES
config.addV3User(
snmpEngine, 'usr-sha-aes128',
config.USM_AUTH_HMAC96_SHA, 'authkey1',
config.USM_PRIV_CFB128_AES, 'privkey1'
)
# Allow full MIB access for each user at VACM
config.addVacmUser(snmpEngine, 3, 'usr-md5-des', 'authPriv', (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
config.addVacmUser(snmpEngine, 3, 'usr-sha-none', 'authNoPriv', (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
config.addVacmUser(snmpEngine, 3, 'usr-sha-aes128', 'authPriv', (1, 3, 6, 1, 2, 1), (1, 3, 6, 1, 2, 1))
# Get default SNMP context this SNMP engine serves
snmpContext = context.SnmpContext(snmpEngine)
# 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)
# Run Twisted main loop
reactor.run()
开发者ID:etingof,项目名称:pysnmp,代码行数:30,代码来源:multiple-usm-users.py
示例20:
snmpEngine,
transportDomain,
udp.UdpTransport().openServerMode(transportAddress)
)
# SNMPv3/USM setup
# user: usr-md5-des, auth: MD5, priv DES
config.addV3User(
snmpEngine, 'usr-md5-des',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usmDESPrivProtocol, 'privkey1'
)
# Allow full MIB access for this user / securityModels at VACM
config.addVacmUser(snmpEngine, 3, 'usr-md5-des', 'authPriv', (1, 3, 6), (1, 3, 6, 1, 2, 1))
# Get default SNMP context this SNMP engine serves
snmpContext = context.SnmpContext(snmpEngine)
# 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
transportDispatcher.jobStarted(1)
# Run I/O dispatcher which would receive queries and send responses
try:
开发者ID:bbmorten,项目名称:pysnmp,代码行数:31,代码来源:multiple-snmp-engines.py
注:本文中的pysnmp.entity.config.addVacmUser函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论