Fault Correlation Ticket Routing Automation

Validating NetFlow Events with Pydantic

Raw NetFlow v9 and IPFIX telemetry is the operational backbone for traffic baselining, capacity forecasting, and anomaly detection in carrier-grade networks. The problem is that raw flow records routinely arrive with malformed headers, truncated payloads, vendor-specific field deviations, and out-of-range ASNs that slip straight past naive dict parsing. When a malformed record reaches the correlation engine, it triggers a silent drop or a half-completed graph traversal, cascades into false-positive ticket routing, and directly inflates mean time to resolution (MTTR) — a single misconfigured collector can add 40–60% to MTTR through retries and spurious escalations. Enforcing strict, schema-driven validation at the ingestion boundary turns those silent failures into deterministic, measurable rejections that NOC engineers and automation pipelines can act on immediately.

Schema alignment and taxonomy anchor

This page is the flow-record application of the contract defined in Event Schema Design, the parent stage of the Core Architecture & Log Taxonomy reference pipeline. That parent stage owns the canonical NetworkEvent shape shared by every protocol; this page owns the narrower job of proving a NetFlow record is well-formed before it is mapped onto that canonical event. Rather than dictionary lookups, ad-hoc casting, or fragile regex chains, a Pydantic model enforces structural integrity at the edge so that routing fields — src_as, dst_as, input_snmp, output_snmp, protocol, and tcp_flags — are validated against known ranges and enumerated types.

Anchoring ingestion rules to that documented schema keeps NetFlow streams interoperable with the SNMP Trap Standardization and Syslog Format Parsing pipelines without introducing schema drift. Every flow record entering the correlation queue carries the same guarantees as a normalized trap or an RFC 5424 syslog line.

NetFlow validation gate and failure routingA raw NetFlow v9 or IPFIX record dictionary feeds a Pydantic strict-mode validator configured with extra="forbid". A schema-validity gate routes valid records to the correlation engine as a canonical NetworkEvent, while invalid records divert to a dead-letter queue carrying a field-level error trace. From the dead-letter queue an optional GeoIP/ASN enrichment step re-injects records whose IP prefix is valid back into the validator.NetFlow v9 / IPFIXraw record dictPydantic validatorstrict mode · O(1) validatorsextra="forbid"Schemavalid?yesCorrelation enginecanonical NetworkEventnoDead-letter queuefield-level error traceGeoIP / ASN enrichmentre-inject when prefix validoptional fallback re-inject

Production-grade Pydantic model

The following model uses strict mode, explicit Field constraints, and targeted validators for telecom edge cases. It is Pydantic V2 syntax (model_config, field_validator classmethods) for deterministic validation traces and minimal coercion overhead.

import ipaddress
from enum import IntEnum
from typing import Optional
from pydantic import BaseModel, Field, field_validator, ConfigDict, ValidationError

class IPProtocol(IntEnum):
    ICMP = 1
    TCP = 6
    UDP = 17
    GRE = 47
    BGP = 179
    # Reference: https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml

class NetFlowRecord(BaseModel):
    # Strict mode prevents silent coercion (e.g., "123" -> 123)
    model_config = ConfigDict(strict=True, extra="forbid")

    timestamp_ns: int = Field(gt=0, description="Nanosecond-precision epoch")
    src_ip: str
    dst_ip: str
    src_port: int = Field(ge=0, le=65535)
    dst_port: int = Field(ge=0, le=65535)
    protocol: IPProtocol
    input_snmp: int = Field(ge=0, le=4294967295)
    output_snmp: int = Field(ge=0, le=4294967295)
    src_as: int = Field(ge=0, le=4294967295)
    dst_as: int = Field(ge=0, le=4294967295)
    bytes_transferred: int = Field(ge=0)
    packets: int = Field(ge=0)
    tcp_flags: Optional[int] = Field(default=None, ge=0, le=255)

    @field_validator("src_ip", "dst_ip")
    @classmethod
    def validate_ip(cls, v: str) -> str:
        try:
            ipaddress.ip_address(v)
            return v
        except ValueError as e:
            raise ValueError(f"Invalid IP address: {v}") from e

    @field_validator("protocol")
    @classmethod
    def validate_protocol(cls, v: IPProtocol) -> IPProtocol:
        # Reject deprecated or experimental codes common in legacy collectors
        if v not in (IPProtocol.ICMP, IPProtocol.TCP, IPProtocol.UDP, IPProtocol.GRE, IPProtocol.BGP):
            raise ValueError(f"Unsupported protocol code: {v}")
        return v

    @field_validator("tcp_flags")
    @classmethod
    def validate_tcp_flags(cls, v: Optional[int]) -> Optional[int]:
        if v is None:
            return v
        # SYN+FIN set together is an illegal combination used by stealth scans.
        # (Bits 6-7 are ECE/CWR for ECN and are perfectly valid, so they are
        # intentionally NOT rejected here.)
        SYN, FIN = 0x02, 0x01
        if (v & SYN) and (v & FIN):
            raise ValueError(f"Illegal SYN+FIN flag combination: {v:#04x}")
        return v

The extra="forbid" setting is deliberate: an unmodeled field from a misbehaving exporter is a schema-drift signal, not something to silently swallow. The 32-bit upper bounds on src_as/dst_as accept 4-byte ASNs (RFC 6793) while still rejecting garbage values produced by truncated templates.

Async ingestion hook

NetFlow collectors run at line rate, so validation must never block the event loop. The hook below batches records, isolates each failure, and routes malformed payloads to a dead-letter queue (DLQ) without stalling the consumer. It is the integration point between this validator and the broader async ingestion pipeline.

import asyncio
import logging
from typing import List, Dict, Any
from pydantic import ValidationError

logger = logging.getLogger("netflow_ingestion")

async def process_flow_batch(raw_records: List[Dict[str, Any]], dlq_queue: asyncio.Queue) -> List[NetFlowRecord]:
    """Parse and validate a batch of raw NetFlow dictionaries."""
    valid_records = []

    for idx, raw in enumerate(raw_records):
        try:
            record = NetFlowRecord.model_validate(raw)
            valid_records.append(record)
        except ValidationError as e:
            # Emit structured error trace for NOC dashboards
            error_payload = {
                "index": idx,
                "raw": raw,
                "errors": e.errors(include_url=False),
                "timestamp_ns": raw.get("timestamp_ns", 0),
            }
            await dlq_queue.put(error_payload)
            logger.warning("NetFlow validation failure", extra=error_payload)

    return valid_records

async def ingestion_worker(dlq_queue: asyncio.Queue, batch_size: int = 500):
    """Async consumer loop with backpressure handling."""
    while True:
        # In production: pull from Kafka/Pulsar or a FastAPI request queue
        raw_batch = await fetch_next_batch(batch_size)
        valid = await process_flow_batch(raw_batch, dlq_queue)

        if valid:
            await route_to_correlation_engine(valid)
        await asyncio.sleep(0)  # Yield to the event loop

Because model_validate is pure-CPU and bounded per record, the only place this loop can stall is downstream I/O — which is exactly why both the correlation handoff and the DLQ put are awaited rather than called synchronously.

Mitigation and hardening

When validation fails the system must degrade gracefully rather than halt. These are the concrete failure paths a production deployment should implement:

  1. Dead-letter queue isolation. Malformed records are serialized with their exact ValidationError.errors() output and pushed to a dedicated DLQ topic. This preserves forensic data for vendor debugging while keeping the primary correlation pipeline unblocked. Alert when the DLQ rejection rate exceeds 0.5% over a five-minute window — a spike is usually a template or firmware change on a single exporter.
  2. Fallback enrichment routing. A record that fails only on a missing src_as/dst_as but carries valid IP prefixes can be enriched via a GeoIP/ASN lookup and re-injected, rather than discarded. This prevents false-negative alert suppression during collector misconfiguration windows.
  3. Structured error traces to runbooks. The machine-readable loc and msg fields map directly to NOC runbooks, so automation can auto-generate a corrective ticket (for example NETFLOW-INVALID-TCP-FLAGS) without manual triage.
  4. Security boundary enforcement. Spoofed source IPs, out-of-range ASNs, and illegal TCP-flag combinations frequently signal reconnaissance or DDoS preparation. Rejecting them at the edge — the same posture defined in Security Boundary Mapping — prevents poisoned telemetry from corrupting anomaly-detection baselines.

Operational hardening notes

Tuning this pattern is about keeping the per-record cost flat under storm load. Strict mode (ConfigDict(strict=True)) removes the runtime type-coercion path entirely, which both speeds validation and makes failures explicit. Keep @field_validator bodies cheap — the IntEnum lookup and the integer bit-mask on tcp_flags are O(1); avoid pulling heavy regex or DNS calls into the hot path. Validate at the consumer-batch level (500–1000 records is a sound starting point for Kafka/Pulsar) rather than per message, so the event loop yields between batches instead of between records. Validation workers should stay stateless and horizontally scalable: deploy several consumers behind a load balancer so a node under memory pressure from oversized payloads fails over without dropping telemetry. Integrated this way, the NetFlow validator becomes one deterministic gate in a unified ingestion fabric that holds the same sub-2-second SLA ticket-creation budget as the rest of the taxonomy.

Frequently Asked Questions

Why use Pydantic strict mode instead of default coercion?

Default mode silently coerces "6" into protocol 6, which masks an exporter that is emitting strings where the template promises integers. Strict mode (ConfigDict(strict=True)) rejects that record so the schema drift surfaces as a DLQ entry instead of corrupting a correlation key. It is also faster, because the coercion path is skipped entirely.

How do I handle 4-byte ASNs without rejecting valid records?

Set the src_as/dst_as bounds to the full 32-bit range (le=4294967295) per RFC 6793. That accepts every legitimate 4-byte ASN while still rejecting truncated or garbage values produced by a malformed NetFlow v9 template, so you get range enforcement without false rejections.

What should happen to records that fail validation?

Route them to a dead-letter queue carrying the raw payload, the flat errors() list, and the source timestamp, rather than dropping them. The DLQ is alertable — flag a rejection rate above 0.5% over five minutes — and replayable, so a malformed-event spike becomes a measurable signal instead of silent data loss.

Will per-record validation become a throughput bottleneck at line rate?

No, if you validate in batches and keep validators O(1). model_validate is pure CPU and bounded per record; the loop only stalls on awaited downstream I/O. Batching 500–1000 records per consumer poll and awaiting the correlation handoff keeps the event loop responsive under alarm-storm load.