本文整理汇总了Python中pysnmp.entity.config.addV1System函数的典型用法代码示例。如果您正苦于以下问题:Python addV1System函数的具体用法?Python addV1System怎么用?Python addV1System使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了addV1System函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
def __init__(self, udpIp, udpPort):
# Create SNMP engine with autogenernated engineID and pre-bound
# to socket transport dispatcher
self.snmpEngine = engine.SnmpEngine()
self.mibBuilder = self.snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder
mibPath = self.mibBuilder.getMibPath() + ('.',)
self.mibBuilder.setMibPath(*mibPath)
# Setup UDP over IPv4 transport endpoint
config.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp.UdpSocketTransport().openServerMode((udpIp, udpPort))
)
print 'Publishing readings via SNMP'
print 'Agent address {}:{}'.format(udpIp, udpPort)
print 'Community name public'
# v1/2 setup
config.addV1System(self.snmpEngine, 'test-agent', 'public')
# v3 setup
config.addV3User(
self.snmpEngine, 'test-user'
)
# VACM setup
config.addContext(self.snmpEngine, '')
config.addRwUser(self.snmpEngine, 1, 'test-agent', 'noAuthNoPriv', (1,3,6)) # v1
config.addRwUser(self.snmpEngine, 2, 'test-agent', 'noAuthNoPriv', (1,3,6)) # v2c
config.addRwUser(self.snmpEngine, 3, 'test-user', 'noAuthNoPriv', (1,3,6)) # v3
# SNMP context
snmpContext = context.SnmpContext(self.snmpEngine)
# Apps registration
cmdrsp.GetCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.SetCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.NextCommandResponder(self.snmpEngine, snmpContext)
cmdrsp.BulkCommandResponder(self.snmpEngine, snmpContext)
MibScalarInstance, = self.mibBuilder.importSymbols('SNMPv2-SMI', 'MibScalarInstance')
class ScalarFromCallback(MibScalarInstance):
def __init__(self, sensorId, valueGetter, typeName, instId, syntax):
MibScalarInstance.__init__(self, typeName, instId, syntax)
self.valueGetter = valueGetter
def readTest(self, name, val, idx, (acFun, acCtx)):
if not self.valueGetter():
raise error.NoAccessError(idx=idx, name=name)
def readGet(self, name, val, idx, (acFun, acCtx)):
value = self.valueGetter()
if not value:
raise error.NoAccessError(idx=idx, name=name)
else:
return name, self.syntax.clone(value)
开发者ID:InviNets,项目名称:ReadingsToSNMP,代码行数:60,代码来源:publish_sensor_readings.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: test2
def test2():
from pysnmp.v4.proto.rfc1902 import ObjectName
from pysnmp.entity import engine, config
from pysnmp.carrier.asynsock.dgram import udp
from pysnmp.entity.rfc3413 import cmdgen
sysName = ObjectName("1.3.6.1.2.1.1.5.0")
ip = "192.168.1.9"
snmp_engine = engine.SnmpEngine()
config.addV1System(snmp_engine, 'test-agent', "public")
config.addTargetParams(snmp_engine, 'myParams', 'test-agent', 'noAuthNoPriv', 0)
config.addTargetAddr(
snmp_engine, 'myRouter', config.snmpUDPDomain,
(ip, 161), 'myParams'
)
config.addSocketTransport(
snmp_engine,
udp.domainName,
udp.UdpSocketTransport().openClientMode()
)
cb = {}
def cbFun(sendRequestHandle, errorIndication, errorStatus, errorIndex,
varBinds, cbCtx):
cbCtx['errorIndication'] = errorIndication
cbCtx['errorStatus'] = errorStatus
cbCtx['errorIndex'] = errorIndex
cbCtx['varBinds'] = varBinds
cmdgen.GetCommandGenerator().sendReq(snmp_engine, 'myRouter', ((sysName, None),), cbFun, cb)
lastmemusage = 0
lastrefs = None
errors = 0
while (errors < 2):
snmp_engine.transportDispatcher.runDispatcher()
print cb['varBinds'][0][1]
snmp_engine.transportDispatcher.closeDispatcher()
#asynCommandGenerator.flushConfig()
newmemusage = resource.getrusage(resource.RUSAGE_SELF)[2]
memdiff = (newmemusage - lastmemusage)
newrefs = get_refcounts()
if memdiff > 0:
print "Leaked %d Kb... printing refcount diff" % memdiff
if lastrefs == None:
print "No previous refcount, skipping"
else:
print_ref_diffs(lastrefs, newrefs)
errors = errors + 1
gc.collect()
lastrefs = newrefs
lastmemusage = newmemusage
#print resource.getrusage(resource.RUSAGE_SELF)[3]
time.sleep(1)
开发者ID:benroeder,项目名称:HEN,代码行数:57,代码来源:clean_switch_memory_test.py
示例4: configure_engine
def configure_engine(engine, host='127.0.0.1', port=162):
# UDP over IPv4, first listening interface/port
config.addTransport(
engine,
udp.domainName + (1,),
udp.UdpTransport().openServerMode((host, port))
)
config.addV1System(engine, 'my-area', 'public')
开发者ID:lwolf,项目名称:tcw122b-scripts,代码行数:9,代码来源:app.py
示例5: _configureUsers
def _configureUsers(self, snmpEngine, snmpContext, params):
logger.debug ( 'Configure users' );
for user in params.users:
logger.debug ( 'Creating user "%s"', user.name );
# Compute a fake context name
contextName = "%sSystem-context-%s" % (self.version,user.name)
self.contexts[user.name] = contextName;
# Register the community
config.addV1System(snmpEngine=snmpEngine, securityName=contextName, communityName=user.name, contextName=contextName)
开发者ID:GillesBouissac,项目名称:agentcluster,代码行数:9,代码来源:snmpsetup.py
示例6: cfgCmdGen
def cfgCmdGen(self, authData, transportTarget, tagList=''):
if self.__knownAuths.has_key(authData):
paramsName = self.__knownAuths[authData]
else:
paramsName = 'p%s' % nextID()
if isinstance(authData, CommunityData):
config.addV1System(
self.snmpEngine,
authData.securityName,
authData.communityName
)
config.addTargetParams(
self.snmpEngine, paramsName,
authData.securityName, authData.securityLevel,
authData.mpModel
)
elif isinstance(authData, UsmUserData):
config.addV3User(
self.snmpEngine,
authData.securityName,
authData.authProtocol, authData.authKey,
authData.privProtocol, authData.privKey
)
config.addTargetParams(
self.snmpEngine, paramsName,
authData.securityName, authData.securityLevel
)
else:
raise error.PySnmpError('Unsupported SNMP version')
self.__knownAuths[authData] = paramsName
if not self.__knownTransports.has_key(transportTarget.transportDomain):
transport = transportTarget.openClientMode()
config.addSocketTransport(
self.snmpEngine,
transportTarget.transportDomain,
transport
)
self.__knownTransports[transportTarget.transportDomain] = transport
k = transportTarget, tagList
if self.__knownTransportAddrs.has_key(k):
addrName = self.__knownTransportAddrs[k]
else:
addrName = 'a%s' % nextID()
config.addTargetAddr(
self.snmpEngine, addrName,
transportTarget.transportDomain,
transportTarget.transportAddr,
paramsName,
transportTarget.timeout * 100,
transportTarget.retries,
tagList
)
self.__knownTransportAddrs[k] = addrName
return addrName, paramsName
开发者ID:rolfschr,项目名称:testerman,代码行数:57,代码来源:cmdgen.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: __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
示例9: __init__
def __init__(self, host, port, community):
self.snmp = engine.SnmpEngine()
self.snmp.registerTransportDispatcher(dispatch.TwistedDispatcher())
config.addV1System(self.snmp, 'my-area', community)
config.addTargetParams(self.snmp,
'my-creds', 'my-area', 'noAuthNoPriv', 0)
config.addSocketTransport(self.snmp,
udp.domainName, udp.UdpTwistedTransport().openClientMode()
)
config.addTargetAddr(self.snmp, 'my-router', udp.domainName,
(host, port), 'my-creds')
开发者ID:calston,项目名称:tensor,代码行数:12,代码来源:snmp.py
示例10: __init__
def __init__(self, host, port, users, naming_scheme):
"""
host = public ip to listen on
port = port to listen on (usually 161)
users = list of ('username', 'password', 'privatekey', 'authPriv') #authentication method for snmp v3
if users is None, authentication will be snmp v1 public community string, read only
"""
self.naming_scheme = naming_scheme
self.run = True
self.users = users
# Create SNMP engine
self.snmpEngine = engine.SnmpEngine()
# Get default SNMP context this SNMP engine serves
self.snmpContext = context.SnmpContext(self.snmpEngine)
# MIB builder
self.mibBuilder = self.snmpContext.getMibInstrum().getMibBuilder()
self.MibScalar, self.MibScalarInstance = self.mibBuilder.importSymbols('SNMPv2-SMI', 'MibScalar', 'MibScalarInstance')
# Transport setup
# UDP over IPv4
try:
config.addSocketTransport(self.snmpEngine,
udp.domainName,
udp.UdpTransport().openServerMode((host, port)))
print('Serving on port %s' % port)
except error.CarrierError as carrier_error :
if "[Errno 98]" in carrier_error.message:
raise RuntimeError('Port %s is in use' % port)
# SNMPv3/USM setup
# user: usr-md5-des, auth: MD5, priv DES
if users:
for user in users:
self._add_v3_md5_des_user(user)
# Allow full MIB access for each user at VACM
else:
# SNMPv1 public community string setup
config.addV1System(self.snmpEngine, 'my-read-area', 'public')
self._add_user_permission("1.3.6.1.2.1") #full walk permission, without this snmpwalk returns None
# Overwrite default strings with custom name
sysDescr, = self.snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder.importSymbols('SNMPv2-MIB', 'sysDescr')
sysDescr = self.MibScalarInstance(sysDescr.name,
(0,),
sysDescr.syntax.clone("PySNMP engine - OVS 1.2.0 SNMP Agent")) # Get from config?
self.snmpEngine.msgAndPduDsp.mibInstrumController.mibBuilder.exportSymbols('SNMPv2-MIB', sysDescr)
self._add_user_permission(self.naming_scheme.replace('.%s', ''))
开发者ID:JasperLue,项目名称:openvstorage,代码行数:49,代码来源:server.py
示例11: 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
示例12: __init__
def __init__(self, community, ip, version=1):
self.__community = community
self.__ip = ip
self.__version = version
self.__errorIndication = None
self.__errorStatus = None
self.__errorIndex = None
self.__varBinds = None
# self.__lock = threading.Lock()
self.__snmp_engine = engine.SnmpEngine()
config.addV1System(self.__snmp_engine, "test-agent", self.__community)
config.addTargetParams(self.__snmp_engine, "myParams", "test-agent", "noAuthNoPriv", self.__version)
config.addTargetAddr(self.__snmp_engine, "myRouter", config.snmpUDPDomain, (self.__ip, 161), "myParams")
config.addSocketTransport(self.__snmp_engine, udp.domainName, udp.UdpSocketTransport().openClientMode())
开发者ID:benroeder,项目名称:HEN,代码行数:15,代码来源:snmp.py
示例13: 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
示例14: __init__
def __init__(self):
self.unReadyNodes = list()
self.identifiers = dict()
self.snmpEngine = engine.SnmpEngine()
self.bcmdgen = cmdgen.BulkCommandGenerator()
self.snmpEngine.registerTransportDispatcher(dispatch.TwistedDispatcher())
config.addV1System(self.snmpEngine, 'test-agent', SNMP_COMMUNITY)
config.addTargetParams(self.snmpEngine, 'myParams', 'test-agent', 'noAuthNoPriv', 1)
config.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp.UdpTwistedTransport().openClientMode()
)
self.carbonFact = CarbonFactory(self)
reactor.connectTCP(GRAPHITE_HOST, 2003, self.carbonFact)
开发者ID:butangero,项目名称:irtg,代码行数:18,代码来源:nodecollection.py
示例15: __init__
def __init__(self):
self.snmpEngine = engine.SnmpEngine()
config.addSocketTransport(
self.snmpEngine,
udp.domainName,
udp.UdpSocketTransport().openServerMode(('0.0.0.0', 162))
)
config.addV1System(self.snmpEngine, 'test-agent', 'public')
config.addV3User(
self.snmpEngine, 'test-user',
config.usmHMACMD5AuthProtocol, 'authkey1',
config.usmDESPrivProtocol, 'privkey1'
# '80004fb81c3dafe69' # ContextEngineID of Notification Originator
)
# Apps registration
ntfrcv.NotificationReceiver(self.snmpEngine, self.recvcallback)
self.snmpEngine.transportDispatcher.jobStarted(1) # this job would never finish
self.snmpEngine.transportDispatcher.runDispatcher()
开发者ID:Nucleoos,项目名称:quinn,代码行数:21,代码来源:snmptraprcv.py
示例16: main
def main(argv):
# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()
dispatcher = TornadoDispatcher()
snmpEngine.registerTransportDispatcher(dispatcher)
# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', 'public')
# Specify security settings per SecurityName (SNMPv1 - 0, SNMPv2c - 1)
config.addTargetParams(snmpEngine, 'my-creds', 'my-area', 'noAuthNoPriv', 1)
# UDP/IPv4
config.addSocketTransport(
snmpEngine,
udp.domainName,
udp.UdpSocketTransport().openClientMode()
)
config.addTargetAddr(
snmpEngine,
'my-router',
udp.domainName,
(argv[0], 161),
'my-creds',
timeout=3.0,
retryCount=1
)
cbCtx = dict(dispatcher=dispatcher)
cmdGen = cmdgen.GetCommandGenerator()
cmdGen.sendReq(
snmpEngine,
'my-router',
( ('1.3.6.1.2.1.1.1.0', None), ),
cbFun,
cbCtx
)
IOLoop.instance().start()
开发者ID:the-invoice,项目名称:pysnmp_tornado,代码行数:40,代码来源:tornado_snmp_get.py
示例17: 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
示例18: __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
示例19: v1TargetName
def v1TargetName(
self, engine,
ip, port=161,
community='public',
snmpVersion='2',
):
"""Find/create target name for v1/v2 connection to given agent"""
key = (community,snmpVersion=='1')
paramName = self._v1ParamCache.get( key )
if paramName is None:
nameID = self._newV1Name()
name = 'v1sys-%s'%(nameID)
config.addV1System(engine, name, community)
paramName = 'v1param-%s'%(nameID)
if snmpVersion == '1':
version = 0
else:
version = 1
config.addTargetParams(
engine, paramName, name, 'noAuthNoPriv', version
)
self._v1ParamCache[ key ] = paramName
return self._targetName( engine, ip, port, paramName )
开发者ID:mmattice,项目名称:TwistedSNMP,代码行数:23,代码来源:agentproxy.py
示例20: __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
注:本文中的pysnmp.entity.config.addV1System函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论