1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
"""
Report SNMP engine processing details
+++++++++++++++++++++++++++++++++++++
Send SNMP GET request with the following options:
* with SNMPv3 with user 'usr-sha-aes', SHA auth and AES128 privacy protocols
* over IPv4/UDP
* to an Agent at 127.0.0.1:161
* for an OID in tuple form
* also registers its own execution observer to snmpEngine
While execution, this script will report some details on request processing
as seen by rfc3412.sendPdu() and rfc3412.receiveMessage() abstract interfaces.
This script performs similar to the following Net-SNMP command:
| $ snmpget -v3 -l authPriv -u usr-sha-aes -a SHA -A authkey1 -x AES -X privkey1 -ObentU 127.0.0.1:161 1.3.6.1.2.1.1.1.0
"""#
from pysnmp.entity import engine, config
from pysnmp.carrier.asyncore.dgram import udp
from pysnmp.entity.rfc3413 import cmdgen
# Create SNMP engine instance
snmpEngine = engine.SnmpEngine()
# Execution point observer setup
# Register a callback to be invoked at specified execution point of
# SNMP Engine and passed local variables at code point's local scope
# noinspection PyUnusedLocal,PyUnusedLocal
def requestObserver(snmpEngine, execpoint, variables, cbCtx):
print("Execution point: %s" % execpoint)
print(
"* transportDomain: %s"
% ".".join([str(x) for x in variables["transportDomain"]])
)
print(
"* transportAddress: %s"
% "@".join([str(x) for x in variables["transportAddress"]])
)
print("* securityModel: %s" % variables["securityModel"])
print("* securityName: %s" % variables["securityName"])
print("* securityLevel: %s" % variables["securityLevel"])
print(
"* contextEngineId: {}".format(
variables["contextEngineId"]
and variables["contextEngineId"].prettyPrint()
or "<empty>"
)
)
print("* contextName: %s" % variables["contextName"].prettyPrint())
print("* PDU: %s" % variables["pdu"].prettyPrint())
snmpEngine.observer.registerObserver(
requestObserver, "rfc3412.sendPdu", "rfc3412.receiveMessage:response"
)
#
# SNMPv3/USM setup
#
# user: usr-sha-aes, auth: SHA, priv AES
config.addV3User(
snmpEngine,
"usr-sha-aes",
config.usmHMACSHAAuthProtocol,
"authkey1",
config.usmAesCfb128Protocol,
"privkey1",
)
config.addTargetParams(snmpEngine, "my-creds", "usr-sha-aes", "authPriv")
#
# Setup transport endpoint and bind it with security settings yielding
# a target name
#
# UDP/IPv4
config.addTransport(
snmpEngine, udp.domainName, udp.UdpSocketTransport().openClientMode()
)
config.addTargetAddr(
snmpEngine, "my-router", udp.domainName, ("127.0.0.1", 161), "my-creds"
)
# Error/response receiver
# noinspection PyUnusedLocal,PyUnusedLocal,PyUnusedLocal
def cbFun(
snmpEngine,
sendRequestHandle,
errorIndication,
errorStatus,
errorIndex,
varBinds,
cbCtx,
):
if errorIndication:
print(errorIndication)
elif errorStatus:
print(
"{} at {}".format(
errorStatus.prettyPrint(),
errorIndex and varBinds[int(errorIndex) - 1][0] or "?",
)
)
else:
for oid, val in varBinds:
print(f"{oid.prettyPrint()} = {val.prettyPrint()}")
# Prepare and send a request message
cmdgen.GetCommandGenerator().sendVarBinds(
snmpEngine,
"my-router",
None,
"", # contextEngineId, contextName
[((1, 3, 6, 1, 2, 1, 1, 1, 0), None)],
cbFun,
)
# Run I/O dispatcher which would send pending queries and process responses
snmpEngine.transportDispatcher.runDispatcher()
snmpEngine.observer.unregisterObserver()
|
