summaryrefslogtreecommitdiff
path: root/examples/v1arch
diff options
context:
space:
mode:
Diffstat (limited to 'examples/v1arch')
-rw-r--r--examples/v1arch/asyncore/agent/cmdrsp/v2c-custom-scalar-mib-objects.py145
-rw-r--r--examples/v1arch/asyncore/agent/ntforg/trap-v1-multiple-transports.py78
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/get-v1.py105
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/get-v2c-spoof-source-address.py117
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/get-v2c-udp-broadcast-agent-discovery.py99
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/getbulk-v2c.py112
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/getnext-v1.py103
-rw-r--r--examples/v1arch/asyncore/manager/cmdgen/set-v2c.py85
-rw-r--r--examples/v1arch/asyncore/manager/ntfrcv/v1-multiple-transports.py100
9 files changed, 944 insertions, 0 deletions
diff --git a/examples/v1arch/asyncore/agent/cmdrsp/v2c-custom-scalar-mib-objects.py b/examples/v1arch/asyncore/agent/cmdrsp/v2c-custom-scalar-mib-objects.py
new file mode 100644
index 0000000..5119a15
--- /dev/null
+++ b/examples/v1arch/asyncore/agent/cmdrsp/v2c-custom-scalar-mib-objects.py
@@ -0,0 +1,145 @@
+"""
+Implementing scalar MIB objects
++++++++++++++++++++++++++++++++
+
+Listen and respond to SNMP GET/GETNEXT queries with the following options:
+
+* SNMPv1 or SNMPv2c
+* with SNMP community "public"
+* over IPv4/UDP, listening at 127.0.0.1:161
+* over IPv6/UDP, listening at [::1]:161
+* serving two Managed Objects Instances (sysDescr.0 and sysUptime.0)
+
+Either of the following Net-SNMP commands will walk this Agent:
+
+| $ snmpwalk -v2c -c public 127.0.0.1 .1.3.6
+| $ snmpwalk -v2c -c public udp6:[::1] .1.3.6
+
+The Command Receiver below uses two distinct transports for communication
+with Command Generators - UDP over IPv4 and UDP over IPv6.
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp, udp6, unix
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto import api
+import time, bisect
+
+class SysDescr:
+ name = (1,3,6,1,2,1,1,1,0)
+ def __eq__(self, other): return self.name == other
+ def __ne__(self, other): return self.name != other
+ def __lt__(self, other): return self.name < other
+ def __le__(self, other): return self.name <= other
+ def __gt__(self, other): return self.name > other
+ def __ge__(self, other): return self.name >= other
+ def __call__(self, protoVer):
+ return api.protoModules[protoVer].OctetString(
+ 'PySNMP example command responder'
+ )
+
+class Uptime:
+ name = (1,3,6,1,2,1,1,3,0)
+ birthday = time.time()
+ def __eq__(self, other): return self.name == other
+ def __ne__(self, other): return self.name != other
+ def __lt__(self, other): return self.name < other
+ def __le__(self, other): return self.name <= other
+ def __gt__(self, other): return self.name > other
+ def __ge__(self, other): return self.name >= other
+ def __call__(self, protoVer):
+ return api.protoModules[protoVer].TimeTicks(
+ (time.time()-self.birthday)*100
+ )
+
+mibInstr = (
+ SysDescr(), Uptime() # sorted by object name
+ )
+
+mibInstrIdx = {}
+for mibVar in mibInstr:
+ mibInstrIdx[mibVar.name] = mibVar
+
+def cbFun(transportDispatcher, transportDomain, transportAddress, wholeMsg):
+ while wholeMsg:
+ msgVer = api.decodeMessageVersion(wholeMsg)
+ if msgVer in api.protoModules:
+ pMod = api.protoModules[msgVer]
+ else:
+ print('Unsupported SNMP version %s' % msgVer)
+ return
+ reqMsg, wholeMsg = decoder.decode(
+ wholeMsg, asn1Spec=pMod.Message(),
+ )
+ rspMsg = pMod.apiMessage.getResponse(reqMsg)
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ reqPDU = pMod.apiMessage.getPDU(reqMsg)
+ varBinds = []; pendingErrors = []
+ errorIndex = 0
+ # GETNEXT PDU
+ if reqPDU.isSameTypeWith(pMod.GetNextRequestPDU()):
+ # Produce response var-binds
+ for oid, val in pMod.apiPDU.getVarBinds(reqPDU):
+ errorIndex = errorIndex + 1
+ # Search next OID to report
+ nextIdx = bisect.bisect(mibInstr, oid)
+ if nextIdx == len(mibInstr):
+ # Out of MIB
+ varBinds.append((oid, val))
+ pendingErrors.append(
+ (pMod.apiPDU.setEndOfMibError, errorIndex)
+ )
+ else:
+ # Report value if OID is found
+ varBinds.append(
+ (mibInstr[nextIdx].name, mibInstr[nextIdx](msgVer))
+ )
+ elif reqPDU.isSameTypeWith(pMod.GetRequestPDU()):
+ for oid, val in pMod.apiPDU.getVarBinds(reqPDU):
+ if oid in mibInstrIdx:
+ varBinds.append((oid, mibInstrIdx[oid](msgVer)))
+ else:
+ # No such instance
+ varBinds.append((oid, val))
+ pendingErrors.append(
+ (pMod.apiPDU.setNoSuchInstanceError, errorIndex)
+ )
+ break
+ else:
+ # Report unsupported request type
+ pMod.apiPDU.setErrorStatus(rspPDU, 'genErr')
+ pMod.apiPDU.setVarBinds(rspPDU, varBinds)
+ # Commit possible error indices to response PDU
+ for f, i in pendingErrors:
+ f(rspPDU, i)
+ transportDispatcher.sendMessage(
+ encoder.encode(rspMsg), transportDomain, transportAddress
+ )
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+transportDispatcher.registerRecvCbFun(cbFun)
+
+# UDP/IPv4
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openServerMode(('localhost', 161))
+)
+
+# UDP/IPv6
+transportDispatcher.registerTransport(
+ udp6.domainName, udp6.Udp6SocketTransport().openServerMode(('::1', 161))
+)
+
+## Local domain socket
+#transportDispatcher.registerTransport(
+# unix.domainName, unix.UnixSocketTransport().openServerMode('/tmp/snmp-agent')
+#)
+
+transportDispatcher.jobStarted(1)
+
+try:
+ # Dispatcher will never finish as job#1 never reaches zero
+ transportDispatcher.runDispatcher()
+except:
+ transportDispatcher.closeDispatcher()
+ raise
diff --git a/examples/v1arch/asyncore/agent/ntforg/trap-v1-multiple-transports.py b/examples/v1arch/asyncore/agent/ntforg/trap-v1-multiple-transports.py
new file mode 100644
index 0000000..e0be25a
--- /dev/null
+++ b/examples/v1arch/asyncore/agent/ntforg/trap-v1-multiple-transports.py
@@ -0,0 +1,78 @@
+"""
+TRAP over multiple transports
++++++++++++++++++++++++++++++
+
+The following script sends two SNMP TRAP notification using the
+following options:
+
+* with SNMPv1
+* with community name 'public'
+* over IPv4/UDP and IPv6/UDP
+* send TRAP notification
+* to a Manager at 127.0.0.1:162 and [::1]
+* with TRAP ID 'coldStart' specified as an OID
+* include managed objects information:
+* with default Uptime value
+* with default Agent Address with '127.0.0.1'
+* overriding Enterprise OID with 1.3.6.1.4.1.20408.4.1.1.2
+
+The following Net-SNMP commands will produce similar SNMP notification:
+
+| $ snmptrap -v1 -c public udp:127.0.0.1 1.3.6.1.4.1.20408.4.1.1.2 127.0.0.1 1 0 12345
+| $ snmptrap -v1 -c public udp6:[::1] 1.3.6.1.4.1.20408.4.1.1.2 127.0.0.1 1 0 12345
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp, udp6, unix
+from pyasn1.codec.ber import encoder
+from pysnmp.proto import api
+
+# Protocol version to use
+pMod = api.protoModules[api.protoVersion1]
+#pMod = api.protoModules[api.protoVersion2c]
+
+# Build PDU
+trapPDU = pMod.TrapPDU()
+pMod.apiTrapPDU.setDefaults(trapPDU)
+
+# Traps have quite different semantics across proto versions
+if pMod == api.protoModules[api.protoVersion1]:
+ pMod.apiTrapPDU.setEnterprise(trapPDU, (1,3,6,1,1,2,3,4,1))
+ pMod.apiTrapPDU.setGenericTrap(trapPDU, 'coldStart')
+
+# Build message
+trapMsg = pMod.Message()
+pMod.apiMessage.setDefaults(trapMsg)
+pMod.apiMessage.setCommunity(trapMsg, 'public')
+pMod.apiMessage.setPDU(trapMsg, trapPDU)
+
+transportDispatcher = AsyncoreDispatcher()
+
+# UDP/IPv4
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openClientMode()
+ )
+transportDispatcher.sendMessage(
+ encoder.encode(trapMsg), udp.domainName, ('localhost', 162)
+ )
+
+# UDP/IPv6
+transportDispatcher.registerTransport(
+ udp6.domainName, udp6.Udp6SocketTransport().openClientMode()
+)
+transportDispatcher.sendMessage(
+ encoder.encode(trapMsg), udp6.domainName, ('::1', 162)
+)
+
+## Local domain socket
+#transportDispatcher.registerTransport(
+# unix.domainName, unix.UnixSocketTransport().openClientMode()
+#)
+#transportDispatcher.sendMessage(
+# encoder.encode(trapMsg), unix.domainName, '/tmp/snmp-manager'
+#)
+
+# Dispatcher will finish as all scheduled messages are sent
+transportDispatcher.runDispatcher()
+
+transportDispatcher.closeDispatcher()
diff --git a/examples/v1arch/asyncore/manager/cmdgen/get-v1.py b/examples/v1arch/asyncore/manager/cmdgen/get-v1.py
new file mode 100644
index 0000000..e7fa029
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/get-v1.py
@@ -0,0 +1,105 @@
+"""
+Fetch scalar MIB variables (SNMPv1)
++++++++++++++++++++++++++++++++++++
+
+Perform SNMP GET operation with the following options:
+
+* with SNMPv1, community 'public'
+* over IPv4/UDP
+* to an Agent at 195.218.195.228:161
+* for OIDs in tuple form
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpget -v1 -c public -ObentU 195.218.195.228 1.3.6.1.2.1.1.1.0 1.3.6.1.2.1.1.3.0
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp, udp6, unix
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto import api
+from time import time
+
+# Protocol version to use
+pMod = api.protoModules[api.protoVersion1]
+#pMod = api.protoModules[api.protoVersion2c]
+
+# Build PDU
+reqPDU = pMod.GetRequestPDU()
+pMod.apiPDU.setDefaults(reqPDU)
+pMod.apiPDU.setVarBinds(
+ reqPDU, ( ('1.3.6.1.2.1.1.1.0', pMod.Null('')),
+ ('1.3.6.1.2.1.1.3.0', pMod.Null('')) )
+ )
+
+# Build message
+reqMsg = pMod.Message()
+pMod.apiMessage.setDefaults(reqMsg)
+pMod.apiMessage.setCommunity(reqMsg, 'public')
+pMod.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > 3:
+ raise Exception("Request timed out")
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ # Match response to request
+ if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
+ # Check for SNMP errors reported
+ errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
+ if errorStatus:
+ print(errorStatus.prettyPrint())
+ else:
+ for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+ transportDispatcher.jobFinished(1)
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+# UDP/IPv4
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openClientMode()
+)
+
+# Pass message to dispatcher
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, ('demo.snmplabs.com', 161)
+)
+transportDispatcher.jobStarted(1)
+
+## UDP/IPv6 (second copy of the same PDU will be sent)
+transportDispatcher.registerTransport(
+ udp6.domainName, udp6.Udp6SocketTransport().openClientMode()
+)
+
+# Pass message to dispatcher
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp6.domainName, ('::1', 161)
+)
+transportDispatcher.jobStarted(1)
+
+## Local domain socket
+#transportDispatcher.registerTransport(
+# unix.domainName, unix.UnixSocketTransport().openClientMode()
+#)
+#
+# Pass message to dispatcher
+#transportDispatcher.sendMessage(
+# encoder.encode(reqMsg), unix.domainName, '/tmp/snmp-agent'
+#)
+#transportDispatcher.jobStarted(1)
+
+# Dispatcher will finish as job#1 counter reaches zero
+transportDispatcher.runDispatcher()
+
+transportDispatcher.closeDispatcher()
diff --git a/examples/v1arch/asyncore/manager/cmdgen/get-v2c-spoof-source-address.py b/examples/v1arch/asyncore/manager/cmdgen/get-v2c-spoof-source-address.py
new file mode 100644
index 0000000..16009ab
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/get-v2c-spoof-source-address.py
@@ -0,0 +1,117 @@
+"""
+Spoof IPv4 source address
++++++++++++++++++++++++++
+
+Send SNMP GET request from a non-local IP address:
+
+* with SNMPv2c, community 'public'
+* over IPv4/UDP
+* to an Agent at 195.218.195.228:161
+* from a non-local, spoofed IP 1.2.3.4 (root and Python 3.3+ required)
+* for OIDs in string form
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpget -v2c -c public -ObentU 195.218.195.228 1.3.6.1.2.1.1.1.0 1.3.6.1.2.1.1.3.0
+
+But unlike the above command, this script issues SNMP request from a
+non-default, non-local IP address.
+
+It is indeed possible to originate SNMP traffic from any valid local IP
+addresses. It could be a secondary IP interface, for instance. Superuser
+privileges are only required to send spoofed packets. Alternatively,
+sending from local interface could also be achieved by binding to
+it (via openClientMode() parameter).
+
+Agent would respond to the IP address you used as a source. So this script
+could only get a response if that source address is somehow routed to the
+host this script is running on. Otherwise it just times out.
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp
+from pysnmp.proto import api
+from pyasn1.codec.ber import encoder, decoder
+from time import time
+
+# Send request message to this address
+transportAddress = udp.UdpTransportAddress(('195.218.195.228', 161))
+
+# Send request message from this non-local (!) IP address
+transportAddress.setLocalAddress(('1.2.3.4', 0))
+
+# Protocol version to use
+#pMod = api.protoModules[api.protoVersion1]
+pMod = api.protoModules[api.protoVersion2c]
+
+# Build PDU
+reqPDU = pMod.GetRequestPDU()
+pMod.apiPDU.setDefaults(reqPDU)
+pMod.apiPDU.setVarBinds(
+ reqPDU, ( ('1.3.6.1.2.1.1.1.0', pMod.Null('')),
+ ('1.3.6.1.2.1.1.3.0', pMod.Null('')) )
+ )
+
+# Build message
+reqMsg = pMod.Message()
+pMod.apiMessage.setDefaults(reqMsg)
+pMod.apiMessage.setCommunity(reqMsg, 'public')
+pMod.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+class StopWaiting(Exception): pass
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > 3:
+ raise StopWaiting()
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ # Match response to request
+ if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
+ # Check for SNMP errors reported
+ errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
+ if errorStatus:
+ print(errorStatus.prettyPrint())
+ else:
+ for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+ transportDispatcher.jobFinished(1)
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+# Initialize UDP/IPv4 transport
+udpSocketTransport = udp.UdpSocketTransport().openClientMode()
+
+# Use sendmsg()/recvmsg() for socket communication (required for
+# IP source spoofing functionality)
+udpSocketTransport.enablePktInfo()
+
+# Enable IP source spoofing (requires root privileges)
+udpSocketTransport.enableTransparent()
+
+transportDispatcher.registerTransport(udp.domainName, udpSocketTransport)
+
+# Pass message to dispatcher
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, transportAddress
+)
+
+# We might never receive any response as we sent request with fake source IP
+transportDispatcher.jobStarted(1)
+
+# Dispatcher will finish as all jobs counter reaches zero
+try:
+ transportDispatcher.runDispatcher()
+except StopWaiting:
+ transportDispatcher.closeDispatcher()
+else:
+ raise
diff --git a/examples/v1arch/asyncore/manager/cmdgen/get-v2c-udp-broadcast-agent-discovery.py b/examples/v1arch/asyncore/manager/cmdgen/get-v2c-udp-broadcast-agent-discovery.py
new file mode 100644
index 0000000..cd8c3c9
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/get-v2c-udp-broadcast-agent-discovery.py
@@ -0,0 +1,99 @@
+"""
+Broadcast SNMP message (IPv4)
++++++++++++++++++++++++++++++
+
+Send SNMP GET request to broadcast address and wait for respons(es):
+
+* with SNMPv2c, community 'public'
+* over IPv4/UDP
+* to all Agents via broadcast address 255.255.255.255:161
+* for OIDs in tuple form
+
+Here we send out a single SNMP request and wait for potentially many SNMP
+responses from multiple SNMP Agents listening in local broadcast domain.
+Since we can't predict the exact number of Agents responding, this script
+just waits for arbitrary time for collecting all responses. This technology
+is also known as SNMP-based discovery.
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpget -v2c -c public -ObentU 255.255.255.255 1.3.6.1.2.1.1.1.0 1.3.6.1.2.1.1.3.0
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto import api
+from time import time
+
+# Broadcast manager settings
+maxWaitForResponses = 5
+maxNumberResponses = 10
+
+# Protocol version to use
+#pMod = api.protoModules[api.protoVersion1]
+pMod = api.protoModules[api.protoVersion2c]
+
+# Build PDU
+reqPDU = pMod.GetRequestPDU()
+pMod.apiPDU.setDefaults(reqPDU)
+pMod.apiPDU.setVarBinds(
+ reqPDU, ( ('1.3.6.1.2.1.1.1.0', pMod.Null('')),
+ ('1.3.6.1.2.1.1.3.0', pMod.Null('')) )
+ )
+
+# Build message
+reqMsg = pMod.Message()
+pMod.apiMessage.setDefaults(reqMsg)
+pMod.apiMessage.setCommunity(reqMsg, 'public')
+pMod.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+class StopWaiting(Exception): pass
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > maxWaitForResponses:
+ raise StopWaiting()
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ # Match response to request
+ if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
+ # Check for SNMP errors reported
+ errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
+ if errorStatus:
+ print(errorStatus.prettyPrint())
+ else:
+ for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+ transportDispatcher.jobFinished(1)
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+# UDP/IPv4
+udpSocketTransport = udp.UdpSocketTransport().openClientMode().enableBroadcast()
+transportDispatcher.registerTransport(udp.domainName, udpSocketTransport)
+
+# Pass message to dispatcher
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, ('255.255.255.255', 161)
+)
+
+# wait for a maximum of 10 responses or time out
+transportDispatcher.jobStarted(1, maxNumberResponses)
+
+# Dispatcher will finish as all jobs counter reaches zero
+try:
+ transportDispatcher.runDispatcher()
+except StopWaiting:
+ transportDispatcher.closeDispatcher()
+else:
+ raise
diff --git a/examples/v1arch/asyncore/manager/cmdgen/getbulk-v2c.py b/examples/v1arch/asyncore/manager/cmdgen/getbulk-v2c.py
new file mode 100644
index 0000000..8ba9af6
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/getbulk-v2c.py
@@ -0,0 +1,112 @@
+"""
+Bulk walk Agent MIB (SNMPv2c)
++++++++++++++++++++++++++++++
+
+Perform SNMP GETBULK operation with the following options:
+
+* with SNMPv2c, community 'public'
+* over IPv4/UDP
+* to an Agent at 195.218.195.228:161
+* for OID in tuple form
+* with non-repeaters=0 and max-repeaters=25
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpbulkwalk -v2c -c public -ObentU -Cn0 -Cr25 195.218.195.228 1.3.6
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto.api import v2c
+from time import time
+
+# SNMP table header
+headVars = [ v2c.ObjectIdentifier((1,3,6)) ]
+
+# Build PDU
+reqPDU = v2c.GetBulkRequestPDU()
+v2c.apiBulkPDU.setDefaults(reqPDU)
+v2c.apiBulkPDU.setNonRepeaters(reqPDU, 0)
+v2c.apiBulkPDU.setMaxRepetitions(reqPDU, 25)
+v2c.apiBulkPDU.setVarBinds(reqPDU, [ (x, v2c.null) for x in headVars ])
+
+# Build message
+reqMsg = v2c.Message()
+v2c.apiMessage.setDefaults(reqMsg)
+v2c.apiMessage.setCommunity(reqMsg, 'public')
+v2c.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > 3:
+ raise Exception("Request timed out")
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU, headVars=headVars):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=v2c.Message())
+
+ rspPDU = v2c.apiMessage.getPDU(rspMsg)
+
+ # Match response to request
+ if v2c.apiBulkPDU.getRequestID(reqPDU)==v2c.apiBulkPDU.getRequestID(rspPDU):
+ # Format var-binds table
+ varBindTable = v2c.apiBulkPDU.getVarBindTable(reqPDU, rspPDU)
+
+ # Check for SNMP errors reported
+ errorStatus = v2c.apiBulkPDU.getErrorStatus(rspPDU)
+ if errorStatus and errorStatus != 2:
+ errorIndex = v2c.apiBulkPDU.getErrorIndex(rspPDU)
+ print('%s at %s' % (errorStatus.prettyPrint(),
+ errorIndex and varBindTable[int(errorIndex)-1] or '?'))
+ transportDispatcher.jobFinished(1)
+ break
+
+ # Report SNMP table
+ for tableRow in varBindTable:
+ for name, val in tableRow:
+ print('from: %s, %s = %s' % (
+ transportAddress, name.prettyPrint(), val.prettyPrint()
+ )
+ )
+
+ # Stop on EOM
+ for oid, val in varBindTable[-1]:
+ if not isinstance(val, v2c.Null):
+ break
+ else:
+ transportDispatcher.jobFinished(1)
+
+ # Generate request for next row
+ v2c.apiBulkPDU.setVarBinds(
+ reqPDU, [ (x, v2c.null) for x,y in varBindTable[-1] ]
+ )
+ v2c.apiBulkPDU.setRequestID(reqPDU, v2c.getNextRequestID())
+ transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), transportDomain, transportAddress
+ )
+ global startedAt
+ if time() - startedAt > 3:
+ raise Exception('Request timed out')
+ startedAt = time()
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openClientMode()
+)
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, ('demo.snmplabs.com', 161)
+)
+transportDispatcher.jobStarted(1)
+
+# Dispatcher will finish as job#1 counter reaches zero
+transportDispatcher.runDispatcher()
+
+transportDispatcher.closeDispatcher()
diff --git a/examples/v1arch/asyncore/manager/cmdgen/getnext-v1.py b/examples/v1arch/asyncore/manager/cmdgen/getnext-v1.py
new file mode 100644
index 0000000..b3e16f1
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/getnext-v1.py
@@ -0,0 +1,103 @@
+"""
+Walk Agent MIB (SNMPv1)
++++++++++++++++++++++++
+
+Perform SNMP GETNEXT operation with the following options:
+
+* with SNMPv1, community 'public'
+* over IPv4/UDP
+* to an Agent at 195.218.195.228:161
+* for OID in tuple form
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpwalk -v1 -c public -ObentU 195.218.195.228 1.3.6
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto import api
+from time import time
+
+# Protocol version to use
+pMod = api.protoModules[api.protoVersion1]
+#pMod = api.protoModules[api.protoVersion2c]
+
+# SNMP table header
+headVars = [ pMod.ObjectIdentifier((1,3,6)) ]
+
+# Build PDU
+reqPDU = pMod.GetNextRequestPDU()
+pMod.apiPDU.setDefaults(reqPDU)
+pMod.apiPDU.setVarBinds(reqPDU, [ (x, pMod.null) for x in headVars ])
+
+# Build message
+reqMsg = pMod.Message()
+pMod.apiMessage.setDefaults(reqMsg)
+pMod.apiMessage.setCommunity(reqMsg, 'public')
+pMod.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > 3:
+ raise Exception("Request timed out")
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU, headVars=headVars):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ # Match response to request
+ if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
+ # Check for SNMP errors reported
+ errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
+ if errorStatus and errorStatus != 2:
+ raise Exception(errorStatus)
+ # Format var-binds table
+ varBindTable = pMod.apiPDU.getVarBindTable(reqPDU, rspPDU)
+ # Report SNMP table
+ for tableRow in varBindTable:
+ for name, val in tableRow:
+ print('from: %s, %s = %s' % (
+ transportAddress, name.prettyPrint(), val.prettyPrint()
+ )
+ )
+ # Stop on EOM
+ for oid, val in varBindTable[-1]:
+ if not isinstance(val, pMod.Null):
+ break
+ else:
+ transportDispatcher.jobFinished(1)
+
+ # Generate request for next row
+ pMod.apiPDU.setVarBinds(
+ reqPDU, [ (x, pMod.null) for x,y in varBindTable[-1] ]
+ )
+ pMod.apiPDU.setRequestID(reqPDU, pMod.getNextRequestID())
+ transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), transportDomain, transportAddress
+ )
+ global startedAt
+ if time() - startedAt > 3:
+ raise Exception('Request timed out')
+ startedAt = time()
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openClientMode()
+)
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, ('demo.snmplabs.com', 161)
+)
+transportDispatcher.jobStarted(1)
+
+transportDispatcher.runDispatcher()
+
+transportDispatcher.closeDispatcher()
diff --git a/examples/v1arch/asyncore/manager/cmdgen/set-v2c.py b/examples/v1arch/asyncore/manager/cmdgen/set-v2c.py
new file mode 100644
index 0000000..a30bebb
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/cmdgen/set-v2c.py
@@ -0,0 +1,85 @@
+"""
+SET string and integer scalars (SNMPv2c)
+++++++++++++++++++++++++++++++++++++++++
+
+Perform SNMP SET operation with the following options:
+
+* with SNMPv2c, community 'public'
+* over IPv4/UDP
+* to an Agent at 195.218.195.228:161
+* for OIDs in string form and values in form of pyasn1 objects
+
+This script performs similar to the following Net-SNMP command:
+
+| $ snmpset -v2c -c public -ObentU 195.218.195.228 1.3.6.1.2.1.1.9.1.3.1 s 'New description' 1.3.6.1.2.1.1.9.1.4.1 t 12
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp
+from pyasn1.codec.ber import encoder, decoder
+from pysnmp.proto import api
+from time import time
+
+# Protocol version to use
+#pMod = api.protoModules[api.protoVersion1]
+pMod = api.protoModules[api.protoVersion2c]
+
+# Build PDU
+reqPDU = pMod.SetRequestPDU()
+pMod.apiPDU.setDefaults(reqPDU)
+pMod.apiPDU.setVarBinds(
+ reqPDU,
+ # A list of Var-Binds to SET
+ ( ('1.3.6.1.2.1.1.9.1.3.1', pMod.OctetString('New system description')),
+ ('1.3.6.1.2.1.1.9.1.4.1', pMod.TimeTicks(12)) )
+ )
+
+# Build message
+reqMsg = pMod.Message()
+pMod.apiMessage.setDefaults(reqMsg)
+pMod.apiMessage.setCommunity(reqMsg, 'public')
+pMod.apiMessage.setPDU(reqMsg, reqPDU)
+
+startedAt = time()
+
+def cbTimerFun(timeNow):
+ if timeNow - startedAt > 3:
+ raise Exception("Request timed out")
+
+def cbRecvFun(transportDispatcher, transportDomain, transportAddress,
+ wholeMsg, reqPDU=reqPDU):
+ while wholeMsg:
+ rspMsg, wholeMsg = decoder.decode(wholeMsg, asn1Spec=pMod.Message())
+ rspPDU = pMod.apiMessage.getPDU(rspMsg)
+ # Match response to request
+ if pMod.apiPDU.getRequestID(reqPDU)==pMod.apiPDU.getRequestID(rspPDU):
+ # Check for SNMP errors reported
+ errorStatus = pMod.apiPDU.getErrorStatus(rspPDU)
+ if errorStatus:
+ print(errorStatus.prettyPrint())
+ else:
+ for oid, val in pMod.apiPDU.getVarBinds(rspPDU):
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+ transportDispatcher.jobFinished(1)
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbRecvFun)
+transportDispatcher.registerTimerCbFun(cbTimerFun)
+
+# UDP/IPv4
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openClientMode()
+)
+
+# Pass message to dispatcher
+transportDispatcher.sendMessage(
+ encoder.encode(reqMsg), udp.domainName, ('demo.snmplabs.com', 161)
+)
+transportDispatcher.jobStarted(1)
+
+# Dispatcher will finish as job#1 counter reaches zero
+transportDispatcher.runDispatcher()
+
+transportDispatcher.closeDispatcher()
diff --git a/examples/v1arch/asyncore/manager/ntfrcv/v1-multiple-transports.py b/examples/v1arch/asyncore/manager/ntfrcv/v1-multiple-transports.py
new file mode 100644
index 0000000..b9073cf
--- /dev/null
+++ b/examples/v1arch/asyncore/manager/ntfrcv/v1-multiple-transports.py
@@ -0,0 +1,100 @@
+"""
+Listen for notifications at IPv4 & IPv6 interfaces
+++++++++++++++++++++++++++++++++++++++++++++++++++
+
+Receive SNMP TRAP messages with the following options:
+
+* SNMPv1/SNMPv2c
+* with SNMP community "public"
+* over IPv4/UDP, listening at 127.0.0.1:162
+* over IPv6/UDP, listening at [::1]:162
+* print received data on stdout
+
+Either of the following Net-SNMP commands will send notifications to this
+receiver:
+
+| $ snmptrap -v1 -c public 127.0.0.1 1.3.6.1.4.1.20408.4.1.1.2 127.0.0.1 1 1 123 1.3.6.1.2.1.1.1.0 s test
+| $ snmptrap -v2c -c public udp6:[::1] 123 1.3.6.1.6.3.1.1.5.1 1.3.6.1.2.1.1.5.0 s test
+
+Notification Receiver below uses two different transports for communication
+with Notification Originators - UDP over IPv4 and UDP over IPv6.
+
+"""#
+from pysnmp.carrier.asyncore.dispatch import AsyncoreDispatcher
+from pysnmp.carrier.asyncore.dgram import udp, udp6, unix
+from pyasn1.codec.ber import decoder
+from pysnmp.proto import api
+
+def cbFun(transportDispatcher, transportDomain, transportAddress, wholeMsg):
+ while wholeMsg:
+ msgVer = int(api.decodeMessageVersion(wholeMsg))
+ if msgVer in api.protoModules:
+ pMod = api.protoModules[msgVer]
+ else:
+ print('Unsupported SNMP version %s' % msgVer)
+ return
+ reqMsg, wholeMsg = decoder.decode(
+ wholeMsg, asn1Spec=pMod.Message(),
+ )
+ print('Notification message from %s:%s: ' % (
+ transportDomain, transportAddress
+ )
+ )
+ reqPDU = pMod.apiMessage.getPDU(reqMsg)
+ if reqPDU.isSameTypeWith(pMod.TrapPDU()):
+ if msgVer == api.protoVersion1:
+ print('Enterprise: %s' % (
+ pMod.apiTrapPDU.getEnterprise(reqPDU).prettyPrint()
+ )
+ )
+ print('Agent Address: %s' % (
+ pMod.apiTrapPDU.getAgentAddr(reqPDU).prettyPrint()
+ )
+ )
+ print('Generic Trap: %s' % (
+ pMod.apiTrapPDU.getGenericTrap(reqPDU).prettyPrint()
+ )
+ )
+ print('Specific Trap: %s' % (
+ pMod.apiTrapPDU.getSpecificTrap(reqPDU).prettyPrint()
+ )
+ )
+ print('Uptime: %s' % (
+ pMod.apiTrapPDU.getTimeStamp(reqPDU).prettyPrint()
+ )
+ )
+ varBinds = pMod.apiTrapPDU.getVarBindList(reqPDU)
+ else:
+ varBinds = pMod.apiPDU.getVarBindList(reqPDU)
+ print('Var-binds:')
+ for oid, val in varBinds:
+ print('%s = %s' % (oid.prettyPrint(), val.prettyPrint()))
+ return wholeMsg
+
+transportDispatcher = AsyncoreDispatcher()
+
+transportDispatcher.registerRecvCbFun(cbFun)
+
+# UDP/IPv4
+transportDispatcher.registerTransport(
+ udp.domainName, udp.UdpSocketTransport().openServerMode(('localhost', 162))
+)
+
+# UDP/IPv6
+transportDispatcher.registerTransport(
+ udp6.domainName, udp6.Udp6SocketTransport().openServerMode(('::1', 162))
+)
+
+## Local domain socket
+#transportDispatcher.registerTransport(
+# unix.domainName, unix.UnixSocketTransport().openServerMode('/tmp/snmp-manager')
+#)
+
+transportDispatcher.jobStarted(1)
+
+try:
+ # Dispatcher will never finish as job#1 never reaches zero
+ transportDispatcher.runDispatcher()
+except:
+ transportDispatcher.closeDispatcher()
+ raise