Mapping SNMP Trap OIDs to Canonical Fault Codes #

The operational failure this page addresses is a ticket routed on a number no human recognizes. An SNMP trap identifies its fault by an object identifier — a dotted string like 1.3.6.1.4.1.9.9.171.2.0.1 — that means “IPsec tunnel down” to a Cisco device and nothing at all to a correlation engine that has never seen it. If the ingestion layer forwards the raw OID, every downstream rule has to carry vendor-specific branches, and the same physical fault raised by a Cisco, a Juniper, and a Nokia device produces three unrelated identifiers that never correlate into one incident. The fix is to translate each vendor OID into a canonical fault code at ingestion, so correlation reasons over TUNNEL_DOWN regardless of who raised it. A trap whose OID is not in the map must never be guessed into the live stream, and it must never be silently dropped — either failure inflates mean-time-to-resolution, one by mis-routing and one by losing a real fault. Deterministic OID mapping with an explicit dead-letter path for the unknown is what keeps this feed trustworthy.

Schema alignment and taxonomy anchor #

This page is the trap-specific specialization of Logparser Integration, the deterministic normalization stage within the Ingestion & Parsing Workflows pipeline. That parent stage owns the priority cascade and the one-in-one-out contract; this page owns the narrower job of turning a decoded trap PDU — a trap OID plus its varbind list — into the canonical fields defined by Event Schema Design. The transport that receives and authenticates the trap is covered in Configuring SNMPv3 Trap Receivers in Python; this page picks up once a validated PDU is in hand. It sits alongside its sibling procedures Parsing RFC 5424 Syslog Structured Data and Parsing NETCONF YANG-Push Diff Notifications, each normalizing a different source format onto the same contract.

How a trap OID resolves to a fault code #

An SNMP trap PDU carries a snmpTrapOID varbind that names the notification type, followed by a list of object varbinds that carry the fault’s parameters — an interface index, an admin state, an alarm string. The enterprise arc of the OID identifies the vendor through an IANA private enterprise number: 1.3.6.1.4.1.9 is Cisco, 1.3.6.1.4.1.2636 is Juniper. Mapping is a two-step resolve: the trap OID selects a canonical fault code from a MIB-backed table, and named varbind OIDs select the parameters that populate the event. Both lookups are exact-match dictionary lookups warmed at load, so the hot path is O(1) with no per-trap MIB compilation.

Diagram: a trap OID resolves through a MIB-backed table to a canonical fault code while named varbinds populate the event parameters, and an unknown OID branches to the dead-letter queue.

Mapping trap OIDs and extracting varbinds #

The implementation below resolves a decoded trap PDU into a canonical event. The OID-to-fault-code table and the varbind-name table are loaded once from compiled MIB exports and cached; the canonical fault code is a Pydantic-validated enum so an out-of-band value cannot leak into the stream. An unknown trap OID raises, and the caller dead-letters it rather than emitting a guessed code.

import hashlib
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator


class FaultCode(str, Enum):
    """Canonical, vendor-neutral fault codes correlation reasons over."""
    LINK_DOWN = "LINK_DOWN"
    TUNNEL_DOWN = "TUNNEL_DOWN"
    BGP_BACKWARD_TRANSITION = "BGP_BACKWARD_TRANSITION"
    POWER_SUPPLY_FAIL = "POWER_SUPPLY_FAIL"
    FAN_FAIL = "FAN_FAIL"


# MIB-backed, warmed at load: exact trap OID -> (fault code, baseline severity).
# Enterprise arcs identify the vendor via the IANA private enterprise number.
OID_FAULT_MAP: dict[str, tuple[FaultCode, int]] = {
    "1.3.6.1.6.3.1.1.5.3": (FaultCode.LINK_DOWN, 3),                # IF-MIB linkDown
    "1.3.6.1.4.1.9.9.171.2.0.1": (FaultCode.TUNNEL_DOWN, 2),        # Cisco IPsec
    "1.3.6.1.2.1.15.7.0.2": (FaultCode.BGP_BACKWARD_TRANSITION, 2), # BGP4-MIB
    "1.3.6.1.4.1.2636.4.1.4": (FaultCode.POWER_SUPPLY_FAIL, 1),     # Juniper PSU
}

# Named object OIDs whose values become event parameters.
VARBIND_NAMES: dict[str, str] = {
    "1.3.6.1.2.1.2.2.1.1": "if_index",
    "1.3.6.1.2.1.2.2.1.7": "if_admin_status",
    "1.3.6.1.2.1.2.2.1.2": "if_descr",
}


class TrapEvent(BaseModel):
    """Canonical fault event derived from one SNMP trap PDU."""
    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    device_id: str
    fault_code: FaultCode
    severity: int = Field(ge=0, le=7)
    event_time: datetime
    parameters: dict[str, str] = Field(default_factory=dict)
    enterprise_number: Optional[int] = None
    raw_payload_hash: str = Field(min_length=64, max_length=64)

    @field_validator("event_time")
    @classmethod
    def require_utc(cls, value: datetime) -> datetime:
        if value.tzinfo is None:
            raise ValueError("event_time must be timezone-aware UTC")
        return value.astimezone(timezone.utc)


def enterprise_number(trap_oid: str) -> Optional[int]:
    """Extract the IANA private enterprise number from an enterprise arc."""
    prefix = "1.3.6.1.4.1."
    if not trap_oid.startswith(prefix):
        return None
    tail = trap_oid[len(prefix):].split(".", 1)[0]
    return int(tail) if tail.isdigit() else None


def map_trap(pdu: dict[str, Any]) -> TrapEvent:
    """Resolve a decoded trap PDU to a canonical event, or raise on unknown OID."""
    trap_oid = pdu["trap_oid"]
    resolved = OID_FAULT_MAP.get(trap_oid)
    if resolved is None:
        # Never guess an unknown trap into the live stream; the caller
        # dead-letters this so the taxonomy table gains an entry instead.
        raise KeyError(f"unmapped trap OID {trap_oid}")
    fault_code, severity = resolved
    params: dict[str, str] = {}
    for oid, value in pdu["varbinds"].items():
        # Match against the object arc, ignoring the trailing instance index.
        for base, name in VARBIND_NAMES.items():
            if oid == base or oid.startswith(base + "."):
                params[name] = str(value)
                break
    return TrapEvent(
        device_id=pdu["source"],
        fault_code=fault_code,
        severity=severity,
        event_time=pdu.get("received_at") or datetime.now(timezone.utc),
        parameters=params,
        enterprise_number=enterprise_number(trap_oid),
        raw_payload_hash=hashlib.sha256(repr(pdu).encode()).hexdigest(),
    )

Async ingestion hook #

Traps arrive in bursts — a chassis losing power emits one per line card in the same instant — so mapping must never block the receiver loop. The hook below drains a batch of decoded PDUs, maps each in isolation, and routes unknown OIDs to a dead-letter queue without stalling the consumer. Mapping is pure dictionary work and bounded per trap, so it stays inline; only the publish and the dead-letter put are awaited.

import asyncio
import logging
from typing import Any

logger = logging.getLogger("snmp_trap_map")


async def process_trap_batch(
    pdus: list[dict[str, Any]],
    publish,
    dlq: asyncio.Queue[dict[str, Any]],
) -> None:
    """Map a batch of decoded trap PDUs, isolating each unknown OID."""
    events = []
    for pdu in pdus:
        try:
            events.append(map_trap(pdu))
        except (KeyError, ValueError) as exc:
            await dlq.put({
                "trap_oid": pdu.get("trap_oid"),
                "source": pdu.get("source"),
                "reason": "unknown_oid" if isinstance(exc, KeyError) else "schema_reject",
                "detail": str(exc),
            })
            logger.warning("trap map failure", extra={"reason": str(exc)})
    if events:
        await publish(events)                     # single non-blocking bus write
    await asyncio.sleep(0)                         # yield to the event loop

Mitigation and hardening #

When an OID does not resolve, degrade deterministically. These are the concrete failure paths a production deployment should implement.

1. Unknown-OID dead-letter, never guess. A trap OID absent from the table is diverted to the dead-letter queue tagged unknown_oid, carrying the raw PDU. It is never mapped to a nearest-match code, because a wrong canonical code routes a wrong ticket. Alert when the unknown-OID rate exceeds 0.5% over five minutes — a spike is a firmware release adding traps the table has not caught up to.
2. Enum-cached fault codes. The canonical fault code is a validated enum, and the OID table is warmed once at load. Resolving a MIB per trap would add unbounded latency and pull file I/O onto the hot path; the cache keeps mapping O(1) even during a chassis-wide burst.
3. Enterprise-number provenance. Preserving the IANA private enterprise number, resolved from the enterprise arc against the IANA Private Enterprise Numbers registry, lets downstream logic distinguish a genuine Cisco trap from a spoofed one claiming a Cisco OID, echoing the posture of Security Boundary Mapping.
4. Varbind tolerance. A trap missing an expected object varbind still maps to its fault code with the parameters it does carry; the missing parameter is recorded, not treated as a parse failure, because losing a known fault over an absent optional field is worse than an incomplete parameter set.

Operational hardening notes #

Keep the per-trap cost flat under burst load. The two lookups — trap OID to fault code, object OID to parameter name — are exact-match dictionary operations warmed at load; never compile or walk a MIB on the hot path. Match varbind OIDs by object arc with a cheap prefix test so the trailing instance index does not defeat the lookup. Map at the batch level, a few hundred PDUs per drain, so the loop yields between batches. Held to these bounds, mapping holds p99 latency well under 10ms per trap and keeps the unknown-OID rate under 0.5%, which is the practical signal that MIB coverage is healthy. Integrated this way, every trap reaching correlation carries a vendor-neutral fault code, so the same physical fault from a Cisco, a Juniper, and a Nokia device collapses into one incident instead of three — the reduction that keeps this feed from inflating MTTR. The async receiver pattern that feeds these PDUs follows the non-blocking model in the official Python asyncio documentation.

Frequently Asked Questions #

Why translate trap OIDs to canonical fault codes at ingestion? #

Because a raw trap OID is vendor-specific: the same physical fault raised by a Cisco, a Juniper, and a Nokia device carries three unrelated OIDs that never correlate into one incident. Translating each to a shared canonical code like TUNNEL_DOWN at ingestion means the correlation tier reasons over one identifier regardless of vendor, and downstream rules no longer need per-vendor branches. It is what lets three symptoms of one outage collapse into a single ticket.

What happens to a trap whose OID is not in the table? #

It is diverted to the dead-letter queue tagged unknown OID with its raw PDU, never guessed into a nearest-match code and never silently dropped. Guessing routes a wrong ticket and dropping loses a real fault, and both inflate mean time to resolution. An unknown-OID rate above 0.5 percent over five minutes is alertable and usually means a firmware release added traps the mapping table has not caught up to.

Do I need to compile MIBs on every trap? #

No, and doing so would be a serious performance mistake. The OID-to-fault-code table and the varbind-name table are compiled from MIB exports once and warmed into memory at load, so each trap resolves through O(1) dictionary lookups with no file I/O on the hot path. This keeps mapping latency under ten milliseconds per trap even during a chassis-wide burst where every line card traps at once.

How does the enterprise number help with spoofed traps? #

The enterprise number is extracted from the enterprise arc of the trap OID and identifies the vendor through the IANA registry. Preserving it lets downstream logic check that a trap claiming a Cisco OID actually originates from an authorized Cisco source, so a spoofed trap carrying a borrowed OID can be caught at the security boundary rather than trusted into correlation.

Related #

• Up to the parent reference: Logparser Integration
• The trap receiver that feeds these PDUs: Configuring SNMPv3 Trap Receivers in Python
• Sibling source normalizer: Parsing RFC 5424 Syslog Structured Data
• Sibling source normalizer: Parsing NETCONF YANG-Push Diff Notifications
• The event contract this mapper emits: Event Schema Design