Forwarding traps¶
Forwarding notifications to other tools¶
Receive SNMP TRAP/INFORM messages and relay each one to any number of downstream tools, with the following options:
SNMPv1/SNMPv2c
with SNMP community “public”
over IPv4/UDP, listening at 127.0.0.1:162
relaying to 127.0.0.1:1162 (everything) and 127.0.0.1:1163 (link state notifications from 127.0.0.0/8 only)
using Asyncio framework for network transport
162 is where a relay listens, which is why it is what this example binds – and it is a privileged port, so run this as root or change LISTEN_AT below to something above 1024 and put the same port in the snmptrap commands.
A notification receiver hands the application the bindings an SMIv2 notification carries – an SNMPv1 trap arrives converted, per RFC 2576 section 3.1 – so forwarding is a matter of sending those same bindings on rather than building a notification from scratch. sysUpTime.0 and snmpTrapOID.0 stay at the front where RFC 3416 wants them, and the original agent’s uptime is relayed rather than this relay’s.
Two things are worth copying out of this example and into anything real.
Do not send from the receiving callback. It runs on the event loop, and a downstream tool that stops answering would otherwise stall the socket every other tool’s notifications arrive on. The callback only queues.
Give each destination its own queue. A bounded queue per destination is what makes one unreachable tool cost that tool its notifications and nothing else. A shared queue spreads the stall to everyone.
Either of the following Net-SNMP commands will send a notification into the relay:
To see what comes out the other side, run multiple-interfaces.py with one of its listening ports changed to 1162, which is where the first destination below sends.
import asyncio
import ipaddress
import logging
from pysnmp.carrier.asyncio.dgram import udp
from pysnmp.entity import config, engine
from pysnmp.entity.rfc3413 import ntfrcv
from pysnmp.hlapi.asyncio import (
CommunityData,
ContextData,
ObjectIdentity,
ObjectType,
SnmpEngine,
UdpTransportTarget,
send_notification,
)
from pysnmp.proto.api import v2c
# snmpTrapOID.0 names the notification; snmpTrapAddress.0 is where a relayed
# one records who actually sent it, since the downstream tool sees this relay
# as the sender.
SNMP_TRAP_OID = v2c.apiTrapPDU.snmpTrapOID
SNMP_TRAP_ADDRESS = v2c.apiTrapPDU.snmpTrapAddress
# linkDown and linkUp, the subset one of the destinations below asks for.
LINK_STATE = ("1.3.6.1.6.3.1.1.5.3", "1.3.6.1.6.3.1.1.5.4")
# Where notifications arrive. Binding 162 needs privilege; an unprivileged port
# works the same way, and the `snmptrap` commands above have to name it too.
LISTEN_AT = ("127.0.0.1", 162)
log = logging.getLogger("forwarder")
class Destination:
"""One downstream tool, the subset it wants, and the queue it reads from.
`match` is called with the source address and the bindings of each
notification and says whether this destination should see it. Anything
goes: the trap OID, the sending agent, the value of a binding.
"""
def __init__(
self, name, address, auth=None, notifyType="trap", match=None, queueSize=1000
):
self.name = name
self.address = address
self.auth = auth or CommunityData("public", mpModel=1)
self.notifyType = notifyType
self.match = match or (lambda sourceAddress, varBinds: True)
self.queue = asyncio.Queue(maxsize=queueSize)
self.sent = self.dropped = self.failed = 0
def trapOid(varBinds):
"""The notification's own OID, which is what a filter usually turns on."""
for oid, value in varBinds:
if oid == SNMP_TRAP_OID:
return str(value)
return ""
def fromNetwork(cidr):
"""Match notifications sent by an agent in `cidr`."""
network = ipaddress.ip_network(cidr)
def match(sourceAddress, varBinds):
return ipaddress.ip_address(sourceAddress) in network
return match
def oneOf(*oids):
"""Match notifications whose snmpTrapOID.0 is one of `oids`."""
def match(sourceAddress, varBinds):
return trapOid(varBinds) in oids
return match
def both(first, second):
"""Match what `first` and `second` both match."""
return lambda sourceAddress, varBinds: (
first(sourceAddress, varBinds) and second(sourceAddress, varBinds)
)
def relayed(sourceAddress, varBinds):
"""The bindings to send on, with the original sender recorded.
A downstream tool sees the relay's address as the sender, so
`snmpTrapAddress.0` is what tells it where the notification really came
from. An SNMPv1 trap already carries one -- holding the agent address the
sender put in the PDU, which is 0.0.0.0 from most stacks and wrong behind
NAT -- so that one is replaced where it says nothing and kept where it
does.
"""
forwarded = []
stamped = False
for oid, value in varBinds:
if oid == SNMP_TRAP_ADDRESS:
stamped = True
if value.prettyPrint() in ("0.0.0.0", ""):
value = v2c.IpAddress(sourceAddress)
forwarded.append(ObjectType(ObjectIdentity(oid), value))
if not stamped:
forwarded.append(
ObjectType(ObjectIdentity(SNMP_TRAP_ADDRESS), v2c.IpAddress(sourceAddress))
)
return forwarded
async def forward(snmpEngine, destination):
"""Send what lands in one destination's queue, for as long as it lands.
One task per destination, so a tool that stops answering delays nothing
but its own.
"""
transportTarget = UdpTransportTarget(destination.address, timeout=2, retries=1)
contextData = ContextData()
while True:
sourceAddress, varBinds = await destination.queue.get()
try:
errorIndication, errorStatus, errorIndex, _ = await send_notification(
snmpEngine,
destination.auth,
transportTarget,
contextData,
destination.notifyType,
relayed(sourceAddress, varBinds),
)
if errorIndication or errorStatus:
destination.failed += 1
log.warning("%s: %s", destination.name, errorIndication or errorStatus)
else:
destination.sent += 1
except Exception:
destination.failed += 1
log.exception("forwarding to %s failed", destination.name)
finally:
destination.queue.task_done()
async def main():
"""Listen, and relay what arrives to the destinations that asked for it."""
destinations = [
Destination("archive", ("127.0.0.1", 1162)),
Destination(
"netops",
("127.0.0.1", 1163),
match=both(fromNetwork("127.0.0.0/8"), oneOf(*LINK_STATE)),
),
]
# Two engines: one bound to the listening socket, one sending. Keeping them
# apart keeps the credentials this relay accepts separate from the ones it
# sends under, and gives the relayed notifications this relay's own engine
# ID rather than the receiving side's.
receivingEngine = engine.SnmpEngine()
sendingEngine = SnmpEngine()
config.addTransport(
receivingEngine,
udp.domainName + (1,),
udp.UdpTransport().openServerMode(LISTEN_AT),
)
config.addV1System(receivingEngine, "my-area", "public")
# noinspection PyUnusedLocal
def cbFun(
snmpEngine, stateReference, contextEngineId, contextName, varBinds, cbCtx
):
# The sending agent's address is not in the PDU; the dispatcher has it.
transportDomain, transportAddress = snmpEngine.msgAndPduDsp.getTransportInfo(
stateReference
)
notification = (transportAddress[0], tuple(varBinds))
for destination in destinations:
if not destination.match(notification[0], notification[1]):
continue
try:
destination.queue.put_nowait(notification)
except asyncio.QueueFull:
# The destination is not keeping up. Shedding its load here is
# what stops it from becoming everyone's problem.
destination.dropped += 1
log.warning("%s is not keeping up, dropped one", destination.name)
ntfrcv.NotificationReceiver(receivingEngine, cbFun)
await asyncio.gather(
*(forward(sendingEngine, destination) for destination in destinations)
)
asyncio.run(main())
Download script.
See also: library reference.