summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorelie <elie>2015-07-06 21:30:04 +0000
committerelie <elie>2015-07-06 21:30:04 +0000
commitf5fff54ff3737b739e7597ab16591d28d3c7e6d6 (patch)
treed7756674acb22b60eaf8a226444d7c1941998605
parent286c02d388e34fc3c0697a5f4ffaa7d5af9070ce (diff)
downloadpysnmp-f5fff54ff3737b739e7597ab16591d28d3c7e6d6.tar.gz
synchronous oneliner apps redesigned to offer Python generator-based
API along with more comprehensive set of accepted parameters
-rw-r--r--CHANGES.txt2
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py42
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py46
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py43
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py44
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py36
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py34
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py27
-rw-r--r--examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py37
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py103
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py87
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py35
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py49
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py37
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py34
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py36
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py35
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py40
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py40
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py33
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py81
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py38
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py40
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py42
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py90
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py52
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py40
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py47
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py36
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py39
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py37
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py42
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py36
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py36
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py43
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py44
-rw-r--r--examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py39
-rw-r--r--pysnmp/entity/rfc3413/oneliner/cmdgen.py553
-rw-r--r--pysnmp/entity/rfc3413/oneliner/ntforg.py78
39 files changed, 1969 insertions, 284 deletions
diff --git a/CHANGES.txt b/CHANGES.txt
index 7b751e8..23478d7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -19,6 +19,8 @@ Revision 4.3.0
* Promote the use of dedicated classes for dealing with OID-value pairs.
Instances of those classes resemble OBJECT-IDENTITY, OBJECT-TYPE and
NOTIFICATION-TYPE MIB structures.
+ * Synchronous oneliner apps redesigned to offer Python generator-based
+ API along with a more comprehensive set of accepted parameters.
* keep backward compatibility for all existing major/documented interfaces
- Execution Observer facility implemented to give app an inside view
of SNMP engine inner workings. This is thought to be a generic
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py
new file mode 100644
index 0000000..6eebaa9
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v2c-with-mib-lookup.py
@@ -0,0 +1,42 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv2c
+# * with community name 'public'
+# * over IPv4/UDP
+# * send INFORM notification
+# * with TRAP ID 'coldStart' specified as a MIB symbol
+# * include managed object information specified as a MIB symbol
+# * perform response OIDs and values resolution at MIB
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'inform',
+ NotificationType(
+ ObjectIdentity('SNMPv2-MIB', 'coldStart')
+ ).addVarBinds(
+ ( ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysName', 0), 'my system') )
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py
new file mode 100644
index 0000000..071c336
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextengineid.py
@@ -0,0 +1,46 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv3
+# * with user 'usr-md5-none', MD5 auth, no priv
+# * send INFORM notification
+# * in behalf of contextEngineId 0x8000000004030201, contextName ''
+# * over IPv4/UDP
+# * with TRAP ID 'warmStart' specified as a string OID
+#
+# Sending SNMPv3 Notification in behalf of non-default ContextEngineId
+# requires having a collection of Managed Objects registered under
+# the ContextEngineId being used.
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1'),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(
+ rfc1902.OctetString(hexValue='8000000004030201')
+ ),
+ 'inform',
+ NotificationType(
+ ObjectIdentity('1.3.6.1.6.3.1.1.5.2')
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py
new file mode 100644
index 0000000..ed87fdc
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3-with-custom-contextname.py
@@ -0,0 +1,43 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv3
+# * with user 'usr-md5-none', MD5 auth, no priv
+# * send INFORM notification
+# * in behalf of contextEngineId = SnmpEngineId, contextName 'my-context'
+# * over IPv4/UDP
+# * with TRAP ID 'warmStart' specified as a string OID
+#
+# Sending SNMPv3 Notification in behalf of non-default ContextName
+# requires having a collection of Managed Objects registered under
+# the ContextName being used.
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1'),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(contextName='my-context'),
+ 'inform',
+ NotificationType(
+ ObjectIdentity('1.3.6.1.6.3.1.1.5.2')
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py
new file mode 100644
index 0000000..2996eb5
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/inform-v3.py
@@ -0,0 +1,44 @@
+##
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv3
+# * with user 'usr-md5-des', auth: MD5, priv DES
+# * over IPv4/UDP
+# * send INFORM notification
+# * with TRAP ID 'warmStart' specified as a string OID
+# * include managed object information 1.3.6.1.2.1.1.5.0 = 'system name'
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'inform',
+ NotificationType(
+ ObjectIdentity('1.3.6.1.6.3.1.1.5.2')
+ ).addVarBinds(
+ ( ObjectType(ObjectIdentity('1.3.6.1.2.1.1.5.0'),
+ rfc1902.OctetString('system name')) )
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py
new file mode 100644
index 0000000..1d2b825
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-custom-values.py
@@ -0,0 +1,36 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv1
+# * with community name 'public'
+# * over IPv4/UDP
+# * send TRAP notification
+# * with Generic Trap #6 (enterpriseSpecific) and Specific Trap 432
+# * overriding Uptime value with 12345
+# * overriding Agent Address with '127.0.0.1'
+# * overriding Enterprise OID with 1.3.6.1.4.1.20408.4.1.1.2
+# * include managed object information '1.3.6.1.2.1.1.1.0' = 'my system'
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'trap',
+ NotificationType(
+ ObjectIdentity('1.3.6.1.4.1.20408.4.1.1.2.0.432'),
+ ).addVarBinds(
+ ('1.3.6.1.2.1.1.3.0', 12345),
+ ('1.3.6.1.6.3.18.1.3.0', '127.0.0.1'),
+ ('1.3.6.1.6.3.1.1.4.3.0', '1.3.6.1.4.1.20408.4.1.1.2'),
+ ('1.3.6.1.2.1.1.1.0', rfc1902.OctetString('my system'))
+ )):
+ if errorIndication:
+ print(errorIndication)
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py
new file mode 100644
index 0000000..a704146
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v1-with-default-values.py
@@ -0,0 +1,34 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv1
+# * with community name 'public'
+# * over IPv4/UDP
+# * send TRAP notification
+# * with Generic Trap #1 (warmStart) and Specific Trap 0
+# * with default Uptime
+# * with default Agent Address
+# * with Enterprise OID 1.3.6.1.4.1.20408.4.1.1.2
+# * include managed object information '1.3.6.1.2.1.1.1.0' = 'my system'
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'trap',
+ NotificationType(
+ ObjectIdentity('1.3.6.1.6.3.1.1.5.2')
+ ).addVarBinds(
+ ('1.3.6.1.6.3.1.1.4.3.0', '1.3.6.1.4.1.20408.4.1.1.2'),
+ ('1.3.6.1.2.1.1.1.0', rfc1902.OctetString('my system'))
+ )):
+ if errorIndication:
+ print(errorIndication)
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py
new file mode 100644
index 0000000..b243f43
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v2c-with-mib-lookup.py
@@ -0,0 +1,27 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv2c
+# * with community name 'public'
+# * over IPv4/UDP
+# * send TRAP notification
+# * with TRAP ID 'coldStart' specified as a MIB symbol
+# * include managed object information specified as a MIB symbol
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'trap',
+ NotificationType(
+ ObjectIdentity('SNMPv2-MIB', 'coldStart')
+ )):
+ if errorIndication:
+ print(errorIndication)
diff --git a/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py
new file mode 100644
index 0000000..bbc9194
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/agent/ntforg/trap-v3-with-custom-engineid.py
@@ -0,0 +1,37 @@
+#
+# Notification Originator
+#
+# Send SNMP notification using the following options:
+#
+# * SNMPv3
+# * with local snmpEngineId = 0x8000000001020304 (must configure at Receiver)
+# * with user 'usr-sha-aes128', auth: SHA, priv: AES128
+# * over IPv4/UDP
+# * send TRAP notification
+# * with TRAP ID 'authenticationFailure' specified as a MIB symbol
+# * do not include any additional managed object information
+#
+# SNMPv3 TRAPs requires pre-sharing the Notification Originator's
+# value of SnmpEngineId with Notification Receiver. To facilitate that
+# we will use static (e.g. not autogenerated) version of snmpEngineId.
+#
+from pysnmp.entity.rfc3413.oneliner.ntforg import *
+from pysnmp.proto.rfc1902 import OctetString
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in \
+ sendNotification(SnmpEngine(
+ OctetString(hexValue='8000000001020304')
+ ),
+ UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1',
+ authProtocol=usmHMACSHAAuthProtocol,
+ privProtocol=usmAesCfb128Protocol),
+ UdpTransportTarget(('localhost', 162)),
+ ContextData(),
+ 'trap',
+ NotificationType(
+ ObjectIdentity('SNMPv2-MIB', 'authenticationFailure')
+ )):
+ if errorIndication:
+ print(errorIndication)
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py
new file mode 100644
index 0000000..85586ed
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-snmp-engines.py
@@ -0,0 +1,103 @@
+#
+# Asynchronous Command Generator
+#
+# Send SNMP GET requests using multiple independend SNMP engines
+# with the following options:
+#
+# * with SNMPv1, community 'public' and
+# with SNMPv2c, community 'public' and
+# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy
+# * over IPv4/UDP and
+# over IPv6/UDP
+# * to an Agent at demo.snmplabs.com:161 and
+# to an Agent at [::1]:161
+# * for instances of SNMPv2-MIB::sysDescr.0 and
+# SNMPv2-MIB::sysLocation.0 MIB objects
+#
+# Within this script we have a single asynchronous TransportDispatcher
+# and a single UDP-based transport serving two independent SNMP engines.
+# We use a single instance of AsyncCommandGenerator with each of
+# SNMP Engines to comunicate GET command request to remote systems.
+#
+# When we receive a [response] message from remote system we use
+# a custom message router to choose what of the two SNMP engines
+# data packet should be handed over. The selection criteria we
+# employ here is based on peer's UDP port number. Other selection
+# criterias are also possible.
+#
+from pysnmp.entity.rfc3413.oneliner import cmdgen
+from pysnmp.entity import engine
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+
+# List of targets in the followin format:
+# ( ( authData, transportTarget, varNames ), ... )
+targets = (
+ # 1-st target (SNMPv1 over IPv4/UDP)
+ ( cmdgen.CommunityData('public', mpModel=0),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # 2-nd target (SNMPv2c over IPv4/UDP)
+ ( cmdgen.CommunityData('public'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 1161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # 3-nd target (SNMPv3 over IPv4/UDP)
+ ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 2161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) )
+ # N-th target
+ # ...
+)
+
+# Wait for responses or errors
+def cbFun(snmpEngine, sendRequestHandle, errorIndication,
+ errorStatus, errorIndex, varBinds, cbCtx):
+ (snmpEngine, authData, transportTarget) = cbCtx
+ print('snmpEngine %s: %s via %s' %
+ (snmpEngine.snmpEngineID.prettyPrint(), authData, transportTarget)
+ )
+ if errorIndication:
+ print(errorIndication)
+ return 1
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ return 1
+
+ for oid, val in varBinds:
+ if val is None:
+ print(oid.prettyPrint())
+ else:
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+
+
+# Instantiate the single transport dispatcher object
+transportDispatcher = AsyncoreDispatcher()
+
+# Setup a custom data routing function to select snmpEngine by transportDomain
+transportDispatcher.registerRoutingCbFun(
+ lambda td,ta,d: ta[1] % 3 and 'A' or 'B'
+)
+
+snmpEngineA = engine.SnmpEngine()
+snmpEngineA.registerTransportDispatcher(transportDispatcher, 'A')
+
+snmpEngineB = engine.SnmpEngine()
+snmpEngineB.registerTransportDispatcher(transportDispatcher, 'B')
+
+cmdGen = cmdgen.AsyncCommandGenerator()
+
+for authData, transportTarget, varBinds in targets:
+ snmpEngine = transportTarget.getTransportInfo()[1][1] % 3 and \
+ snmpEngineA or snmpEngineB
+ cmdGen.getCmd(
+ snmpEngine, authData, transportTarget, cmdgen.ContextData(), varBinds,
+ (cbFun, (snmpEngine, authData, transportTarget))
+ )
+
+transportDispatcher.runDispatcher()
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py
new file mode 100644
index 0000000..e50a54b
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-async-multiple-transports-and-protocols.py
@@ -0,0 +1,87 @@
+#
+# Asynchronous Command Generator
+#
+# Send a bunch of SNMP GET requests all at once using the following options:
+#
+# * with SNMPv1, community 'public' and
+# with SNMPv2c, community 'public' and
+# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy
+# * over IPv4/UDP and
+# over IPv6/UDP
+# * to an Agent at demo.snmplabs.com:161 and
+# to an Agent at [::1]:161
+# * for instances of SNMPv2-MIB::sysDescr.0 and
+# SNMPv2-MIB::sysLocation.0 MIB objects
+#
+from pysnmp.entity import engine
+from pysnmp.entity.rfc3413.oneliner import cmdgen
+
+# List of targets in the followin format:
+# ( ( authData, transportTarget, varNames ), ... )
+targets = (
+ # 1-st target (SNMPv1 over IPv4/UDP)
+ ( cmdgen.CommunityData('public', mpModel=0),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # 2-nd target (SNMPv2c over IPv4/UDP)
+ ( cmdgen.CommunityData('public'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # 3-nd target (SNMPv2c over IPv4/UDP) - same community and
+ # different transport address.
+ ( cmdgen.CommunityData('public'),
+ cmdgen.UdpTransportTarget(('localhost', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysContact', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysName', 0))) ),
+ # 4-nd target (SNMPv3 over IPv4/UDP)
+ ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # 5-th target (SNMPv3 over IPv6/UDP)
+ ( cmdgen.UsmUserData('usr-md5-none', 'authkey1'),
+ cmdgen.Udp6TransportTarget(('::1', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'sysLocation', 0) )) ),
+ # N-th target
+ # ...
+)
+
+# Wait for responses or errors
+def cbFun(snmpEngine, sendRequestHandle, errorIndication,
+ errorStatus, errorIndex, varBinds, cbCtx):
+ (authData, transportTarget) = cbCtx
+ print('%s via %s' % (authData, transportTarget))
+ if errorIndication:
+ print(errorIndication)
+ return 1
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ return 1
+
+ for oid, val in varBinds:
+ if val is None:
+ print(oid.prettyPrint())
+ else:
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+
+snmpEngine = engine.SnmpEngine()
+
+cmdGen = cmdgen.AsyncCommandGenerator()
+
+# Submit GET requests
+for authData, transportTarget, varNames in targets:
+ cmdGen.getCmd(
+ snmpEngine, authData, transportTarget, cmdgen.ContextData(), varNames,
+ # User-space callback function and its context
+ (cbFun, (authData, transportTarget)),
+ lookupNames=True, lookupValues=True
+ )
+
+snmpEngine.transportDispatcher.runDispatcher()
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py
new file mode 100644
index 0000000..aca4129
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v1.py
@@ -0,0 +1,35 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv1, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for two instances of SNMPv2-MIB::sysDescr.0 MIB object,
+# * one in label and another in MIB symbol form
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py
new file mode 100644
index 0000000..d9fc491
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-custom-timeout-and-retries.py
@@ -0,0 +1,49 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for an OID in string form
+# * use custom timeout and request retries values
+#
+# Transport timing settings (maximum number of request retries and
+# individual request timeout in seconds) can be set on a per-target basis
+# as explained by the code that follows.
+#
+# Keep in mind that while timeout value can be specified in fractions of a
+# second, default pysnmp timer resolution is quite low (about 0.5 sec)
+# so there's no much point in using timeouts which is not a multiple of 0.5
+# Internal timer can be programmatically adjusted to finer resolution if needed.
+#
+# If retries value is set to 0, pysnmp will issue a single request. Even
+# if no response arrives, there will be no retry. Likewise, retries=1
+# means one initial request plus one retry.
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(
+ ('demo.snmplabs.com', 161), timeout=1.5, retries=0
+ ),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py
new file mode 100644
index 0000000..fba316f
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c-with-mib-compilation-and-lookup.py
@@ -0,0 +1,37 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for IF-MIB::ifInOctets.1 MIB object
+# * Pass attached MIB compiler non-default ASN.1 MIB source
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(
+ ObjectIdentity('IF-MIB', 'ifInOctets', 1).addAsn1MibSource('file:///usr/share/snmp', 'http://mibs.snmplabs.com/asn1/@mib@')
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py
new file mode 100644
index 0000000..5fe5bfe
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v2c.py
@@ -0,0 +1,34 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for two OIDs in string form
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.6.0'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py
new file mode 100644
index 0000000..716d1e9
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-over-ipv6-with-mib-lookup.py
@@ -0,0 +1,36 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols
+# * over IPv6/UDP
+# * to an Agent at [::1]:161
+# * for three OIDs: one passed as a ObjectIdentity object while others are
+# * in string form
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ Udp6TransportTarget(('::1', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0')),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.2.0')),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.3.0'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py
new file mode 100644
index 0000000..e83e0d2
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-index.py
@@ -0,0 +1,35 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3, user 'usr-none-none', no authentication, no privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for IF-MIB::ifInOctets.1 MIB object
+# * perform response OIDs and values resolution at MIB
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ UsmUserData('usr-none-none'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1)),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py
new file mode 100644
index 0000000..7b78a35
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-table-object-by-multiple-indices.py
@@ -0,0 +1,40 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3, user 'usr-sha-aes128', SHA auth, AES128 privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for TCP-MIB::tcpConnLocalAddress."0.0.0.0".22."0.0.0.0".0 MIB object
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1',
+ authProtocol=usmHMACSHAAuthProtocol,
+ privProtocol=usmAesCfb128Protocol ),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(
+ ObjectIdentity('TCP-MIB',
+ 'tcpConnLocalAddress',
+ '0.0.0.0', 22,
+ '0.0.0.0', 0)
+ )):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py
new file mode 100644
index 0000000..4770c5d
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-custom-security-name.py
@@ -0,0 +1,40 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3, user 'usr-md5-none', securityName 'myuser'
+# MD5 authentication, no privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for an OID in text form
+#
+# The securityName parameter can be thought as an alias to userName and
+# allows you to address a USM Table row just as userName does. However
+# securityName can be made human-readable, also it is not an index in
+# usmUserTable, thus duplicate securityName parameters are possible.
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1',
+ securityName='myuser'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py
new file mode 100644
index 0000000..c01a61d
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-mib-lookup.py
@@ -0,0 +1,33 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for IF-MIB::ifInOctets.1 MIB object
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in getCmd(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifInOctets', 1))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py
new file mode 100644
index 0000000..ca1d1fe
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/get-v3-with-specific-engine-id.py
@@ -0,0 +1,81 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following scenario and options:
+#
+# * try to communicate with a SNMPv3 Engine using:
+# ** a non-existing user
+# ** over IPv4/UDP
+# ** to an Agent at demo.snmplabs.com:161
+# * if remote SNMP Engine ID is discovered, send SNMP GET request:
+# ** with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy
+# at discovered securityEngineId
+# ** to the same SNMP Engine ID
+# ** for an OID in text form
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+snmpEngine = SnmpEngine()
+
+transportTarget = UdpTransportTarget(('demo.snmplabs.com', 161))
+
+#
+# To discover remote SNMP EngineID we will tap on SNMP engine inner workings
+# by setting up execution point observer setup on INTERNAL class PDU processing
+#
+
+observerContext = {}
+
+# Register a callback to be invoked at specified execution point of
+# SNMP Engine and passed local variables at execution point's local scope
+snmpEngine.observer.registerObserver(
+ lambda e,p,v,c: c.update(securityEngineId=v['securityEngineId']),
+ 'rfc3412.prepareDataElements:internal',
+ cbCtx=observerContext
+)
+
+# Send probe SNMP request with invalid credentials
+
+authData = UsmUserData('non-existing-user')
+
+for errorIndication, errorStatus, errorIndex, \
+ varBinds in getCmd(snmpEngine, authData,
+ transportTarget, ContextData()):
+ break
+
+# See if our SNMP engine received REPORT PDU containing securityEngineId
+
+if 'securityEngineId' not in observerContext:
+ print("Can't discover peer EngineID, errorIndication: %s" % errorIndication)
+ raise Exception()
+
+securityEngineId = observerContext.pop('securityEngineId')
+
+print('Remote securityEngineId = %s' % securityEngineId.prettyPrint())
+
+#
+# Query remote SNMP Engine using usmUserTable entry configured for it
+#
+
+authData = UsmUserData('usr-md5-none', 'authkey1',
+ securityEngineId=securityEngineId)
+
+for errorIndication, errorStatus, errorIndex, \
+ varBinds in getCmd(snmpEngine, authData, \
+ transportTarget, ContextData(), \
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1.1.0'))):
+ break
+
+# Check for errors and print out results
+if errorIndication:
+ print(errorIndication)
+else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for name, val in varBinds:
+ print('%s = %s' % (name.prettyPrint(), val.prettyPrint()))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py
new file mode 100644
index 0000000..0bed0f0
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v2c.py
@@ -0,0 +1,38 @@
+#
+# Command Generator
+#
+# Send SNMP GETBULK request using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * with values non-repeaters = 0, max-repetitions = 25
+# * for two OIDs in string form
+# * stop when response OIDs leave the scopes of initial OIDs
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in bulkCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ 0, 25,
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2')),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.2.3'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py
new file mode 100644
index 0000000..9b255f3
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-over-ipv6-with-mib-lookup.py
@@ -0,0 +1,40 @@
+#
+# Command Generator
+#
+# Send SNMP GETBULK request using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols
+# * over IPv6/UDP
+# * to an Agent at [::1]:161
+# * with values non-repeaters = 1, max-repetitions = 25
+# * for IP-MIB::ipAdEntAddr and all columns of the IF-MIB::ifEntry table
+# * stop when response OIDs leave the scopes of the table OR maxRows == 20
+# * perform response OIDs and values resolution at MIB
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in bulkCmd(SnmpEngine(),
+ UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ Udp6TransportTarget(('::1', 161)),
+ ContextData(),
+ 1, 25,
+ ObjectType(ObjectIdentity('IP-MIB', 'ipAdEntAddr')),
+ ObjectType(ObjectIdentity('IP-MIB', 'ipAddrEntry')),
+ lookupNames=True, lookupValues=True, maxRows=20):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py
new file mode 100644
index 0000000..5d1eb81
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getbulk-v3-with-custom-mib-path-and-options.py
@@ -0,0 +1,42 @@
+#
+# Command Generator
+#
+# Send SNMP GETBULK request using the following options:
+#
+# * with SNMPv3, user 'usr-none-none', no authentication, no privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for all OIDs past TCP-MIB::tcpConnTable
+# * TCP-MIB will be searched by a user-specified path
+# * run till end-of-mib condition is reported by Agent OR maxRows == 100 OR
+# maxCalls == 10 request-response interactions occur
+# * ignoring non-increasing OIDs whenever reported by Agent
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in bulkCmd(SnmpEngine(),
+ UsmUserData('usr-none-none'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ 0, 50,
+ ObjectType(ObjectIdentity('TCP-MIB', 'tcpConnTable').addMibSource('/tmp/mibs')),
+ maxRows=100, maxCalls=10,
+ lexicographicMode=True,
+ ignoreNonIncreasingOid=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py
new file mode 100644
index 0000000..b1304ae
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-async-multiple-transports-and-protocols.py
@@ -0,0 +1,90 @@
+##
+# Asynchronous Command Generator
+#
+# Send a bunch of SNMP GETNEXT requests all at once using the following options:
+#
+# * with SNMPv1, community 'public' and
+# with SNMPv2c, community 'public' and
+# with SNMPv3, user 'usr-md5-des', MD5 auth and DES privacy
+# * over IPv4/UDP and
+# over IPv6/UDP
+# * to an Agent at demo.snmplabs.com:161 and
+# to an Agent at [::1]:161
+# * for multiple MIB subtrees and tables
+#
+from pysnmp.entity import engine
+from pysnmp.entity.rfc3413.oneliner import cmdgen
+
+# List of targets in the followin format:
+# ( ( authData, transportTarget, varNames ), ... )
+targets = (
+ # 1-st target (SNMPv1 over IPv4/UDP)
+ ( cmdgen.CommunityData('public', mpModel=0),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.2.1')),
+ cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.3.1')) ) ),
+ # 2-nd target (SNMPv2c over IPv4/UDP)
+ ( cmdgen.CommunityData('public'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('1.3.6.1.4.1')), ) ),
+ # 3-nd target (SNMPv3 over IPv4/UDP)
+ ( cmdgen.UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ cmdgen.UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('SNMPv2-MIB', 'system')), ) ),
+ # 4-th target (SNMPv3 over IPv6/UDP)
+ ( cmdgen.UsmUserData('usr-md5-none', 'authkey1'),
+ cmdgen.Udp6TransportTarget(('::1', 161)),
+ ( cmdgen.ObjectType(cmdgen.ObjectIdentity('IF-MIB', 'ifTable')), ) )
+ # N-th target
+ # ...
+)
+
+# Wait for responses or errors, submit GETNEXT requests for further OIDs
+def cbFun(snmpEngine, sendRequestHandle, errorIndication,
+ errorStatus, errorIndex, varBindTable, cbCtx):
+ (varBindHead, authData, transportTarget) = cbCtx
+ print('%s via %s' % (authData, transportTarget))
+ if errorIndication:
+ print(errorIndication)
+ return
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBindTable[-1][int(errorIndex)-1][0] or '?'
+ )
+ )
+ return
+ varBindTableRow = varBindTable[-1]
+ for idx in range(len(varBindTableRow)):
+ name, val = varBindTableRow[idx]
+ if val is not None and varBindHead[idx] <= name:
+ # still in table
+ break
+ else:
+ print('went out of table at %s' % (name, ))
+ return
+
+ for varBindRow in varBindTable:
+ for oid, val in varBindRow:
+ if val is None:
+ print(oid.prettyPrint())
+ else:
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+
+ return True # continue table retrieval
+
+snmpEngine = engine.SnmpEngine()
+
+cmdGen = cmdgen.AsyncCommandGenerator()
+
+# Submit initial GETNEXT requests and wait for responses
+for authData, transportTarget, varBinds in targets:
+ varBindHead = [ x[0] for x in cmdGen.makeVarBinds(snmpEngine, varBinds ) ]
+ cmdGen.nextCmd(
+ snmpEngine, authData, transportTarget, cmdgen.ContextData(), varBinds,
+ # User-space callback function and its context
+ (cbFun, (varBindHead, authData, transportTarget)),
+ lookupNames=True, lookupValues=True
+ )
+
+snmpEngine.transportDispatcher.runDispatcher()
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py
new file mode 100644
index 0000000..0367eb3
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1-with-custom-mibs-location.py
@@ -0,0 +1,52 @@
+#
+# GETNEXT Command Generator Application
+#
+# Perform SNMP GETNEXT operation with the following options:
+#
+# with SNMPv1, community 'public'
+# over IPv4/UDP
+# to an Agent at demo.snmplabs.com:161
+# two columns of the IF-MIB::ifEntry table
+# search for pysnmp MIBs at specific filesystem location
+# stop when response OIDs leave the scopes of initial OIDs
+#
+# This script performs similar to the following Net-SNMP command:
+#
+# $ snmpwalk -v1 -c public -ObentU demo.snmplabs.com IF-MIB::ifDescr IF-MIB::ifType
+#
+# The pysnmp engine maintains its own MIB search path which is usued
+# for searching for MIB modules by name. By default pysnmp will search
+# its built-in MIB repository (pysnmp.smi.mibs.instances, pysnmp.smi.mibs)
+# and pysnmp-mibs package (pysnmp_mibs). Those two entries are normally
+# found relative to $PYTHONPATH. Besides searching MIBs as Python modules
+# (which could be egg-ed or not), pysnmp can also search for stand-alone
+# files in given directories. The latter could be specified by
+# ObjectIdentity.addMibSource() calls. New search path entries are added
+# in front of existing ones in search path.
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifDescr').addMibSource('/tmp/mymibs')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifType')),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py
new file mode 100644
index 0000000..1c3a998
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v1.py
@@ -0,0 +1,40 @@
+#
+# Command Generator
+#
+# Send SNMP GETNEXT request using the following options:
+#
+# * with SNMPv1, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for some columns of the IF-MIB::ifEntry table
+# * stop when response OIDs leave the scopes of initial OIDs
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifDescr')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifType')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifMtu')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifSpeed')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifPhysAddress')),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifType'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py
new file mode 100644
index 0000000..e59b07b
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c-with-mib-resolution.py
@@ -0,0 +1,47 @@
+#
+# GETNEXT Command Generator Application
+#
+# Perform SNMP GETNEXT operation with the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for an OID in string form
+# * resolve response OIDs and values into human-freidly form
+# * stop when response OIDs leave the scopes of initial OIDs
+#
+# This script performs similar to the following Net-SNMP command:
+#
+# $ snmpwalk -v2c -c public demo.snmplabs.com 1.3.6.1.2.1.1
+#
+# The lookupNames and lookupValues keyword arguments will make pysnmp
+# trying to resolve OIDs and values, in response variable-bindings,
+# into human-friendly form. If response OIDs do not belong to any of
+# currently loaded MIBs, unresolved OIDs and values will still be
+# returned.
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.1')),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py
new file mode 100644
index 0000000..178cd09
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v2c.py
@@ -0,0 +1,36 @@
+#
+# Command Generator
+#
+# Send SNMP GETNEXT requests using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for two OIDs in string form
+# * stop when response OIDs leave the scopes of initial OIDs
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.2')),
+ ObjectType(ObjectIdentity('1.3.6.1.2.1.2.2.1.3'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py
new file mode 100644
index 0000000..01c74be
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-over-ipv6-with-mib-lookup.py
@@ -0,0 +1,39 @@
+#
+# Command Generator
+#
+# Send SNMP GETNEXT requests using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols
+# * over IPv6/UDP
+# * to an Agent at [::1]:161
+# * for all columns of the IF-MIB::ifEntry table
+# * stop when response OIDs leave the scopes of the table
+# * perform response OIDs and values resolution at MIB
+#
+# make sure IF-MIB.py is in search path
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ Udp6TransportTarget(('::1', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', 'ifEntry')),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py
new file mode 100644
index 0000000..658f6fc
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-mib-lookup.py
@@ -0,0 +1,37 @@
+#
+# Command Generator
+#
+# Send SNMP GETNEXT requests using the following options:
+#
+# * with SNMPv3, user 'usr-md5-none', MD5 authentication, no privacy
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * for all OIDs in IF-MIB
+# * stop when response OIDs leave the scopes of the table
+# * perform response values resolution at MIB
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', '')),
+ lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py
new file mode 100644
index 0000000..99eba51
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/getnext-v3-pull-whole-mib-with-options.py
@@ -0,0 +1,42 @@
+#
+# Command Generator
+#
+# Send SNMP GETNEXT requests using the following options:
+#
+# * with SNMPv3, user 'usr-sha-aes128', SHA auth, AES128 privacy
+# * over Local Domain Sockets
+# * to an Agent at demo.snmplabs.com:161
+# * for all OIDs past IF-MIB (load up all MIBs in search path)
+# * run till end-of-mib condition is reported by Agent OR maxRows == 100
+# * ignoring non-increasing OIDs whenever reported by Agent
+#
+# make sure IF-MIB.py is search path
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(SnmpEngine(),
+ UsmUserData('usr-sha-aes128', 'authkey1', 'privkey1',
+ authProtocol=usmHMACSHAAuthProtocol,
+ privProtocol=usmAesCfb128Protocol),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(ObjectIdentity('IF-MIB', '').loadMibs()),
+ lexicographicMode=True, maxRows=100,
+ ignoreNonIncreasingOid=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ break
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ break
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py
new file mode 100644
index 0000000..bba4a84
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v1-multiple-values.py
@@ -0,0 +1,36 @@
+#
+# Command Generator
+#
+# Send SNMP SET request using the following options:
+#
+# * with SNMPv1, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * setting two OIDs to new values (types explicitly specified)
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in setCmd(SnmpEngine(),
+ CommunityData('public', mpModel=0),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ('1.3.6.1.2.1.1.9.1.2.1', rfc1902.ObjectName('1.3.6.1.4.1.20408.1.1')),
+ ('1.3.6.1.2.1.1.9.1.2.1', '1.3.6.1.4.1.20408.1.1'),
+ ('1.3.6.1.2.1.1.9.1.3.1', rfc1902.OctetString('new system name'))):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py
new file mode 100644
index 0000000..46d4c5e
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v2c-with-value-type-mib-lookup.py
@@ -0,0 +1,36 @@
+#
+# Command Generator
+#
+# Send SNMP SET request using the following options:
+#
+# * with SNMPv2c, community 'public'
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB)
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in setCmd(SnmpEngine(),
+ CommunityData('public'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(
+ ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1),
+ 'new system name'
+ )):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py
new file mode 100644
index 0000000..9c5b20f
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid-and-contextname.py
@@ -0,0 +1,43 @@
+#
+# Command Generator
+#
+# Send SNMP SET request using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-none', MD5 auth and no privacy protocols
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * addressing particular set of Managed Objects at remote SNMP Engine by:
+# * contextEngineId 0x80004fb805636c6f75644dab22cc and
+# * contextName 'a172334d7d97871b72241397f713fa12'
+# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB)
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in setCmd(SnmpEngine(),
+ UsmUserData('usr-md5-none', 'authkey1'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(
+ contextEngineId=rfc1902.OctetString(hexValue='80004fb805636c6f75644dab22cc'),
+ contextName='da761cfc8c94d3aceef4f60f049105ba'
+ ),
+ ObjectType(
+ ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1),
+ 'new system name'
+ )):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py
new file mode 100644
index 0000000..22d165d
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-custom-contextengineid.py
@@ -0,0 +1,44 @@
+#
+# Command Generator
+#
+# Send SNMP GET request using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols
+# * use remote SNMP Engine ID 0x80004fb805636c6f75644dab22cc (USM
+# autodiscovery will run)
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB)
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in setCmd(SnmpEngine(),
+ UsmUserData(
+ 'usr-md5-des', 'authkey1', 'privkey1',
+ securityEngineId=rfc1902.OctetString(
+ hexValue='80004fb805636c6f75644dab22cc'
+ )
+ ),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(
+ ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1),
+ 'new system name'
+ )):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py
new file mode 100644
index 0000000..379a0dd
--- /dev/null
+++ b/examples/v3arch/asyncore/oneliner/manager/cmdgen/set-v3-with-value-type-mib-lookup.py
@@ -0,0 +1,39 @@
+#
+# Command Generator
+#
+# Send SNMP SET request using the following options:
+#
+# * with SNMPv3 with user 'usr-md5-des', MD5 auth and DES privacy protocols
+# * over IPv4/UDP
+# * to an Agent at demo.snmplabs.com:161
+# * setting SNMPv2-MIB::sysName.0 to new value (type taken from MIB)
+# * perform response OIDs and values resolution at MIB
+#
+from pysnmp.entity.rfc3413.oneliner.cmdgen import *
+from pysnmp.proto import rfc1902
+
+for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in setCmd(SnmpEngine(),
+ UsmUserData('usr-md5-des', 'authkey1', 'privkey1'),
+ UdpTransportTarget(('demo.snmplabs.com', 161)),
+ ContextData(),
+ ObjectType(
+ ObjectIdentity('SNMPv2-MIB', 'sysORDescr', 1),
+ 'new system name'
+ ),
+ lookupNames=True, lookupValues=True):
+ # Check for errors and print out results
+ if errorIndication:
+ print(errorIndication)
+ else:
+ if errorStatus:
+ print('%s at %s' % (
+ errorStatus.prettyPrint(),
+ errorIndex and varBinds[int(errorIndex)-1][0] or '?'
+ )
+ )
+ else:
+ for varBind in varBinds:
+ print(' = '.join([ x.prettyPrint() for x in varBind ]))
+ break
diff --git a/pysnmp/entity/rfc3413/oneliner/cmdgen.py b/pysnmp/entity/rfc3413/oneliner/cmdgen.py
index 2e2a88c..d646cf6 100644
--- a/pysnmp/entity/rfc3413/oneliner/cmdgen.py
+++ b/pysnmp/entity/rfc3413/oneliner/cmdgen.py
@@ -26,6 +26,9 @@ usmAesCfb192Protocol = config.usmAesCfb192Protocol
usmAesCfb256Protocol = config.usmAesCfb256Protocol
usmNoPrivProtocol = config.usmNoPrivProtocol
+# SNMP engine
+SnmpEngine = engine.SnmpEngine
+
nextID = nextid.Integer(0xffffffff)
class AsyncCommandGenerator:
@@ -241,7 +244,7 @@ class AsyncCommandGenerator:
# Async SNMP apps
def getCmd(self, snmpEngine, authData, transportTarget, contextData,
- varNames, cbInfo, lookupNames=False, lookupValues=False):
+ varBinds, cbInfo, lookupNames=False, lookupValues=False):
def __cbFun(snmpEngine, sendRequestHandle,
errorIndication, errorStatus, errorIndex,
varBinds, cbCtx):
@@ -268,7 +271,7 @@ class AsyncCommandGenerator:
addrName,
contextData.contextEngineId,
contextData.contextName,
- self.makeVarBinds(snmpEngine, [(x, self._null) for x in varNames]),
+ self.makeVarBinds(snmpEngine, varBinds),
__cbFun,
(lookupNames, lookupValues, cbFun, cbCtx)
)
@@ -307,7 +310,7 @@ class AsyncCommandGenerator:
)
def nextCmd(self, snmpEngine, authData, transportTarget, contextData,
- varNames, cbInfo, lookupNames=False, lookupValues=False):
+ varBinds, cbInfo, lookupNames=False, lookupValues=False):
def __cbFun(snmpEngine, sendRequestHandle,
errorIndication, errorStatus, errorIndex,
varBindTable, cbCtx):
@@ -330,13 +333,13 @@ class AsyncCommandGenerator:
snmpEngine,
addrName,
contextData.contextEngineId, contextData.contextName,
- self.makeVarBinds(snmpEngine, [(x, self._null) for x in varNames]),
+ self.makeVarBinds(snmpEngine, varBinds),
__cbFun,
(lookupNames, lookupValues, cbFun, cbCtx)
)
def bulkCmd(self, snmpEngine, authData, transportTarget, contextData,
- nonRepeaters, maxRepetitions, varNames, cbInfo,
+ nonRepeaters, maxRepetitions, varBinds, cbInfo,
lookupNames=False, lookupValues=False):
def __cbFun(snmpEngine, sendRequestHandle,
errorIndication, errorStatus, errorIndex,
@@ -362,12 +365,246 @@ class AsyncCommandGenerator:
contextData.contextEngineId,
contextData.contextName,
nonRepeaters, maxRepetitions,
- self.makeVarBinds(snmpEngine, [(x, self._null) for x in varNames]),
+ self.makeVarBinds(snmpEngine, varBinds),
__cbFun,
(lookupNames, lookupValues, cbFun, cbCtx)
)
-# compatibility implementation, never use this class for new applications
+# Synchronous one-liner SNMP apps
+
+def getCmd(snmpEngine, authData, transportTarget, contextData,
+ *varBinds, **kwargs):
+
+ def cbFun(snmpEngine, sendRequestHandle,
+ errorIndication, errorStatus, errorIndex,
+ varBinds, cbCtx):
+ cbCtx['errorIndication'] = errorIndication
+ cbCtx['errorStatus'] = errorStatus
+ cbCtx['errorIndex'] = errorIndex
+ cbCtx['varBinds'] = varBinds
+
+ cbCtx = {}
+
+ AsyncCommandGenerator().getCmd(
+ snmpEngine,
+ authData,
+ transportTarget,
+ contextData,
+ varBinds,
+ (cbFun, cbCtx),
+ kwargs.get('lookupNames'),
+ kwargs.get('lookupValues')
+ )
+
+ snmpEngine.transportDispatcher.runDispatcher()
+
+ yield cbCtx['errorIndication'], \
+ cbCtx['errorStatus'], cbCtx['errorIndex'], \
+ cbCtx['varBinds']
+
+def setCmd(snmpEngine, authData, transportTarget, contextData,
+ *varBinds, **kwargs):
+
+ def cbFun(snmpEngine, sendRequestHandle,
+ errorIndication, errorStatus, errorIndex,
+ varBinds, cbCtx):
+ cbCtx['errorIndication'] = errorIndication
+ cbCtx['errorStatus'] = errorStatus
+ cbCtx['errorIndex'] = errorIndex
+ cbCtx['varBinds'] = varBinds
+
+ cbCtx = {}
+
+ AsyncCommandGenerator().setCmd(
+ snmpEngine,
+ authData,
+ transportTarget,
+ contextData,
+ varBinds,
+ (cbFun, cbCtx),
+ kwargs.get('lookupNames'),
+ kwargs.get('lookupValues')
+ )
+
+ snmpEngine.transportDispatcher.runDispatcher()
+
+ yield cbCtx['errorIndication'], \
+ cbCtx['errorStatus'], cbCtx['errorIndex'], \
+ cbCtx['varBinds']
+
+def nextCmd(snmpEngine, authData, transportTarget, contextData,
+ *varBinds, **kwargs):
+
+ def cbFun(snmpEngine, sendRequestHandle,
+ errorIndication, errorStatus, errorIndex,
+ varBindTable, cbCtx):
+ cbCtx['errorIndication'] = errorIndication
+ cbCtx['errorStatus'] = errorStatus
+ cbCtx['errorIndex'] = errorIndex
+ cbCtx['varBindTable'] = varBindTable
+
+ lookupNames = kwargs.get('lookupNames', False)
+ lookupValues = kwargs.get('lookupValues', False)
+ lexicographicMode = kwargs.get('lexicographicMode', False)
+ ignoreNonIncreasingOid = kwargs.get('ignoreNonIncreasingOid', False)
+ maxRows = kwargs.get('maxRows', 0)
+ maxCalls = kwargs.get('maxCalls', 0)
+
+ cbCtx = {}
+
+ cmdGen = AsyncCommandGenerator()
+
+ initialVars = [ x[0] for x in cmdGen.makeVarBinds(snmpEngine, varBinds) ]
+
+ totalRows = totalCalls = 0
+
+ while True:
+ cmdGen.nextCmd(snmpEngine,
+ authData,
+ transportTarget,
+ contextData,
+ [ (x[0], univ.Null()) for x in varBinds ],
+ (cbFun, cbCtx),
+ kwargs.get('lookupNames'),
+ kwargs.get('lookupValues'))
+
+ snmpEngine.transportDispatcher.runDispatcher()
+
+ errorIndication = cbCtx['errorIndication']
+ errorStatus = cbCtx['errorStatus']
+ errorIndex = cbCtx['errorIndex']
+
+ if ignoreNonIncreasingOid and errorIndication and \
+ isinstance(errorIndication, errind.OidNotIncreasing):
+ errorIndication = None
+ if errorStatus or errorIndication:
+ if errorStatus == 2:
+ # Hide SNMPv1 noSuchName error which leaks in here
+ # from SNMPv1 Agent through internal pysnmp proxy.
+ errorStatus = errorStatus.clone(0)
+ errorIndex = errorIndex.clone(0)
+ yield errorIndication, errorStatus, errorIndex, varBinds
+ continue
+ else:
+ varBinds = cbCtx['varBindTable'] and cbCtx['varBindTable'][0]
+ for idx, varBind in enumerate(varBinds):
+ name, val = varBind
+ if not isinstance(val, univ.Null):
+ if lexicographicMode or initialVars[idx].isPrefixOf(name):
+ break
+ else:
+ return
+
+ totalRows += 1
+ totalCalls += 1
+
+ yield errorIndication, errorStatus, errorIndex, varBinds
+
+ if maxRows and totalRows >= maxRows or \
+ maxCalls and totalCalls >= maxCalls:
+ return
+
+def bulkCmd(snmpEngine, authData, transportTarget, contextData,
+ nonRepeaters, maxRepetitions, *varBinds, **kwargs):
+
+ def cbFun(snmpEngine, sendRequestHandle,
+ errorIndication, errorStatus, errorIndex,
+ varBindTable, cbCtx):
+ cbCtx['errorIndication'] = errorIndication
+ cbCtx['errorStatus'] = errorStatus
+ cbCtx['errorIndex'] = errorIndex
+ cbCtx['varBindTable'] = varBindTable
+
+ lookupNames = kwargs.get('lookupNames', False)
+ lookupValues = kwargs.get('lookupValues', False)
+ lexicographicMode = kwargs.get('lexicographicMode', False)
+ ignoreNonIncreasingOid = kwargs.get('ignoreNonIncreasingOid', False)
+ maxRows = kwargs.get('maxRows', 0)
+ maxCalls = kwargs.get('maxCalls', 0)
+
+ cbCtx = {}
+
+ cmdGen = AsyncCommandGenerator()
+
+ initialVars = [ x[0] for x in cmdGen.makeVarBinds(snmpEngine, varBinds) ]
+ nullVarBinds = [ False ] * len(initialVars)
+
+ totalRows = totalCalls = 0
+ stopFlag = False
+
+ while not stopFlag:
+ if maxRows and totalRows < maxRows:
+ maxRepetitions = min(maxRepetitions, maxRows-totalRows)
+
+ cmdGen.bulkCmd(snmpEngine,
+ authData,
+ transportTarget,
+ contextData,
+ nonRepeaters, maxRepetitions,
+ [ (x[0], univ.Null()) for x in varBinds ],
+ (cbFun, cbCtx),
+ kwargs.get('lookupNames'),
+ kwargs.get('lookupValues'))
+
+ snmpEngine.transportDispatcher.runDispatcher()
+
+ errorIndication = cbCtx['errorIndication']
+ errorStatus = cbCtx['errorStatus']
+ errorIndex = cbCtx['errorIndex']
+ varBindTable = cbCtx['varBindTable']
+
+ if ignoreNonIncreasingOid and errorIndication and \
+ isinstance(errorIndication, errind.OidNotIncreasing):
+ errorIndication = None
+ if errorStatus or errorIndication:
+ if errorStatus == 2:
+ # Hide SNMPv1 noSuchName error which leaks in here
+ # from SNMPv1 Agent through internal pysnmp proxy.
+ errorStatus = errorStatus.clone(0)
+ errorIndex = errorIndex.clone(0)
+ yield errorIndication, errorStatus, errorIndex, varBindTable and varBindTable[0] or []
+ continue
+ else:
+ for i in range(len(varBindTable)):
+ stopFlag = True
+ if len(varBindTable[i]) != len(initialVars):
+ varBindTable = i and varBindTable[:i-1] or []
+ break
+ for j in range(len(varBindTable[i])):
+ name, val = varBindTable[i][j]
+ if nullVarBinds[j]:
+ varBindTable[i][j] = name, rfc1905.endOfMibView
+ continue
+ stopFlag = False
+ if isinstance(val, univ.Null):
+ nullVarBinds[j] = True
+ elif not lexicographicMode and \
+ not initialVars[j].isPrefixOf(name):
+ varBindTable[i][j] = name, rfc1905.endOfMibView
+ nullVarBinds[j] = True
+ if stopFlag:
+ varBindTable = i and varBindTable[:i-1] or []
+ break
+
+ totalRows += len(varBindTable)
+ totalCalls += 1
+
+ if maxRows and totalRows >= maxRows:
+ if totalRows > maxRows:
+ varBindTable = varBindTable[:-(totalRows-maxRows)]
+ stopFlag = True
+
+ if maxCalls and totalCalls >= maxCalls:
+ stopFlag = True
+
+ for varBinds in varBindTable:
+ yield errorIndication, errorStatus, errorIndex, varBinds
+
+#
+# The rest of code in this file belongs to obsolete, compatibility wrappers.
+# Never use interfaces below for new applications!
+#
+
class AsynCommandGenerator:
def __init__(self, snmpEngine=None):
if snmpEngine is None:
@@ -428,7 +665,9 @@ class AsynCommandGenerator:
return self.__asyncCmdGen.getCmd(
self.snmpEngine,
authData, transportTarget,
- ContextData(contextEngineId, contextName), varNames, cbInfo,
+ ContextData(contextEngineId, contextName),
+ [(x, self._null) for x in varNames],
+ cbInfo,
lookupNames, lookupValues
)
@@ -482,7 +721,9 @@ class AsynCommandGenerator:
return self.__asyncCmdGen.nextCmd(
self.snmpEngine,
authData, transportTarget,
- ContextData(contextEngineId, contextName), varNames, cbInfo,
+ ContextData(contextEngineId, contextName),
+ [(x, self._null) for x in varNames],
+ cbInfo,
lookupNames, lookupValues
)
@@ -512,7 +753,8 @@ class AsynCommandGenerator:
authData, transportTarget,
ContextData(contextEngineId, contextName),
nonRepeaters, maxRepetitions,
- varNames, cbInfo,
+ [(x, self._null) for x in varNames],
+ cbInfo,
lookupNames, lookupValues
)
@@ -520,262 +762,57 @@ class AsynCommandGenerator:
class CommandGenerator:
def __init__(self, snmpEngine=None, asynCmdGen=None):
- if asynCmdGen is None:
- self.__asynCmdGen = AsynCommandGenerator(snmpEngine)
- else:
- self.__asynCmdGen = asynCmdGen
-
# compatibility attributes
- self.snmpEngine = self.__asynCmdGen.snmpEngine
- self.mibViewController = self.__asynCmdGen.mibViewController
+ self.snmpEngine = snmpEngine or SnmpEngine()
+ self.mibViewController = AsyncCommandGenerator().getMibViewController(self.snmpEngine)
def getCmd(self, authData, transportTarget, *varNames, **kwargs):
- def __cbFun(sendRequestHandle,
- errorIndication, errorStatus, errorIndex,
- varBinds, appReturn):
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBinds'] = varBinds
-
- appReturn = {}
- self.__asynCmdGen.getCmd(
- authData,
- transportTarget,
- varNames,
- (__cbFun, appReturn),
- kwargs.get('lookupNames', False),
- kwargs.get('lookupValues', False),
- kwargs.get('contextEngineId'),
- kwargs.get('contextName', null)
- )
- self.snmpEngine.transportDispatcher.runDispatcher()
- return (
- appReturn['errorIndication'],
- appReturn['errorStatus'],
- appReturn['errorIndex'],
- appReturn['varBinds']
- )
+ for x in getCmd(self.snmpEngine, authData, transportTarget,
+ ContextData(kwargs.get('contextEngineId'),
+ kwargs.get('contextName', null)),
+ *[ (x, univ.Null()) for x in varNames ],
+ **kwargs):
+ return x
def setCmd(self, authData, transportTarget, *varBinds, **kwargs):
- def __cbFun(sendRequestHandle,
- errorIndication, errorStatus, errorIndex,
- varBinds, appReturn):
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBinds'] = varBinds
-
- appReturn = {}
- self.__asynCmdGen.setCmd(
- authData,
- transportTarget,
- varBinds,
- (__cbFun, appReturn),
- kwargs.get('lookupNames', False),
- kwargs.get('lookupValues', False),
- kwargs.get('contextEngineId'),
- kwargs.get('contextName', null)
- )
- self.snmpEngine.transportDispatcher.runDispatcher()
- return (
- appReturn['errorIndication'],
- appReturn['errorStatus'],
- appReturn['errorIndex'],
- appReturn['varBinds']
- )
+ for x in setCmd(self.snmpEngine, authData, transportTarget,
+ ContextData(kwargs.get('contextEngineId'),
+ kwargs.get('contextName', null)),
+ *varBinds,
+ **kwargs):
+ return x
def nextCmd(self, authData, transportTarget, *varNames, **kwargs):
- def __cbFun(sendRequestHandle, errorIndication,
- errorStatus, errorIndex, varBindTable, cbCtx):
- (self, varBindHead, varBindTotalTable, appReturn) = cbCtx
- if (ignoreNonIncreasingOid or \
- hasattr(self, 'ignoreNonIncreasingOid') and \
- self.ignoreNonIncreasingOid ) and \
- errorIndication and \
- isinstance(errorIndication, errind.OidNotIncreasing):
- errorIndication = None
- if errorStatus or errorIndication:
- appReturn['errorIndication'] = errorIndication
- if errorStatus == 2:
- # Hide SNMPv1 noSuchName error which leaks in here
- # from SNMPv1 Agent through internal pysnmp proxy.
- appReturn['errorStatus'] = errorStatus.clone(0)
- appReturn['errorIndex'] = errorIndex.clone(0)
- else:
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBindTable'] = varBindTotalTable
- return
- else:
- varBindTableRow = varBindTable and varBindTable[-1] or varBindTable
- for idx in range(len(varBindTableRow)):
- name, val = varBindTableRow[idx]
- # XXX extra rows
- if not isinstance(val, univ.Null):
- if lexicographicMode or \
- hasattr(self, 'lexicographicMode') and \
- self.lexicographicMode: # obsolete
- if varBindHead[idx] <= name:
- break
- else:
- if varBindHead[idx].isPrefixOf(name):
- break
- else:
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBindTable'] = varBindTotalTable
- return
-
- varBindTotalTable.extend(varBindTable)
-
- if maxRows and len(varBindTotalTable) >= maxRows or \
- hasattr(self, 'maxRows') and self.maxRows and \
- len(varBindTotalTable) >= self.maxRows:
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- if hasattr(self, 'maxRows'):
- appReturn['varBindTable'] = varBindTotalTable[:self.maxRows]
- else:
- appReturn['varBindTable'] = varBindTotalTable[:maxRows]
- return
-
- if maxCalls[0] > 0:
- maxCalls[0] -= 1
- if maxCalls[0] == 0:
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBindTable'] = varBindTotalTable
- return
-
- return 1 # continue table retrieval
-
- lookupNames = kwargs.get('lookupNames', False)
- lookupValues = kwargs.get('lookupValues', False)
- contextEngineId = kwargs.get('contextEngineId')
- contextName = kwargs.get('contextName', null)
- lexicographicMode = kwargs.get('lexicographicMode', False)
- maxRows = kwargs.get('maxRows', 0)
- maxCalls = [ kwargs.get('maxCalls', 0) ]
- ignoreNonIncreasingOid = kwargs.get('ignoreNonIncreasingOid', False)
-
- varBindHead = [ univ.ObjectIdentifier(x[0]) for x in self.__asynCmdGen.makeReadVarBinds(varNames) ]
-
- appReturn = {}
- self.__asynCmdGen.nextCmd(
- authData,
- transportTarget,
- varNames,
- (__cbFun, (self, varBindHead, [], appReturn)),
- lookupNames, lookupValues,
- contextEngineId, contextName
- )
- self.snmpEngine.transportDispatcher.runDispatcher()
- return (
- appReturn['errorIndication'],
- appReturn['errorStatus'],
- appReturn['errorIndex'],
- appReturn['varBindTable']
- )
+ varBindTable = []
+ for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in nextCmd(self.snmpEngine, authData, transportTarget,
+ ContextData(kwargs.get('contextEngineId'),
+ kwargs.get('contextName', null)),
+ *[ (x, univ.Null()) for x in varNames ],
+ **kwargs):
+ if errorIndication or errorStatus:
+ return errorIndication, errorStatus, errorIndex, varBinds
+
+ varBindTable.append(varBinds)
+
+ return errorIndication, errorStatus, errorIndex, varBindTable
def bulkCmd(self, authData, transportTarget,
nonRepeaters, maxRepetitions, *varNames, **kwargs):
- def __cbFun(sendRequestHandle, errorIndication,
- errorStatus, errorIndex, varBindTable, cbCtx):
- (self, varBindHead, nullVarBinds, varBindTotalTable, appReturn) = cbCtx
- if (ignoreNonIncreasingOid or \
- hasattr(self, 'ignoreNonIncreasingOid') and \
- self.ignoreNonIncreasingOid ) and \
- errorIndication and \
- isinstance(errorIndication, errind.OidNotIncreasing):
- errorIndication = None
- if errorStatus or errorIndication:
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBindTable'] = varBindTable
- return
- else:
- stopFlag = False
- if not lexicographicMode: # cut possible extra OIDs
- stopFlag = True
- for i in range(len(varBindTable)):
- stopFlag = True
- if len(varBindTable[i]) != len(varBindHead):
- varBindTable = i and varBindTable[:i-1] or []
- break
- for j in range(len(varBindTable[i])): # dichotomy?
- name, val = varBindTable[i][j]
- if nullVarBinds[j]:
- varBindTable[i][j] = name, rfc1905.endOfMibView
- continue
- stopFlag = False
- if not isinstance(val, univ.Null):
- if not varBindHead[j].isPrefixOf(name):
- varBindTable[i][j] = name, rfc1905.endOfMibView
- nullVarBinds[j] = True
- if stopFlag:
- varBindTable = i and varBindTable[:i-1] or []
- break
-
- varBindTotalTable.extend(varBindTable)
-
- appReturn['errorIndication'] = errorIndication
- appReturn['errorStatus'] = errorStatus
- appReturn['errorIndex'] = errorIndex
- appReturn['varBindTable'] = varBindTotalTable
-
- if maxCalls[0] > 0:
- maxCalls[0] -= 1
- if maxCalls[0] == 0:
- return
-
- if maxRows and len(varBindTotalTable) >= maxRows or \
- hasattr(self, 'maxRows') and self.maxRows and \
- len(varBindTotalTable) >= self.maxRows: # obsolete
- if hasattr(self, 'maxRows'):
- appReturn['varBindTable'] = varBindTotalTable[:self.maxRows]
- else:
- appReturn['varBindTable'] = varBindTotalTable[:maxRows]
- return
-
- return not stopFlag # continue table retrieval
-
- lookupNames = kwargs.get('lookupNames', False)
- lookupValues = kwargs.get('lookupValues', False)
- contextEngineId = kwargs.get('contextEngineId')
- contextName = kwargs.get('contextName', null)
- lexicographicMode = kwargs.get('lexicographicMode', False)
- if not lexicographicMode: # obsolete
- if hasattr(self, 'lexicographicMode') and self.lexicographicMode:
- lexicographicMode = True
- maxRows = kwargs.get('maxRows', 0)
- maxCalls = [ kwargs.get('maxCalls', 0) ]
- ignoreNonIncreasingOid = kwargs.get('ignoreNonIncreasingOid', False)
-
- varBindHead = [ univ.ObjectIdentifier(x[0]) for x in self.__asynCmdGen.makeReadVarBinds(varNames) ]
- nullVarBinds = [ False ] * len(varBindHead)
-
- appReturn = {}
-
- self.__asynCmdGen.bulkCmd(
- authData,
- transportTarget,
- nonRepeaters, maxRepetitions,
- varNames,
- (__cbFun, (self, varBindHead, nullVarBinds, [], appReturn)),
- lookupNames, lookupValues,
- contextEngineId, contextName
- )
+ varBindTable = []
+ for errorIndication, \
+ errorStatus, errorIndex, \
+ varBinds in bulkCmd(self.snmpEngine, authData, transportTarget,
+ ContextData(kwargs.get('contextEngineId'),
+ kwargs.get('contextName', null)),
+ nonRepeaters, maxRepetitions,
+ *[ (x, univ.Null()) for x in varNames ],
+ **kwargs):
+ if errorIndication or errorStatus:
+ return errorIndication, errorStatus, errorIndex, varBinds
+
+ varBindTable.append(varBinds)
+
+ return errorIndication, errorStatus, errorIndex, varBindTable
- self.snmpEngine.transportDispatcher.runDispatcher()
-
- return (
- appReturn['errorIndication'],
- appReturn['errorStatus'],
- appReturn['errorIndex'],
- appReturn['varBindTable']
- )
diff --git a/pysnmp/entity/rfc3413/oneliner/ntforg.py b/pysnmp/entity/rfc3413/oneliner/ntforg.py
index 8fc20d0..6ff9215 100644
--- a/pysnmp/entity/rfc3413/oneliner/ntforg.py
+++ b/pysnmp/entity/rfc3413/oneliner/ntforg.py
@@ -23,6 +23,7 @@ usmAesCfb192Protocol = config.usmAesCfb192Protocol
usmAesCfb256Protocol = config.usmAesCfb256Protocol
usmNoPrivProtocol = config.usmNoPrivProtocol
+SnmpEngine = engine.SnmpEngine
ContextData = cmdgen.ContextData
nextID = nextid.Integer(0xffffffff)
@@ -177,6 +178,47 @@ class AsyncNotificationOriginator:
return ntforg.NotificationOriginator().sendVarBinds(snmpEngine, notifyName, contextData.contextEngineId, contextData.contextName, self.makeVarBinds(snmpEngine, varBinds), __cbFun, (lookupNames, lookupValues, cbFun, cbCtx))
+#
+# Synchronous one-liner Notification Originator application
+#
+
+def sendNotification(snmpEngine, authData, transportTarget, contextData,
+ notifyType, notificationType, **kwargs):
+
+ def cbFun(snmpEngine, sendRequestHandle,
+ errorIndication, errorStatus, errorIndex,
+ varBinds, cbCtx):
+ cbCtx['errorIndication'] = errorIndication
+ cbCtx['errorStatus'] = errorStatus
+ cbCtx['errorIndex'] = errorIndex
+ cbCtx['varBinds'] = varBinds
+
+ cbCtx = {}
+
+ AsyncNotificationOriginator().sendNotification(
+ snmpEngine,
+ authData,
+ transportTarget,
+ contextData,
+ notifyType,
+ notificationType,
+ (cbFun, cbCtx),
+ kwargs.get('lookupNames'),
+ kwargs.get('lookupValues')
+ )
+
+ snmpEngine.transportDispatcher.runDispatcher()
+
+ if cbCtx:
+ yield cbCtx['errorIndication'], \
+ cbCtx['errorStatus'], cbCtx['errorIndex'], \
+ cbCtx['varBinds']
+
+#
+# The rest of code in this file belongs to obsolete, compatibility wrappers.
+# Never use interfaces below for new applications!
+#
+
# substitute sendNotification return object for backward compatibility
class ErrorIndicationReturn:
def __init__(self, *vars): self.__vars = vars
@@ -185,9 +227,6 @@ class ErrorIndicationReturn:
def __bool__(self): return bool(self.__vars[0])
def __str__(self): return str(self.__vars[0])
-#
-# Compatibility implementation, never use this class for new applications
-#
class AsynNotificationOriginator:
def __init__(self, snmpEngine=None, snmpContext=None):
if snmpEngine is None:
@@ -274,31 +313,18 @@ class AsynNotificationOriginator:
class NotificationOriginator:
def __init__(self, snmpEngine=None, snmpContext=None, asynNtfOrg=None):
- if asynNtfOrg is None:
- self.__asynNtfOrg = AsynNotificationOriginator(
- snmpEngine, snmpContext
- )
- else:
- self.__asynNtfOrg = asynNtfOrg
+ # compatibility attributes
+ self.snmpEngine = snmpEngine or SnmpEngine()
+ self.mibViewController = AsyncNotificationOriginator().getMibViewController(self.snmpEngine)
# the varBinds parameter is legacy, use NotificationType instead
def sendNotification(self, authData, transportTarget, notifyType,
notificationType, *varBinds, **kwargs):
- def __cbFun(sendRequestHandle, errorIndication,
- errorStatus, errorIndex, varBinds, appReturn):
- appReturn[0] = ErrorIndicationReturn(
- errorIndication, errorStatus, errorIndex, varBinds
- )
-
- appReturn = { 0: ErrorIndicationReturn(None, 0, 0, ()) }
- self.__asynNtfOrg.sendNotification(
- authData, transportTarget, notifyType, notificationType,
- varBinds, (__cbFun, appReturn),
- kwargs.get('lookupNames', False),
- kwargs.get('lookupValues', False),
- kwargs.get('contextEngineId'),
- kwargs.get('contextName', null)
- )
- self.__asynNtfOrg.snmpEngine.transportDispatcher.runDispatcher()
- return appReturn[0]
+ for x in sendNotification(self.snmpEngine, authData, transportTarget,
+ ContextData(kwargs.get('contextEngineId'),
+ kwargs.get('contextName', null)),
+ notifyType,
+ notificationType.addVarBinds(*varBinds),
+ **kwargs):
+ return x