代碼一：在本地測驗可以接收，上云主機收不到
```
from pysnmp.entity import engine, config
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.entity.rfc3413 import ntfrcv
import asyncio
# 需要Python 3.4及更高版本！
# Get the event loop for this thread
loop = asyncio.get_event_loop()
# Create SNMP engine with autogenernated engineID and pre-bound
# to socket transport dispatcher
snmpEngine = engine.SnmpEngine()
# Transport setup
# UDP over IPv4, first listening interface/port
config.addTransport(
snmpEngine,
udp.domainName + (1,),
udp.UdpTransport().openServerMode(('192.168.6.1', 162))
)
# UDP over IPv4, second listening interface/port
# config.addTransport(
# snmpEngine,
# udp.domainName + (2,),
# udp.UdpTransport().openServerMode(('127.0.0.1', 2162))
# )
config.addTransport(
snmpEngine,
udp.domainName + (2,),
udp.UdpTransport().openServerMode(('localhost', 2162))
)
config.addTransport(
snmpEngine,
udp.domainName + (3,),
udp.UdpTransport().openServerMode(('106.13.174.184', 162))
)
# SNMPv1/2c setup
# SecurityName <-> CommunityName mapping
config.addV1System(snmpEngine, 'my-area', 'public')
# Callback function for receiving notifications
# noinspection PyUnusedLocal
def cbFun(snmpEngine,
stateReference,
contextEngineId, contextName,
varBinds,
cbCtx):
transportDomain, transportAddress = snmpEngine.msgAndPduDsp.getTransportInfo(stateReference)
print('Notification from %s, SNMP Engine %s, Context %s' % (transportAddress,
contextEngineId.prettyPrint(),
contextName.prettyPrint()))
for name, val in varBinds:
print('%s = %s' % (name.prettyPrint(), val.prettyPrint()))
# Register SNMP Application at the SNMP engine
ntfrcv.NotificationReceiver(snmpEngine, cbFun)
# Run asyncio main loop
loop.run_forever()
```
代碼二：本地，部署到云主機都可以收到trap,怎么不用密碼的，不怕被攻擊吧，奇怪
```
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
# 只能接收本地的？
#
# noinspection PyUnusedLocal
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.getVarBinds(reqPDU)
else:
varBinds = pMod.apiPDU.getVarBinds(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(('192.168.6.1', 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
```
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/80778.html
上一篇：爬取郵箱
下一篇：求問一個python基礎問題