diff options
Diffstat (limited to 'examples/v3arch/asyncore/oneliner')
46 files changed, 2040 insertions, 1273 deletions
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-snmp-engines.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-snmp-engines.py new file mode 100644 index 0000000..aa96c1e --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-snmp-engines.py @@ -0,0 +1,88 @@ +""" +Multiple SNMP Engines ++++++++++++++++++++++ + +Send SNMP notifications in behalf of multiple independend SNMP engines +using the following options: + +* with a single transport dispatcher and two independent SNMP engines +* SNMPv2c and SNMPv3 +* with community name 'public' or USM username usr-md5-des +* over IPv4/UDP +* send IMFORM notification +* to multiple Managers +* with TRAP ID 'coldStart' specified as a MIB symbol +* include managed object information specified as var-bind objects pair + +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 AsyncNotificationOriginator with each of +SNMP Engines to communicate INFORM notification 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.ntforg import * +from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher + +# List of targets in the following format: +# ( ( authData, transportTarget ), ... ) +targets = ( + # 1-st target (SNMPv2c over IPv4/UDP) + ( CommunityData('public'), + UdpTransportTarget(('localhost', 1162)), + ContextData() ), + # 2-nd target (SNMPv3 over IPv4/UDP) + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('localhost', 162)), + ContextData() ), +) + +def cbFun(snmpEngine, sendRequestHandle, errorIndication, + errorStatus, errorIndex, varBinds, cbCtx): + snmpEngine = cbCtx + if errorIndication: + print('Notification %s for %s not sent: %s' % (sendRequestHandle, snmpEngine.snmpEngineID.prettyPrint(), errorIndication)) + elif errorStatus: + print('Notification Receiver returned error for request %s, SNMP Engine %s: %s @%s' % (sendRequestHandle, snmpEngine.snmpEngineID.prettyPrint(), errorStatus, errorIndex)) + else: + print('Notification %s for SNMP Engine %s delivered:' % (sendRequestHandle, snmpEngine.snmpEngineID.prettyPrint())) + for name, val in varBinds: + print('%s = %s' % (name.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') + +ntfOrg = AsyncNotificationOriginator() + +for authData, transportTarget, contextData in targets: + snmpEngine = transportTarget.getTransportInfo()[1][1] % 3 and \ + snmpEngineA or snmpEngineB + sendPduHandle = ntfOrg.sendNotification( + snmpEngine, + authData, + transportTarget, + contextData, + 'inform', # NotifyType + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + ).addVarBinds( ( '1.3.6.1.2.1.1.1.0', 'my name' ) ), + cbInfo=(cbFun, snmpEngine) + ) + +transportDispatcher.runDispatcher() diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-transports-and-protocols.py new file mode 100644 index 0000000..cfa2acd --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-async-multiple-transports-and-protocols.py @@ -0,0 +1,60 @@ +""" +Multiple concurrent notifications ++++++++++++++++++++++++++++++++++ + +Send multiple SNMP notifications at once using the following options: + +* SNMPv2c and SNMPv3 +* with community name 'public' or USM username usr-md5-des +* over IPv4/UDP +* send INFORM notification +* to multiple Managers +* with TRAP ID 'coldStart' specified as a MIB symbol +* include managed object information specified as var-bind objects pair + +"""# +from pysnmp.entity.rfc3413.oneliner.ntforg import * + +# List of targets in the followin format: +# ( ( authData, transportTarget ), ... ) +targets = ( + # 1-st target (SNMPv2c over IPv4/UDP) + ( CommunityData('public'), + UdpTransportTarget(('localhost', 162)), + ContextData() ), + # 2-nd target (SNMPv3 over IPv4/UDP) + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('localhost', 162)), + ContextData() ), +) + +def cbFun(snmpEngine, sendRequestHandle, errorIndication, + errorStatus, errorIndex, varBinds, cbctx): + if errorIndication: + print('Notification %s not sent: %s' % (sendRequestHandle, errorIndication)) + elif errorStatus: + print('Notification Receiver returned error for %s: %s @%s' % + (sendRequestHandle, errorStatus, errorIndex)) + else: + print('Notification %s delivered:' % sendRequestHandle) + for name, val in varBinds: + print('%s = %s' % (name.prettyPrint(), val.prettyPrint())) + +snmpEngine = engine.SnmpEngine() + +ntfOrg = AsyncNotificationOriginator() + +for authData, transportTarget, contextData in targets: + sendPduHandle = ntfOrg.sendNotification( + snmpEngine, + authData, + transportTarget, + contextData, + 'inform', # NotifyType + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + ).addVarBinds( ( '1.3.6.1.2.1.1.1.0', 'my name' ) ), + cbInfo=(cbFun, None) + ) + +snmpEngine.transportDispatcher.runDispatcher() 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 index 6eebaa9..4d99a12 100644 --- 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 @@ -1,42 +1,49 @@ -# -# 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 -# +""" +Sending additional var-binds +++++++++++++++++++++++++++++ + +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 + +Functionally similar to: + +| $ snmpinform -v2c -c public +| demo.snmplabs.com \ +| 12345 \ +| 1.3.6.1.6.3.1.1.5.1 \ +| 1.3.6.1.2.1.1.1.0 s 'my system' + +"""# 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 ])) +errorIndication, errorStatus, errorIndex, varBinds = next( + sendNotification(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('localhost', 162)), + ContextData(), + 'inform', + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + ).addVarBinds( + ObjectType(ObjectIdentity('SNMPv2-MIB','sysName',0), + 'my system') + )) +) + +if errorIndication: + print(errorIndication) +elif 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 index 071c336..0c75c21 100644 --- 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 @@ -1,46 +1,50 @@ -# -# 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. -# +""" +INFORM with custom ContextEngineId +++++++++++++++++++++++++++++++++++ + +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. + +Functionally similar to: + +| $ snmpinform -v3 -l authNoPriv -u usr-md5-none -A authkey1 \ +| -E 0x8000000004030201 +| demo.snmplabs.com \ +| 12345 \ +| 1.3.6.1.6.3.1.1.5.2 + +"""# 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 ])) + +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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 index ed87fdc..1b3e1f8 100644 --- 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 @@ -1,43 +1,48 @@ -# -# 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. -# +""" +INFORM with custom ContextName +++++++++++++++++++++++++++++++ + +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. + +Functionally similar to: + +| $ snmpinform -v3 -l authNoPriv -u usr-md5-none -A authkey1 \ +| -n my-context \ +| demo.snmplabs.com \ +| 12345 \ +| 1.3.6.1.6.3.1.1.5.2 + +"""# 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 ])) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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 index 2996eb5..161427a 100644 --- a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py @@ -1,44 +1,49 @@ -## -# 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' -# +""" +INFORM, auth: MD5 privacy: DES +++++++++++++++++++++++++++++++ + +Send SNMP INFORM 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' + +Functionally similar to: + +| $ snmpinform -v3 -l authPriv -u usr-sha-aes -A authkey1 -X privkey1 \ +| demo.snmplabs.com \ +| 12345 \ +| 1.3.6.1.4.1.20408.4.1.1.2 \ +| '1.3.6.1.2.1.1.1.0' s 'my system' + +"""# 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 ])) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'), + 'system name') + )) +) +if errorIndication: + print(errorIndication) +elif 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-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-async-multiple-transports-and-protocols.py new file mode 100644 index 0000000..60c557a --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-async-multiple-transports-and-protocols.py @@ -0,0 +1,51 @@ +""" +Multiple concurrent queries ++++++++++++++++++++++++++++ + +Send a bunch of different SNMP Notifications to different peers all at once, +wait for responses asynchronously: + +* SNMPv1 and SNMPv2c +* with community name 'public' +* over IPv4/UDP +* send TRAP notification +* to multiple Managers +* with TRAP ID 'coldStart' specified as a MIB symbol +* include managed object information specified as var-bind objects pair + +"""# +from pysnmp.entity.rfc3413.oneliner.ntforg import * + +# List of targets in the followin format: +# ( ( authData, transportTarget ), ... ) +targets = ( + # 1-st target (SNMPv1 over IPv4/UDP) + ( CommunityData('public', mpModel=0), + UdpTransportTarget(('localhost', 162)), + ContextData() ), + # 2-nd target (SNMPv2c over IPv4/UDP) + ( CommunityData('public'), + UdpTransportTarget(('localhost', 162)), + ContextData() ), +) + +snmpEngine = SnmpEngine() + +ntfOrg = AsyncNotificationOriginator() + +for authData, transportTarget, contextData in targets: + ntfOrg.sendNotification( + snmpEngine, + authData, + transportTarget, + contextData, + 'trap', # NotifyType + NotificationType( + ObjectIdentity('SNMPv2-MIB', 'coldStart') + ).addVarBinds( + ( ObjectName('1.3.6.1.2.1.1.1.0'), + OctetString('my name') ) + ) + ) + +snmpEngine.transportDispatcher.runDispatcher() 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 index 1d2b825..85e130b 100644 --- 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 @@ -1,36 +1,50 @@ -# -# 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' -# +""" +Custom SNMPv1 TRAP +++++++++++++++++++ + +Send SNMPv1 TRAP through unified SNMPv3 message processing framework. + +Original v1 TRAP fields are mapped into dedicated variable-bindings, +(see `RFC2576 <https://www.ietf.org/rfc/rfc2576.txt>`_) for details. + +* 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' + +Functionally similar to: + +| $ snmptrap -v1 -c public demo.snmplabs.com \ +| 1.3.6.1.4.1.20408.4.1.1.2 \ +| 127.0.0.1 \ +| 6 \ +| 432 \ +| 12345 \ +| '1.3.6.1.2.1.1.1.0' s '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) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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 index a704146..5ad5ef7 100644 --- 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 @@ -1,34 +1,46 @@ -# -# 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' -# +""" +SNMPv1 TRAP with defaults ++++++++++++++++++++++++++ + +Send SNMPv1 TRAP through unified SNMPv3 message processing framework +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' + +Functionally similar to: + +| $ snmptrap -v1 -c public demo.snmplabs.com \ +| 1.3.6.1.4.1.20408.4.1.1.2 \ +| 0.0.0.0 \ +| 1 \ +| 0 \ +| 0 +| '1.3.6.1.2.1.1.1.0' s '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) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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 index b243f43..a328a2c 100644 --- 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 @@ -1,27 +1,37 @@ -# -# 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 -# +""" +SNMPv2c TRAP via NOTIFICATION-TYPE +++++++++++++++++++++++++++++++++++ + +Initialize TRAP message contents from variables specified +in *NOTIFICATION-TYPE* SMI macro. + +* 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 + +Functionally similar to: + +| $ snmptrap -v2c -c public demo.snmplabs.com \ +| 12345 +| 1.3.6.1.4.1.20408.4.1.1.2 \ + +"""# 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) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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 index bbc9194..f03948f 100644 --- 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 @@ -1,37 +1,45 @@ -# -# 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. -# +""" +SNMPv3 TRAP: auth SHA, privacy: AES128 +++++++++++++++++++++++++++++++++++++++ + +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. + +Functionally similar to: + +| $ snmptrap -v3 -l authPriv -u usr-sha-aes -A authkey1 -X privkey1 \ +| -a SHA -x AES \ +| demo.snmplabs.com \ +| 12345 \ +| 1.3.6.1.4.1.20408.4.1.1.2 \ +| '1.3.6.1.2.1.1.1.0' s 'my system' + +"""# 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) +errorIndication, errorStatus, errorIndex, varBinds = next( + 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 index 85586ed..96ce80f 100644 --- 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 @@ -1,52 +1,54 @@ -# -# 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 +""" +Multiple SNMP engines ++++++++++++++++++++++ + +Send multiple SNMP GET requests to multiple peers using multiple +independend SNMP engines. Deal with peers asynchronously. SNMP options +are: + +* 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.cmdgen import * from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher -# List of targets in the followin format: +# List of targets in the following 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) )) ), + ( CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(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) )) ), + ( CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 1161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(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) )) ) + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 2161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ) # N-th target # ... ) @@ -61,20 +63,16 @@ def cbFun(snmpEngine, sendRequestHandle, errorIndication, if errorIndication: print(errorIndication) return 1 - if errorStatus: + elif 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())) - + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) # Instantiate the single transport dispatcher object transportDispatcher = AsyncoreDispatcher() @@ -84,19 +82,19 @@ transportDispatcher.registerRoutingCbFun( lambda td,ta,d: ta[1] % 3 and 'A' or 'B' ) -snmpEngineA = engine.SnmpEngine() +snmpEngineA = SnmpEngine() snmpEngineA.registerTransportDispatcher(transportDispatcher, 'A') -snmpEngineB = engine.SnmpEngine() +snmpEngineB = SnmpEngine() snmpEngineB.registerTransportDispatcher(transportDispatcher, 'B') -cmdGen = cmdgen.AsyncCommandGenerator() +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, + snmpEngine, authData, transportTarget, ContextData(), varBinds, (cbFun, (snmpEngine, authData, transportTarget)) ) 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 index e50a54b..f5c0569 100644 --- 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 @@ -1,50 +1,52 @@ -# -# 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 +""" +Multiple concurrent queries ++++++++++++++++++++++++++++ + +Send a bunch of different SNMP GET requests to different peers all at once, +wait for responses asynchronously: + +* 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.rfc3413.oneliner.cmdgen import * # 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) )) ), + ( CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(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) )) ), + ( CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(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))) ), + ( CommunityData('public'), + UdpTransportTarget(('localhost', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysContact', 0)), + ObjectType(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) )) ), + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(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) )) ), + ( UsmUserData('usr-md5-none', 'authkey1'), + Udp6TransportTarget(('::1', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ), # N-th target # ... ) @@ -52,36 +54,32 @@ targets = ( # Wait for responses or errors def cbFun(snmpEngine, sendRequestHandle, errorIndication, errorStatus, errorIndex, varBinds, cbCtx): - (authData, transportTarget) = cbCtx + authData, transportTarget = cbCtx print('%s via %s' % (authData, transportTarget)) if errorIndication: print(errorIndication) - return 1 - if errorStatus: + return True + elif errorStatus: print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' + 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())) + return True + else: + for varBind in varBinds: + print(' = '.join([ x.prettyPrint() for x in varBind ])) -snmpEngine = engine.SnmpEngine() +snmpEngine = SnmpEngine() -cmdGen = cmdgen.AsyncCommandGenerator() +cmdGen = AsyncCommandGenerator() # Submit GET requests for authData, transportTarget, varNames in targets: cmdGen.getCmd( - snmpEngine, authData, transportTarget, cmdgen.ContextData(), varNames, + snmpEngine, authData, transportTarget, ContextData(), varNames, # User-space callback function and its context - (cbFun, (authData, transportTarget)), - lookupNames=True, lookupValues=True + (cbFun, (authData, transportTarget)) ) snmpEngine.transportDispatcher.runDispatcher() diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-threaded-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-threaded-multiple-transports-and-protocols.py new file mode 100644 index 0000000..47630ea --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-threaded-multiple-transports-and-protocols.py @@ -0,0 +1,127 @@ +""" +Query Agents from multiple threads +++++++++++++++++++++++++++++++++++ + +Send a bunch of SNMP GET requests simultaneously using the following options: + +* process 5 GET requests in 3 parallel threads +* 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 sys import version_info +if version_info[0] == 2: + from Queue import Queue +else: + from queue import Queue +from threading import Thread +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +# List of targets in the followin format: +# ( ( authData, transportTarget, varNames ), ... ) +targets = ( + # 1-st target (SNMPv1 over IPv4/UDP) + ( CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0)))), + # 2-nd target (SNMPv2c over IPv4/UDP) + ( CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0)))), + # 3-nd target (SNMPv2c over IPv4/UDP) - same community and + # different transport address. + ( CommunityData('public'), + UdpTransportTarget(('localhost', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysContact', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysName', 0)))), + # 4-nd target (SNMPv3 over IPv4/UDP) + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) ))), + # 5-th target (SNMPv3 over IPv6/UDP) + ( UsmUserData('usr-md5-none', 'authkey1'), + Udp6TransportTarget(('::1', 161)), + ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0)))), + # N-th target + # ... +) + +class Worker(Thread): + def __init__(self, requests, responses): + Thread.__init__(self) + self.snmpEngine = SnmpEngine() + self.requests = requests + self.responses = responses + self.setDaemon(True) + self.start() + + def run(self): + while True: + authData, transportTarget, varBinds = self.requests.get() + self.responses.append( + next( + getCmd( + self.snmpEngine, + authData, transportTarget, ContextData(), *varBinds + ) + ) + ) + if hasattr(self.requests, 'task_done'): # 2.5+ + self.requests.task_done() + +class ThreadPool: + def __init__(self, num_threads): + self.requests = Queue(num_threads) + self.responses = [] + for _ in range(num_threads): + Worker(self.requests, self.responses) + + def addRequest(self, authData, transportTarget, varBinds): + self.requests.put((authData, transportTarget, varBinds)) + + def getResponses(self): return self.responses + + def waitCompletion(self): + if hasattr(self.requests, 'join'): + self.requests.join() # 2.5+ + else: + from time import sleep + # this is a lame substitute for missing .join() + # adding an explicit synchronization might be a better solution + while not self.requests.empty(): + sleep(1) + +pool = ThreadPool(3) + +# Submit GET requests +for authData, transportTarget, varBinds in targets: + pool.addRequest(authData, transportTarget, varBinds) + +# Wait for responses or errors +pool.waitCompletion() + +# Walk through responses +for errorIndication, errorStatus, errorIndex, varBinds in pool.getResponses(): + print('Response for %s from %s:' % (authData, transportTarget)) + if errorIndication: + print(errorIndication) + elif 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/manager/cmdgen/get-v1.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py index aca4129..2e9c298 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py @@ -1,35 +1,37 @@ -# -# 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 -# +""" +SNMPv1 +++++++ + +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, + +Functionally similar to: + +| $ snmpget -v1 -c public demo.snmplabs.com SNMPv2-MIB::sysDescr.0 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0))) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index d9fc491..1439aeb 100644 --- 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 @@ -1,49 +1,53 @@ -# -# 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. -# +""" +Custom request timeout +++++++++++++++++++++++ + +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. + +Functionally similar to: + +| $ snmpget -v1 -c public demo.snmplabs.com 1.3.6.1.2.1.1.1.0 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index fba316f..e994dba 100644 --- 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 @@ -1,37 +1,39 @@ -# -# 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 -# +""" +Custom ASN.1 MIB path ++++++++++++++++++++++ + +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 non-default ASN.1 MIB source to MIB compiler + +Functionally similar to: + +| $ snmpget -v2c -c public -M /usr/share/snmp demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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@'))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py index 5fe5bfe..dcc4ccf 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py @@ -1,34 +1,39 @@ -# -# 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 -# +""" +SNMPv2c ++++++++ + +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 + +Functionally similar to: + +| $ snmpget -v2c -c public demo.snmplabs.com 1.3.6.1.2.1.1.1.0 \ +| 1.3.6.1.2.1.1.6.0 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-auth-priv-sha-aes128.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-auth-priv-sha-aes128.py new file mode 100644 index 0000000..85a0399 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-auth-priv-sha-aes128.py @@ -0,0 +1,57 @@ +""" +SNMPv3: auth SHA, privacy AES128 +++++++++++++++++++++++++++++++++ + +Send SNMP GET request using the following options: + +* with SNMPv3, user 'usr-sha-aes', SHA authentication, AES128 encryption +* over IPv4/UDP +* to an Agent at demo.snmplabs.com:161 +* for SNMPv2-MIB::sysDescr.0 MIB object + +Available authentication protocols: + +#. usmHMACMD5AuthProtocol +#. usmHMACSHAAuthProtocol +#. usmNoAuthProtocol + +Available privacy protocols: + +#. usmDESPrivProtocol +#. usm3DESEDEPrivProtocol +#. usmAesCfb128Protocol +#. usmAesCfb192Protocol +#. usmAesCfb256Protocol +#. usmNoPrivProtocol + +Functionally similar to: + +| $ snmpget -v3 -l authPriv -u usr-sha-aes -A authkey1 -X privkey1 \ +| -a SHA -x AES \ +| demo.snmplabs.com \ +| SNMPv2-MIB::sysDescr.0 + +"""# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + UsmUserData('usr-sha-aes', 'authkey1', 'privkey1', + authProtocol=usmHMACSHAAuthProtocol, + privProtocol=usmAesCfb128Protocol), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-auth-priv.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-auth-priv.py new file mode 100644 index 0000000..2c53b29 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-auth-priv.py @@ -0,0 +1,39 @@ +""" +SNMPv3: auth MD5, privacy DES ++++++++++++++++++++++++++++++ + +Send SNMP GET request using the following options: + +* with SNMPv3, user 'usr-md5-des', MD5 authentication, DES encryption +* over IPv4/UDP +* to an Agent at demo.snmplabs.com:161 +* for IF-MIB::ifInOctets.1 MIB object + +Functionally similar to: + +| $ snmpget -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 \ +| demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 + +"""# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-multiple-calls.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-multiple-calls.py new file mode 100644 index 0000000..116d99f --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-multiple-calls.py @@ -0,0 +1,47 @@ +""" +Sequence Of GET's ++++++++++++++++++ + +Send two SNMP GET requests in a row 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 and IF-MIB::ifOutOctets.1 MIB objects + +Use a queue of MIB objects to query. + +The next() call is used to forward Python iterator to the position where it +could consume input + +Functionally similar to: + +| $ snmpget -v3 -l authNoPriv -u usr-md5-none -A authkey1 demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 + +"""# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +queue = [ [ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1)) ], + [ ObjectType(ObjectIdentity('IF-MIB', 'ifOutOctets', 1)) ] ] + +iter = getCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData()) + +next(iter) + +while queue: + errorIndication, errorStatus, errorIndex, varBinds = iter.send(queue.pop()) + if errorIndication: + print(errorIndication) + elif 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/manager/cmdgen/get-v3-noauth-nopriv.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-noauth-nopriv.py new file mode 100644 index 0000000..976e0f4 --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-noauth-nopriv.py @@ -0,0 +1,39 @@ +""" +SNMPv3: no auth, no privacy ++++++++++++++++++++++++++++ + +Send SNMP GET request using the following options: + +* with SNMPv3, user 'usr-none-none', no authentication, no encryption +* over IPv4/UDP +* to an Agent at demo.snmplabs.com:161 +* for IF-MIB::ifInOctets.1 MIB object + +Functionally similar to: + +| $ snmpget -v3 -l noAuthNoPriv -u usr-none-none +| demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 + +"""# +from pysnmp.entity.rfc3413.oneliner.cmdgen import * + +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + UsmUserData('usr-none-none'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1))) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index 716d1e9..9499a38 100644 --- 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 @@ -1,36 +1,42 @@ -# -# 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 -# +""" +GET over IPv6 ++++++++++++++ + +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 in string form + +Functionally similar to: + +| $ snmpget -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 \ +| udp6:[::1]:161 \ +| 1.3.6.1.2.1.1.1.0 \ +| 1.3.6.1.2.1.1.2.0 \ +| 1.3.6.1.2.1.1.3.0 +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-table-object-by-index.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py index e83e0d2..84b9105 100644 --- 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 @@ -1,35 +1,41 @@ -# -# 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 -# +""" +GET table row ++++++++++++++ + +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 and IF-MIB::ifOutOctets.1 MIB object +* perform response OIDs and values resolution at MIB + +Functionally similar to: + +| $ snmpget -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 \ +| demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 IF-MIB::ifOutOctets.1 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + UsmUserData('usr-none-none'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1)), + ObjectType(ObjectIdentity('IF-MIB', 'ifOutOctets', 1))) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index 7b78a35..ebc2b0c 100644 --- 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 @@ -1,40 +1,46 @@ -# -# 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 -# +""" +Fetch table row by composite index +++++++++++++++++++++++++++++++++++ + +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 + +Functionally similar to: + +| $ snmpget -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 \ +| -a SHA -x AES \ +| demo.snmplabs.com \ +| TCP-MIB::tcpConnLocalAddress."0.0.0.0".22."0.0.0.0".0 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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)) + ) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-with-custom-security-name.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py index 4770c5d..ae5e656 100644 --- 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 @@ -1,40 +1,38 @@ -# -# 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. -# +""" +Custom SecurityName ++++++++++++++++++++ + +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 +errorIndication, errorStatus, errorIndex, varBinds = next( + 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'))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py index c01a61d..6e0d1c5 100644 --- 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 @@ -1,33 +1,38 @@ -# -# 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 -# +""" +SNMPv3: auth MD5, no privacy +++++++++++++++++++++++++++++ + +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 + +Functionally similar to: + +| $ snmpget -v3 -l authNoPriv -u usr-md5-none -A authkey1 demo.snmplabs.com \ +| IF-MIB::ifInOctets.1 + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1))) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/get-v3-with-specific-engine-id.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py index ca1d1fe..18b3a47 100644 --- 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 @@ -1,18 +1,23 @@ -# -# 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 -# +""" +Discover SNMPv3 SecurityEngineId +++++++++++++++++++++++++++++++++ + +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() @@ -38,10 +43,10 @@ snmpEngine.observer.registerObserver( authData = UsmUserData('non-existing-user') -for errorIndication, errorStatus, errorIndex, \ - varBinds in getCmd(snmpEngine, authData, - transportTarget, ContextData()): - break +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(snmpEngine, authData, transportTarget, ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0))) +) # See if our SNMP engine received REPORT PDU containing securityEngineId @@ -60,22 +65,22 @@ print('Remote securityEngineId = %s' % securityEngineId.prettyPrint()) 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + getCmd(snmpEngine, + authData, + transportTarget, + 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 '?' - ) +elif 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())) + ) +else: + for name, val in varBinds: + print('%s = %s' % (name.prettyPrint(), val.prettyPrint())) diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c-limiting-number-of-rows.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c-limiting-number-of-rows.py new file mode 100644 index 0000000..ac4887a --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c-limiting-number-of-rows.py @@ -0,0 +1,47 @@ +""" +Walk MIB, limit number of response rows ++++++++++++++++++++++++++++++++++++++++ + +Send a series of SNMP GETBULK requests 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 OR + number of response rows reach fixed value (20) + +Functionally similar to: + +| $ snmpbulkwalk -v2c -c public demo.snmplabs.com \ +| -Cn0 -Cr25 \ +| 1.3.6.1.2.1.2.2 1.3.6.1.2.1.2.3 + +"""# +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')), + lexicographicMode=False, maxRows=20): + + if errorIndication: + print(errorIndication) + break + elif 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-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py index 0bed0f0..bf167e4 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py @@ -1,15 +1,23 @@ -# -# 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 -# +""" +Bulk walk MIB ++++++++++++++ + +Send a series of SNMP GETBULK requests 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 + +Functionally similar to: + +| $ snmpbulkwalk -v2c -c public demo.snmplabs.com \ +| -Cn0 -Cr25 \ +| 1.3.6.1.2.1.2.2 1.3.6.1.2.1.2.3 + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -20,19 +28,19 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.3')), + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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-limiting-number-of-packets.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-limiting-number-of-packets.py new file mode 100644 index 0000000..4cb303e --- /dev/null +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-limiting-number-of-packets.py @@ -0,0 +1,44 @@ +""" +Walk Agent, limit number of packets ++++++++++++++++++++++++++++++++++++ + +Send a series of SNMP GETBULK requests 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 SNMPv2-MIB::system +* run till end-of-mib condition is reported by Agent OR + maxCalls == 10 request-response interactions occur + +Functionally similar to: + +| $ snmpbulkwalk -v3 -lnoAuthNoPriv -u usr-none-none -Cn0 -Cr50 \ +| demo.snmplabs.com SNMPv2-MIB::system + +"""# +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('SNMPv2-MIB', 'system')), + maxCalls=10): + + if errorIndication: + print(errorIndication) + break + elif 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 index 9b255f3..6a09a13 100644 --- 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 @@ -1,16 +1,25 @@ -# -# 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 -# +""" +Fetch scalar and table variables +++++++++++++++++++++++++++++++++ + +Send a series of SNMP GETBULK 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 +* 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 + +Functionally similar to: + +| $ snmpbulkwalk -v3 -lauthPriv -u usr-md5-none -A authkey1 -X privkey1 \ +| -Cn1, -Cr25 \ +| demo.snmplabs.com \ +| IP-MIB::ipAdEntAddr \ +| IP-MIB::ipAddrEntry + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -22,19 +31,18 @@ for errorIndication, \ 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 + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 5d1eb81..9c97465 100644 --- 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 @@ -1,17 +1,23 @@ -# -# 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 -# +""" +Custom PySNMP MIBs location ++++++++++++++++++++++++++++ + +Send a series of SNMP GETBULK requests 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 within TCP-MIB::tcpConnTable column +* TCP-MIB Python module will be searched by a user-specified filesystem + path (/opt/mib/pysnmp) and in Python package (python_packaged_mibs) + which should be in sys.path + +Functionally similar to: + +| $ snmpbulkwalk -v3 -lnoAuthNoPriv -u usr-none-none -Cn0 -Cr50 \ +| demo.snmplabs.com TCP-MIB::tcpConnTable + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -21,22 +27,19 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('TCP-MIB', 'tcpConnTable').addMibSource('/opt/mibs/pysnmp').addMibSource('python_packaged_mibs')), + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index b1304ae..a77dc49 100644 --- 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 @@ -1,40 +1,42 @@ -## -# 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 +""" +Walk multiple Agents at once +++++++++++++++++++++++++++++ + +Iterate over MIBs of multiple SNMP Agents asynchronously 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.rfc3413.oneliner.cmdgen import * # 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')) ) ), + ( CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(ObjectIdentity('1.3.6.1.2.1')), + ObjectType(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')), ) ), + ( CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(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')), ) ), + ( UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ( ObjectType(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')), ) ) + ( UsmUserData('usr-md5-none', 'authkey1'), + Udp6TransportTarget(('::1', 161)), + ( ObjectType(ObjectIdentity('IF-MIB', 'ifTable')), ) ) # N-th target # ... ) @@ -47,44 +49,38 @@ def cbFun(snmpEngine, sendRequestHandle, errorIndication, if errorIndication: print(errorIndication) return - if errorStatus: + elif 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 idx, varBind in enumerate(varBindTable[-1]): + if varBind[1] is not None and varBindHead[idx] <= varBind[0]: + break # still in table + 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())) + for varBindRow in varBindTable: + for varBind in varBindRow: + print(' = '.join([ x.prettyPrint() for x in varBind ])) - return True # continue table retrieval + return True # continue table retrieval -snmpEngine = engine.SnmpEngine() +snmpEngine = SnmpEngine() -cmdGen = cmdgen.AsyncCommandGenerator() +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, + snmpEngine, authData, transportTarget, ContextData(), varBinds, # User-space callback function and its context - (cbFun, (varBindHead, authData, transportTarget)), - lookupNames=True, lookupValues=True + (cbFun, (varBindHead, authData, transportTarget)) ) 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 index 0367eb3..1d459ef 100644 --- 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 @@ -1,29 +1,33 @@ -# -# 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. -# +""" +Search PySNMP MIBs at custom location ++++++++++++++++++++++++++++++++++++++ + +Perform SNMP GETNEXT operation and resolve received var-binds with +PySNMP MIB(s) found at a custom location. Use the following SNMP 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 + +The pysnmp engine maintains its own MIB search path which is used +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. + +Functionally similar to: + +| $ snmpwalk -v1 -c public -ObentU demo.snmplabs.com \ +| IF-MIB::ifDescr IF-MIB::ifType + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -34,19 +38,18 @@ for errorIndication, \ 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 + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 1c3a998..7530a39 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py @@ -1,18 +1,31 @@ -# -# 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 -# +""" +Fetch whole SNMP table +++++++++++++++++++++++ + +Send a series of SNMP GETNEXT requests 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 + +Functionally similar to: + +| $ snmpwalk -v1 -c public demo.snmplabs.com \ +| IF-MIB::ifDescr \ +| IF-MIB::ifType \ +| IF-MIB::ifMtu \ +| IF-MIB::ifSpeed \ +| IF-MIB::ifPhysAddress \ +| IF-MIB::ifType \ + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ - errorStatus, errorIndex, \ + errorStatus, \ + errorIndex, \ varBinds in nextCmd(SnmpEngine(), CommunityData('public', mpModel=0), UdpTransportTarget(('demo.snmplabs.com', 161)), @@ -22,19 +35,19 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('IF-MIB', 'ifType')), + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index e59b07b..a79fefb 100644 --- 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 @@ -1,25 +1,24 @@ -# -# 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. -# +""" +Waive MIB lookup +++++++++++++++++ + +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 +* do not resolve response OIDs and values into human-freidly form + +False lookupNames and lookupValues keyword arguments could make pysnmp +waiving OIDs and values resolution in response variable-bindings, +into human-friendly form. + +Functionally similar to: + +| $ snmpwalk -v2c -c public -ObentU demo.snmplabs.com 1.3.6.1.2.1 + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -29,19 +28,18 @@ for errorIndication, \ 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 + lookupMib=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 178cd09..514a6ce 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py @@ -1,14 +1,21 @@ -# -# 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 -# +""" +Fetch two OID branches +++++++++++++++++++++++ + +Send a series of 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 + +Functionally similar to: + +| $ snmpwalk -v2c -c public demo.snmplabs.com \ +| 1.3.6.1.2.1.2.2.1.2 1.3.6.1.2.1.2.2.1.3 + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -18,19 +25,19 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.3')), + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 01c74be..4a0877d 100644 --- 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 @@ -1,17 +1,23 @@ -# -# 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 -# +""" +Preload PySNMP MIBs ++++++++++++++++++++ + +Send a series of 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 OIDs starting from 1.3.6 +* preload all Python MIB modules found in search path + +Functionally similar to: + +| $ snmpwalk -v3 -l authPriv -u usr-md5-des -A authkey1 -X privkey1 \ +| -m ALL +| udp6:[::1]:161 \ +| 1.3.6 + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -20,20 +26,18 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('1.3.6').loadMibs())): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 658f6fc..4c0568a 100644 --- 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 @@ -1,15 +1,21 @@ -# -# 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 -# +""" +Walk whole MIB +++++++++++++++ + +Send a series of 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 + +Functionally similar to: + +| $ snmpwalk -v3 -lauthPriv -u usr-md5-none -A authkey1 -X privkey1 \ +| demo.snmplabs.com IF-MIB:: + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -18,20 +24,19 @@ for errorIndication, \ UsmUserData('usr-md5-none', 'authkey1'), UdpTransportTarget(('demo.snmplabs.com', 161)), ContextData(), - ObjectType(ObjectIdentity('IF-MIB', '')), - lookupValues=True): - # Check for errors and print out results + ObjectType(ObjectIdentity('IF-MIB')), + lexicographicMode=False): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index 99eba51..19d7950 100644 --- 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 @@ -1,17 +1,23 @@ -# -# 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 -# +""" +Fetch fixed amount of MIB variables ++++++++++++++++++++++++++++++++++++ + +Send a series of SNMP GETNEXT requests using the following options: + +* with SNMPv3, user 'usr-sha-aes128', SHA auth, AES128 privacy +* over UDP/IPv4 +* to an Agent at demo.snmplabs.com:161 +* for all OIDs past SNMPv2-MIB +* run till end-of-mib condition is reported by Agent OR maxRows == 100 +* ignoring non-increasing OIDs whenever reported by Agent + +Functionally similar to: + +| $ snmpwalk -v3 -lauthPriv -u usr-sha-aes128 -A authkey1 -X privkey1 \ +| -a MD5 -x AES demo.snmplabs.com \ +| SNMPv2-MIB::system + +"""# from pysnmp.entity.rfc3413.oneliner.cmdgen import * for errorIndication, \ @@ -22,21 +28,19 @@ for errorIndication, \ 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 + ObjectType(ObjectIdentity('SNMPv2-MIB')), + maxRows=100, ignoreNonIncreasingOid=True): + if errorIndication: print(errorIndication) break - else: - if errorStatus: - print('%s at %s' % ( - errorStatus.prettyPrint(), - errorIndex and varBinds[int(errorIndex)-1][0] or '?' - ) + elif 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 ])) + ) + 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 index bba4a84..9b9fae7 100644 --- a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py +++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py @@ -1,36 +1,49 @@ -# -# 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) -# +""" +SET scalars values +++++++++++++++++++ + +Send SNMP SET request using the following options: + +* with SNMPv1, community 'public' +* over IPv4/UDP +* to an Agent at demo.snmplabs.com:161 +* setting three var-bindings to new values + +Please note, that in this example MIB lookup is only used +for the second var-bindins. For the rest, value types are +inferred from passed objects. + +Functionally similar to: + +| $ snmpset -v1 -c public demo.snmplabs.com \ +| 1.3.6.1.2.1.1.9.1.2.1 o 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 s "new system name" + +"""# 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 + +errorIndication, errorStatus, errorIndex, varBinds = next( + setCmd(SnmpEngine(), + CommunityData('public', mpModel=0), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.9.1.2.1'), + ObjectIdentifier('1.3.6.1.4.1.20408.1.1')), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.9.1.2.1'), + '1.3.6.1.4.1.20408.1.1'), + ObjectType(ObjectIdentity('1.3.6.1.2.1.1.9.1.3.1'), + OctetString('new system name'))) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index 46d4c5e..62f7595 100644 --- 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 @@ -1,36 +1,39 @@ -# -# 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) -# +""" +Coerce value to SET to MIB spec ++++++++++++++++++++++++++++++++ + +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) + +Functionally similar to: + +| $ snmpset -v2c -c public demo.snmplabs.com SNMPv2-MIB::sysDescr.0 \ +| = "new system name" + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + setCmd(SnmpEngine(), + CommunityData('public'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name')) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index 9c5b20f..f017882 100644 --- 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 @@ -1,43 +1,45 @@ -# -# 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) -# +""" +Custom ContextEngineId and ContextName +++++++++++++++++++++++++++++++++++++++ + +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) + +Functionally similar to: + +| $ snmpset -v3 -u usr-md5-des -l authPriv -A authkey1 -X privkey1 \ +| -E 80004fb805636c6f75644dab22cc -n a172334d7d97871b72241397f713fa12 \ +| demo.snmplabs.com \ +| SNMPv2-MIB::sysORDescr.1 = "new system name" + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + setCmd(SnmpEngine(), + UsmUserData('usr-md5-none', 'authkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(contextEngineId=OctetString(hexValue='80004fb805636c6f75644dab22cc'), + contextName='da761cfc8c94d3aceef4f60f049105ba'), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name')) +) + +if errorIndication: + print(errorIndication) +elif 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/manager/cmdgen/set-v3-with-custom-contextengineid.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py index 22d165d..46ef275 100644 --- 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 @@ -1,44 +1,43 @@ -# -# 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) -# +""" +Custom ContextEngineId +++++++++++++++++++++++ + +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 coerced from MIB) + +Functionally similar to: + +| $ snmpset -v3 -u usr-md5-des -l authPriv -A authkey1 -X privkey1 \ +| -E 80004fb805636c6f75644dab22cc demo.snmplabs.com \ +| SNMPv2-MIB::sysORDescr.1 = "new system name" + +"""# 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 +errorIndication, errorStatus, errorIndex, varBinds = next( + setCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1', + securityEngineId=OctetString(hexValue='80004fb805636c6f75644dab22cc')), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysName', 0), + 'new system name')) +) + +if errorIndication: + print(errorIndication) +elif 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/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 index 379a0dd..e72587d 100644 --- 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 @@ -1,39 +1,41 @@ -# -# 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 -# +""" +Coerce value to SET to MIB spec ++++++++++++++++++++++++++++++++ + + +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) + +Functionally similar to: + +| $ snmpset -v3 -u usr-md5-des -l authPriv -A authkey1 -X privkey1 \ +| demo.snmplabs.com SNMPv2-MIB::sysORDescr.1 = "new system name" +| + +"""# 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 + +errorIndication, errorStatus, errorIndex, varBinds = next( + setCmd(SnmpEngine(), + UsmUserData('usr-md5-des', 'authkey1', 'privkey1'), + UdpTransportTarget(('demo.snmplabs.com', 161)), + ContextData(), + ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1), + 'new system name')) +) + +if errorIndication: + print(errorIndication) +elif 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 ])) |