diff options
Diffstat (limited to 'examples/v3arch')
36 files changed, 1620 insertions, 0 deletions
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py new file mode 100644 index 0000000..6eebaa9 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py @@ -0,0 +1,42 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv2c +# * with community name 'public' +# * over IPv4/UDP +# * send INFORM notification +# * with TRAP ID 'coldStart' specified as a MIB symbol +# * include managed object information specified as a MIB symbol +# * perform response OIDs and values resolution at MIB +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'inform', + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + ).addVarBinds( + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysName', 0), 'my system') ) + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py new file mode 100644 index 0000000..071c336 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py @@ -0,0 +1,46 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv3 +# * with user 'usr-md5-none', MD5 auth, no priv +# * send INFORM notification +# * in behalf of contextEngineId 0x8000000004030201, contextName '' +# * over IPv4/UDP +# * with TRAP ID 'warmStart' specified as a string OID +# +# Sending SNMPv3 Notification in behalf of non-default ContextEngineId +# requires having a collection of Managed Objects registered under +# the ContextEngineId being used. +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('localhost', 162)), + ContextData( + rfc1902.OctetString(hexValue='8000000004030201') + ), + 'inform', + NotificationType( + ObjectIdentity('1.3.6.1.6.3.1.1.5.2') + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py new file mode 100644 index 0000000..ed87fdc --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py @@ -0,0 +1,43 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv3 +# * with user 'usr-md5-none', MD5 auth, no priv +# * send INFORM notification +# * in behalf of contextEngineId = SnmpEngineId, contextName 'my-context' +# * over IPv4/UDP +# * with TRAP ID 'warmStart' specified as a string OID +# +# Sending SNMPv3 Notification in behalf of non-default ContextName +# requires having a collection of Managed Objects registered under +# the ContextName being used. +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('localhost', 162)), + ContextData(contextName='my-context'), + 'inform', + NotificationType( + ObjectIdentity('1.3.6.1.6.3.1.1.5.2') + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py new file mode 100644 index 0000000..2996eb5 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py @@ -0,0 +1,44 @@ +## +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv3 +# * with user 'usr-md5-des', auth: MD5, priv DES +# * over IPv4/UDP +# * send INFORM notification +# * with TRAP ID 'warmStart' specified as a string OID +# * include managed object information 1.3.6.1.2.1.1.5.0 = 'system name' +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'inform', + NotificationType( + ObjectIdentity('1.3.6.1.6.3.1.1.5.2') + ).addVarBinds( + ( ObjectType(ObjectIdentity('1.3.6.1.2.1.1.5.0'), + rfc1902.OctetString('system name')) ) + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py new file mode 100644 index 0000000..1d2b825 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py @@ -0,0 +1,36 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv1 +# * with community name 'public' +# * over IPv4/UDP +# * send TRAP notification +# * with Generic Trap #6 (enterpriseSpecific) and Specific Trap 432 +# * overriding Uptime value with 12345 +# * overriding Agent Address with '127.0.0.1' +# * overriding Enterprise OID with 1.3.6.1.4.1.20408.4.1.1.2 +# * include managed object information '1.3.6.1.2.1.1.1.0' = 'my system' +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'trap', + NotificationType( + ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2.0.432'), + ).addVarBinds( + ('1.3.6.1.2.1.1.3.0', 12345), + ('1.3.6.1.6.3.18.1.3.0', '127.0.0.1'), + ('1.3.6.1.6.3.1.1.4.3.0', '1.3.6.1.4.1.20408.4.1.1.2'), + ('1.3.6.1.2.1.1.1.0', rfc1902.OctetString('my system')) + )): + if errorIndication: + print(errorIndication) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py new file mode 100644 index 0000000..a704146 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py @@ -0,0 +1,34 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv1 +# * with community name 'public' +# * over IPv4/UDP +# * send TRAP notification +# * with Generic Trap #1 (warmStart) and Specific Trap 0 +# * with default Uptime +# * with default Agent Address +# * with Enterprise OID 1.3.6.1.4.1.20408.4.1.1.2 +# * include managed object information '1.3.6.1.2.1.1.1.0' = 'my system' +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'trap', + NotificationType( + ObjectIdentity('1.3.6.1.6.3.1.1.5.2') + ).addVarBinds( + ('1.3.6.1.6.3.1.1.4.3.0', '1.3.6.1.4.1.20408.4.1.1.2'), + ('1.3.6.1.2.1.1.1.0', rfc1902.OctetString('my system')) + )): + if errorIndication: + print(errorIndication) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py new file mode 100644 index 0000000..b243f43 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py @@ -0,0 +1,27 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv2c +# * with community name 'public' +# * over IPv4/UDP +# * send TRAP notification +# * with TRAP ID 'coldStart' specified as a MIB symbol +# * include managed object information specified as a MIB symbol +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'trap', + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + )): + if errorIndication: + print(errorIndication) diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py new file mode 100644 index 0000000..bbc9194 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py @@ -0,0 +1,37 @@ +# +# Notification Originator +# +# Send SNMP notification using the following options: +# +# * SNMPv3 +# * with local snmpEngineId = 0x8000000001020304 (must configure at Receiver) +# * with user 'usr-sha-aes128', auth: SHA, priv: AES128 +# * over IPv4/UDP +# * send TRAP notification +# * with TRAP ID 'authenticationFailure' specified as a MIB symbol +# * do not include any additional managed object information +# +# SNMPv3 TRAPs requires pre-sharing the Notification Originator's +# value of SnmpEngineId with Notification Receiver. To facilitate that +# we will use static (e.g. not autogenerated) version of snmpEngineId. +# +from pysnmp.entity.rfc3413.oneliner.ntforg import * +from pysnmp.proto.rfc1902 import OctetString + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in \ + sendNotification(SnmpEngine( + OctetString(hexValue='8000000001020304') + ), + UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1', + authProtocol=usmHMACSHAAuthProtocol, + privProtocol=usmAesCfb128Protocol), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'trap', + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'authenticationFailure') + )): + if errorIndication: + print(errorIndication) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py new file mode 100644 index 0000000..85586ed --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py @@ -0,0 +1,103 @@ +# +# Asynchronous Command Generator +# +# Send SNMP GET requests using multiple independend SNMP engines +# with the following options: +# +# * with SNMPv1, community 'public' and +# with SNMPv2c, community 'public' and +# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy +# * over IPv4/UDP and +# over IPv6/UDP +# * to an Agent at demo.snmplabs.com:161 and +# to an Agent at [::1]:161 +# * for instances of SNMPv2-MIB::sysDescr.0 and +# SNMPv2-MIB::sysLocation.0 MIB objects +# +# Within this script we have a single asynchronous TransportDispatcher +# and a single UDP-based transport serving two independent SNMP engines. +# We use a single instance of AsyncCommandGenerator with each of +# SNMP Engines to comunicate GET command request to remote systems. +# +# When we receive a [response] message from remote system we use +# a custom message router to choose what of the two SNMP engines +# data packet should be handed over. The selection criteria we +# employ here is based on peer's UDP port number. Other selection +# criterias are also possible. +# +from pysnmp.entity.rfc3413.oneliner import cmdgen +from pysnmp.entity import engine +from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher + +# List of targets in the followin format: +# ( ( authData, transportTarget, varNames ), ... ) +targets = ( + # 1-st target (SNMPv1 over IPv4/UDP) + ( cmdgen.CommunityData('public', mpModel=0), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # 2-nd target (SNMPv2c over IPv4/UDP) + ( cmdgen.CommunityData('public'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 1161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # 3-nd target (SNMPv3 over IPv4/UDP) + ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 2161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ) + # N-th target + # ... +) + +# Wait for responses or errors +def cbFun(snmpEngine, sendRequestHandle, errorIndication, + errorStatus, errorIndex, varBinds, cbCtx): + (snmpEngine, authData, transportTarget) = cbCtx + print('snmpEngine %s: %s via %s' % + (snmpEngine.snmpEngineID.prettyPrint(), authData, transportTarget) + ) + if errorIndication: + print(errorIndication) + return 1 + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + return 1 + + for oid, val in varBinds: + if val is None: + print(oid.prettyPrint()) + else: + print('%s = %s' % (oid.prettyPrint(), val.prettyPrint())) + + +# Instantiate the single transport dispatcher object +transportDispatcher = AsyncoreDispatcher() + +# Setup a custom data routing function to select snmpEngine by transportDomain +transportDispatcher.registerRoutingCbFun( + lambda td,ta,d: ta[1] % 3 and 'A' or 'B' +) + +snmpEngineA = engine.SnmpEngine() +snmpEngineA.registerTransportDispatcher(transportDispatcher, 'A') + +snmpEngineB = engine.SnmpEngine() +snmpEngineB.registerTransportDispatcher(transportDispatcher, 'B') + +cmdGen = cmdgen.AsyncCommandGenerator() + +for authData, transportTarget, varBinds in targets: + snmpEngine = transportTarget.getTransportInfo()[1][1] % 3 and \ + snmpEngineA or snmpEngineB + cmdGen.getCmd( + snmpEngine, authData, transportTarget, cmdgen.ContextData(), varBinds, + (cbFun, (snmpEngine, authData, transportTarget)) + ) + +transportDispatcher.runDispatcher() diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py new file mode 100644 index 0000000..e50a54b --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py @@ -0,0 +1,87 @@ +# +# Asynchronous Command Generator +# +# Send a bunch of SNMP GET requests all at once using the following options: +# +# * with SNMPv1, community 'public' and +# with SNMPv2c, community 'public' and +# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy +# * over IPv4/UDP and +# over IPv6/UDP +# * to an Agent at demo.snmplabs.com:161 and +# to an Agent at [::1]:161 +# * for instances of SNMPv2-MIB::sysDescr.0 and +# SNMPv2-MIB::sysLocation.0 MIB objects +# +from pysnmp.entity import engine +from pysnmp.entity.rfc3413.oneliner import cmdgen + +# List of targets in the followin format: +# ( ( authData, transportTarget, varNames ), ... ) +targets = ( + # 1-st target (SNMPv1 over IPv4/UDP) + ( cmdgen.CommunityData('public', mpModel=0), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # 2-nd target (SNMPv2c over IPv4/UDP) + ( cmdgen.CommunityData('public'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # 3-nd target (SNMPv2c over IPv4/UDP) - same community and + # different transport address. + ( cmdgen.CommunityData('public'), + cmdgen.UdpTransportTarget(('localhost', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysContact', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysName', 0))) ), + # 4-nd target (SNMPv3 over IPv4/UDP) + ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # 5-th target (SNMPv3 over IPv6/UDP) + ( cmdgen.UsmUserData('usr-md5-none', 'authkey1'), + cmdgen.Udp6TransportTarget(('::1', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), + # N-th target + # ... +) + +# Wait for responses or errors +def cbFun(snmpEngine, sendRequestHandle, errorIndication, + errorStatus, errorIndex, varBinds, cbCtx): + (authData, transportTarget) = cbCtx + print('%s via %s' % (authData, transportTarget)) + if errorIndication: + print(errorIndication) + return 1 + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + return 1 + + for oid, val in varBinds: + if val is None: + print(oid.prettyPrint()) + else: + print('%s = %s' % (oid.prettyPrint(), val.prettyPrint())) + +snmpEngine = engine.SnmpEngine() + +cmdGen = cmdgen.AsyncCommandGenerator() + +# Submit GET requests +for authData, transportTarget, varNames in targets: + cmdGen.getCmd( + snmpEngine, authData, transportTarget, cmdgen.ContextData(), varNames, + # User-space callback function and its context + (cbFun, (authData, transportTarget)), + lookupNames=True, lookupValues=True + ) + +snmpEngine.transportDispatcher.runDispatcher() diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py new file mode 100644 index 0000000..aca4129 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py @@ -0,0 +1,35 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv1, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for two instances of SNMPv2-MIB::sysDescr.0 MIB object, +# * one in label and another in MIB symbol form +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py new file mode 100644 index 0000000..d9fc491 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py @@ -0,0 +1,49 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for an OID in string form +# * use custom timeout and request retries values +# +# Transport timing settings (maximum number of request retries and +# individual request timeout in seconds) can be set on a per-target basis +# as explained by the code that follows. +# +# Keep in mind that while timeout value can be specified in fractions of a +# second, default pysnmp timer resolution is quite low (about 0.5 sec) +# so there's no much point in using timeouts which is not a multiple of 0.5 +# Internal timer can be programmatically adjusted to finer resolution if needed. +# +# If retries value is set to 0, pysnmp will issue a single request. Even +# if no response arrives, there will be no retry. Likewise, retries=1 +# means one initial request plus one retry. +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget( + ('demo.snmplabs.com', 161), timeout=1.5, retries=0 + ), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py new file mode 100644 index 0000000..fba316f --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py @@ -0,0 +1,37 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for IF-MIB::ifInOctets.1 MIB object +# * Pass attached MIB compiler non-default ASN.1 MIB source +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType( + ObjectIdentity('IF-MIB', 'ifInOctets', 1).addAsn1MibSource('file:///usr/share/snmp', 'http://mibs.snmplabs.com/asn1/@mib@') + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py new file mode 100644 index 0000000..5fe5bfe --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py @@ -0,0 +1,34 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for two OIDs in string form +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.6.0'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py new file mode 100644 index 0000000..716d1e9 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py @@ -0,0 +1,36 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols +# * over IPv6/UDP +# * to an Agent at [::1]:161 +# * for three OIDs: one passed as a ObjectIdentity object while others are +# * in string form +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + Udp6TransportTarget(('::1', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.2.0')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.3.0'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py new file mode 100644 index 0000000..e83e0d2 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py @@ -0,0 +1,35 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3, user 'usr-none-none', no authentication, no privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for IF-MIB::ifInOctets.1 MIB object +# * perform response OIDs and values resolution at MIB +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + UsmUserData('usr-none-none'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1)), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py new file mode 100644 index 0000000..7b78a35 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py @@ -0,0 +1,40 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3, user 'usr-sha-aes128', SHA auth, AES128 privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for TCP-MIB::tcpConnLocalAddress."0.0.0.0".22."0.0.0.0".0 MIB object +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1', + authProtocol=usmHMACSHAAuthProtocol, + privProtocol=usmAesCfb128Protocol ), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType( + ObjectIdentity('TCP-MIB', + 'tcpConnLocalAddress', + '0.0.0.0', 22, + '0.0.0.0', 0) + )): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py new file mode 100644 index 0000000..4770c5d --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py @@ -0,0 +1,40 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3, user 'usr-md5-none', securityName 'myuser' +# MD5 authentication, no privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for an OID in text form +# +# The securityName parameter can be thought as an alias to userName and +# allows you to address a USM Table row just as userName does. However +# securityName can be made human-readable, also it is not an index in +# usmUserTable, thus duplicate securityName parameters are possible. +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1', + securityName='myuser'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py new file mode 100644 index 0000000..c01a61d --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py @@ -0,0 +1,33 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for IF-MIB::ifInOctets.1 MIB object +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in getCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py new file mode 100644 index 0000000..ca1d1fe --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py @@ -0,0 +1,81 @@ +# +# Command Generator +# +# Send SNMP GET request using the following scenario and options: +# +# * try to communicate with a SNMPv3 Engine using: +# ** a non-existing user +# ** over IPv4/UDP +# ** to an Agent at demo.snmplabs.com:161 +# * if remote SNMP Engine ID is discovered, send SNMP GET request: +# ** with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy +# at discovered securityEngineId +# ** to the same SNMP Engine ID +# ** for an OID in text form +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +snmpEngine = SnmpEngine() + +transportTarget = UdpTransportTarget(('demo.snmplabs.com', 161)) + +# +# To discover remote SNMP EngineID we will tap on SNMP engine inner workings +# by setting up execution point observer setup on INTERNAL class PDU processing +# + +observerContext = {} + +# Register a callback to be invoked at specified execution point of +# SNMP Engine and passed local variables at execution point's local scope +snmpEngine.observer.registerObserver( + lambda e,p,v,c: c.update(securityEngineId=v['securityEngineId']), + 'rfc3412.prepareDataElements:internal', + cbCtx=observerContext +) + +# Send probe SNMP request with invalid credentials + +authData = UsmUserData('non-existing-user') + +for errorIndication, errorStatus, errorIndex, \ + varBinds in getCmd(snmpEngine, authData, + transportTarget, ContextData()): + break + +# See if our SNMP engine received REPORT PDU containing securityEngineId + +if 'securityEngineId' not in observerContext: + print("Can't discover peer EngineID, errorIndication: %s" % errorIndication) + raise Exception() + +securityEngineId = observerContext.pop('securityEngineId') + +print('Remote securityEngineId = %s' % securityEngineId.prettyPrint()) + +# +# Query remote SNMP Engine using usmUserTable entry configured for it +# + +authData = UsmUserData('usr-md5-none', 'authkey1', + securityEngineId=securityEngineId) + +for errorIndication, errorStatus, errorIndex, \ + varBinds in getCmd(snmpEngine, authData, \ + transportTarget, ContextData(), \ + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))): + break + +# Check for errors and print out results +if errorIndication: + print(errorIndication) +else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for name, val in varBinds: + print('%s = %s' % (name.prettyPrint(), val.prettyPrint())) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py new file mode 100644 index 0000000..0bed0f0 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py @@ -0,0 +1,38 @@ +# +# Command Generator +# +# Send SNMP GETBULK request using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * with values non-repeaters = 0, max-repetitions = 25 +# * for two OIDs in string form +# * stop when response OIDs leave the scopes of initial OIDs +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in bulkCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + 0, 25, + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.3'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py new file mode 100644 index 0000000..9b255f3 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py @@ -0,0 +1,40 @@ +# +# Command Generator +# +# Send SNMP GETBULK request using the following options: +# +# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols +# * over IPv6/UDP +# * to an Agent at [::1]:161 +# * with values non-repeaters = 1, max-repetitions = 25 +# * for IP-MIB::ipAdEntAddr and all columns of the IF-MIB::ifEntry table +# * stop when response OIDs leave the scopes of the table OR maxRows == 20 +# * perform response OIDs and values resolution at MIB +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in bulkCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + Udp6TransportTarget(('::1', 161)), + ContextData(), + 1, 25, + ObjectType(ObjectIdentity('IP-MIB', 'ipAdEntAddr')), + ObjectType(ObjectIdentity('IP-MIB', 'ipAddrEntry')), + lookupNames=True, lookupValues=True, maxRows=20): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py new file mode 100644 index 0000000..5d1eb81 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py @@ -0,0 +1,42 @@ +# +# Command Generator +# +# Send SNMP GETBULK request using the following options: +# +# * with SNMPv3, user 'usr-none-none', no authentication, no privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for all OIDs past TCP-MIB::tcpConnTable +# * TCP-MIB will be searched by a user-specified path +# * run till end-of-mib condition is reported by Agent OR maxRows == 100 OR +# maxCalls == 10 request-response interactions occur +# * ignoring non-increasing OIDs whenever reported by Agent +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in bulkCmd(SnmpEngine(), + UsmUserData('usr-none-none'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + 0, 50, + ObjectType(ObjectIdentity('TCP-MIB', 'tcpConnTable').addMibSource('/tmp/mibs')), + maxRows=100, maxCalls=10, + lexicographicMode=True, + ignoreNonIncreasingOid=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py new file mode 100644 index 0000000..b1304ae --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py @@ -0,0 +1,90 @@ +## +# Asynchronous Command Generator +# +# Send a bunch of SNMP GETNEXT requests all at once using the following options: +# +# * with SNMPv1, community 'public' and +# with SNMPv2c, community 'public' and +# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy +# * over IPv4/UDP and +# over IPv6/UDP +# * to an Agent at demo.snmplabs.com:161 and +# to an Agent at [::1]:161 +# * for multiple MIB subtrees and tables +# +from pysnmp.entity import engine +from pysnmp.entity.rfc3413.oneliner import cmdgen + +# List of targets in the followin format: +# ( ( authData, transportTarget, varNames ), ... ) +targets = ( + # 1-st target (SNMPv1 over IPv4/UDP) + ( cmdgen.CommunityData('public', mpModel=0), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.2.1')), + cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.3.1')) ) ), + # 2-nd target (SNMPv2c over IPv4/UDP) + ( cmdgen.CommunityData('public'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.4.1')), ) ), + # 3-nd target (SNMPv3 over IPv4/UDP) + ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'system')), ) ), + # 4-th target (SNMPv3 over IPv6/UDP) + ( cmdgen.UsmUserData('usr-md5-none', 'authkey1'), + cmdgen.Udp6TransportTarget(('::1', 161)), + ( cmdgen.ObjectType(cmdgen.ObjectIdentity('IF-MIB', 'ifTable')), ) ) + # N-th target + # ... +) + +# Wait for responses or errors, submit GETNEXT requests for further OIDs +def cbFun(snmpEngine, sendRequestHandle, errorIndication, + errorStatus, errorIndex, varBindTable, cbCtx): + (varBindHead, authData, transportTarget) = cbCtx + print('%s via %s' % (authData, transportTarget)) + if errorIndication: + print(errorIndication) + return + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBindTable[-1][int(errorIndex)-1][0] or '?' + ) + ) + return + varBindTableRow = varBindTable[-1] + for idx in range(len(varBindTableRow)): + name, val = varBindTableRow[idx] + if val is not None and varBindHead[idx] <= name: + # still in table + break + else: + print('went out of table at %s' % (name, )) + return + + for varBindRow in varBindTable: + for oid, val in varBindRow: + if val is None: + print(oid.prettyPrint()) + else: + print('%s = %s' % (oid.prettyPrint(), val.prettyPrint())) + + return True # continue table retrieval + +snmpEngine = engine.SnmpEngine() + +cmdGen = cmdgen.AsyncCommandGenerator() + +# Submit initial GETNEXT requests and wait for responses +for authData, transportTarget, varBinds in targets: + varBindHead = [ x[0] for x in cmdGen.makeVarBinds(snmpEngine, varBinds ) ] + cmdGen.nextCmd( + snmpEngine, authData, transportTarget, cmdgen.ContextData(), varBinds, + # User-space callback function and its context + (cbFun, (varBindHead, authData, transportTarget)), + lookupNames=True, lookupValues=True + ) + +snmpEngine.transportDispatcher.runDispatcher() diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py new file mode 100644 index 0000000..0367eb3 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py @@ -0,0 +1,52 @@ +# +# GETNEXT Command Generator Application +# +# Perform SNMP GETNEXT operation with the following options: +# +# with SNMPv1, community 'public' +# over IPv4/UDP +# to an Agent at demo.snmplabs.com:161 +# two columns of the IF-MIB::ifEntry table +# search for pysnmp MIBs at specific filesystem location +# stop when response OIDs leave the scopes of initial OIDs +# +# This script performs similar to the following Net-SNMP command: +# +# $ snmpwalk -v1 -c public -ObentU demo.snmplabs.com IF-MIB::ifDescr IF-MIB::ifType +# +# The pysnmp engine maintains its own MIB search path which is usued +# for searching for MIB modules by name. By default pysnmp will search +# its built-in MIB repository (pysnmp.smi.mibs.instances, pysnmp.smi.mibs) +# and pysnmp-mibs package (pysnmp_mibs). Those two entries are normally +# found relative to $PYTHONPATH. Besides searching MIBs as Python modules +# (which could be egg-ed or not), pysnmp can also search for stand-alone +# files in given directories. The latter could be specified by +# ObjectIdentity.addMibSource() calls. New search path entries are added +# in front of existing ones in search path. +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifDescr').addMibSource('/tmp/mymibs')), + ObjectType(ObjectIdentity('IF-MIB', 'ifType')), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py new file mode 100644 index 0000000..1c3a998 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py @@ -0,0 +1,40 @@ +# +# Command Generator +# +# Send SNMP GETNEXT request using the following options: +# +# * with SNMPv1, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for some columns of the IF-MIB::ifEntry table +# * stop when response OIDs leave the scopes of initial OIDs +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifDescr')), + ObjectType(ObjectIdentity('IF-MIB', 'ifType')), + ObjectType(ObjectIdentity('IF-MIB', 'ifMtu')), + ObjectType(ObjectIdentity('IF-MIB', 'ifSpeed')), + ObjectType(ObjectIdentity('IF-MIB', 'ifPhysAddress')), + ObjectType(ObjectIdentity('IF-MIB', 'ifType'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py new file mode 100644 index 0000000..e59b07b --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py @@ -0,0 +1,47 @@ +# +# GETNEXT Command Generator Application +# +# Perform SNMP GETNEXT operation with the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for an OID in string form +# * resolve response OIDs and values into human-freidly form +# * stop when response OIDs leave the scopes of initial OIDs +# +# This script performs similar to the following Net-SNMP command: +# +# $ snmpwalk -v2c -c public demo.snmplabs.com 1.3.6.1.2.1.1 +# +# The lookupNames and lookupValues keyword arguments will make pysnmp +# trying to resolve OIDs and values, in response variable-bindings, +# into human-friendly form. If response OIDs do not belong to any of +# currently loaded MIBs, unresolved OIDs and values will still be +# returned. +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1')), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py new file mode 100644 index 0000000..178cd09 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py @@ -0,0 +1,36 @@ +# +# Command Generator +# +# Send SNMP GETNEXT requests using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for two OIDs in string form +# * stop when response OIDs leave the scopes of initial OIDs +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.2')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.3'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py new file mode 100644 index 0000000..01c74be --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py @@ -0,0 +1,39 @@ +# +# Command Generator +# +# Send SNMP GETNEXT requests using the following options: +# +# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols +# * over IPv6/UDP +# * to an Agent at [::1]:161 +# * for all columns of the IF-MIB::ifEntry table +# * stop when response OIDs leave the scopes of the table +# * perform response OIDs and values resolution at MIB +# +# make sure IF-MIB.py is in search path +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + Udp6TransportTarget(('::1', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifEntry')), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py new file mode 100644 index 0000000..658f6fc --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py @@ -0,0 +1,37 @@ +# +# Command Generator +# +# Send SNMP GETNEXT requests using the following options: +# +# * with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * for all OIDs in IF-MIB +# * stop when response OIDs leave the scopes of the table +# * perform response values resolution at MIB +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', '')), + lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py new file mode 100644 index 0000000..99eba51 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py @@ -0,0 +1,42 @@ +# +# Command Generator +# +# Send SNMP GETNEXT requests using the following options: +# +# * with SNMPv3, user 'usr-sha-aes128', SHA auth, AES128 privacy +# * over Local Domain Sockets +# * to an Agent at demo.snmplabs.com:161 +# * for all OIDs past IF-MIB (load up all MIBs in search path) +# * run till end-of-mib condition is reported by Agent OR maxRows == 100 +# * ignoring non-increasing OIDs whenever reported by Agent +# +# make sure IF-MIB.py is search path +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in nextCmd(SnmpEngine(), + UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1', + authProtocol=usmHMACSHAAuthProtocol, + privProtocol=usmAesCfb128Protocol), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', '').loadMibs()), + lexicographicMode=True, maxRows=100, + ignoreNonIncreasingOid=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + break + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + break + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py new file mode 100644 index 0000000..bba4a84 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py @@ -0,0 +1,36 @@ +# +# Command Generator +# +# Send SNMP SET request using the following options: +# +# * with SNMPv1, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * setting two OIDs to new values (types explicitly specified) +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in setCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ('1.3.6.1.2.1.1.9.1.2.1', rfc1902.ObjectName('1.3.6.1.4.1.20408.1.1')), + ('1.3.6.1.2.1.1.9.1.2.1', '1.3.6.1.4.1.20408.1.1'), + ('1.3.6.1.2.1.1.9.1.3.1', rfc1902.OctetString('new system name'))): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py new file mode 100644 index 0000000..46d4c5e --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py @@ -0,0 +1,36 @@ +# +# Command Generator +# +# Send SNMP SET request using the following options: +# +# * with SNMPv2c, community 'public' +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB) +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in setCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType( + ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name' + )): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py new file mode 100644 index 0000000..9c5b20f --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py @@ -0,0 +1,43 @@ +# +# Command Generator +# +# Send SNMP SET request using the following options: +# +# * with SNMPv3 with user 'usr-md5-none', MD5 auth and no privacy protocols +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * addressing particular set of Managed Objects at remote SNMP Engine by: +# * contextEngineId 0x80004fb805636c6f75644dab22cc and +# * contextName 'a172334d7d97871b72241397f713fa12' +# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB) +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in setCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData( + contextEngineId=rfc1902.OctetString(hexValue='80004fb805636c6f75644dab22cc'), + contextName='da761cfc8c94d3aceef4f60f049105ba' + ), + ObjectType( + ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name' + )): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py new file mode 100644 index 0000000..22d165d --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py @@ -0,0 +1,44 @@ +# +# Command Generator +# +# Send SNMP GET request using the following options: +# +# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols +# * use remote SNMP Engine ID 0x80004fb805636c6f75644dab22cc (USM +# autodiscovery will run) +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB) +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in setCmd(SnmpEngine(), + UsmUserData( + 'usr-md5-des', 'authkey1', 'privkey1', + securityEngineId=rfc1902.OctetString( + hexValue='80004fb805636c6f75644dab22cc' + ) + ), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType( + ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name' + )): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py new file mode 100644 index 0000000..379a0dd --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py @@ -0,0 +1,39 @@ +# +# Command Generator +# +# Send SNMP SET request using the following options: +# +# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols +# * over IPv4/UDP +# * to an Agent at demo.snmplabs.com:161 +# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB) +# * perform response OIDs and values resolution at MIB +# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * +from pysnmp.proto import rfc1902 + +for errorIndication, \ + errorStatus, errorIndex, \ + varBinds in setCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType( + ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name' + ), + lookupNames=True, lookupValues=True): + # Check for errors and print out results + if errorIndication: + print(errorIndication) + else: + if errorStatus: + print('%s at %s' % ( + errorStatus.prettyPrint(), + errorIndex and varBinds[int(errorIndex)-1][0] or '?' + ) + ) + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) + break |