Mapping Common Event Format (CEF) to Elastic Common Schema (ECS) fails the moment a vendor’s flat, stringly-typed cs1Label/cs1 extension pairs meet ECS’s strict typed hierarchy — leaving severities uncomparable, IPs unindexed, and correlation rules silently dropping events. This page is one precise technique inside JSON event normalization, part of the broader SOC Log Architecture & Taxonomy discipline.

Root-Cause Context

CEF was designed as a transport envelope, not a semantic model. The seven pipe-delimited header fields (CEF:Version|DeviceVendor|DeviceProduct|DeviceVersion|DeviceEventClassId|Name|Severity) are positional and predictable, but everything that actually drives detection lives in the trailing extension string — an unbounded set of key=value pairs where every value is a string and the keys are whatever the device vendor decided to emit. That design choice produces three distinct failure modes when the stream lands in an ECS-backed SIEM unmodified.

The first is type collapse. ECS declares source.ip as type ip, source.port as long, @timestamp as date, and event.severity as long. CEF hands all of them over as strings. An un-coerced spt=443 indexed as a keyword cannot satisfy a range query, so a rule that fires on destination.port < 1024 never matches, and a threat-intel lookup against source.ip compares text to a CIDR and returns nothing.

The second is custom-label drift. CEF reserves generic slots — cs1..cs6 (custom strings), cn1..cn3 (custom numbers), cfp1..cfp4 (floats) — each paired with a *Label field that names what the slot means for this device. One firewall emits cs1Label=FirewallRule cs1=DROP; another emits cs2Label=Rule cs2=block. Indexing the raw cs1/cs2 keys fragments identical telemetry across mismatched fields, so correlation logic has to be rewritten per device instead of expressed once against rule.name.

The third is silent malformation. A truncated payload, an unescaped = inside a value, or a missing pipe yields a half-parsed event that looks valid enough to index but is missing the fields a rule depends on. Without an explicit quarantine path these events vanish into the index and surface later as a correlation gap during an incident — exactly when forensic completeness matters. A disciplined mapper resolves all three before indexing: decode the header, route extensions by their label, coerce every typed field, and dead-letter anything that fails so the error categorization frameworks can act on it instead of losing it.

CEF-to-ECS mapping lifecycle with a dead-letter branch A raw CEF syslog line passes through a six-stage chain: split the seven pipe-delimited header fields, parse the trailing key=value extension blob with backslash unescaping, coerce typed fields to ECS ip, long, and date, route custom cs and cn slots to stable ECS fields by their paired label, then validate against an ECS strict template before emitting a canonical ECS document. A malformed header drops to the dead-letter sink as ERR_CEF_001; uncoercible values drop as ERR_CEF_002 for timestamps, ERR_CEF_003 for IP addresses, and ERR_CEF_004 for numerics; a strict-template rejection drops a stray field; and an unmapped label logs the non-fatal ERR_CEF_005 while the event still indexes. The dead-letter sink preserves every rejected raw payload with its ERR_CEF code for replay. Mapping CEF to ECS — six-stage lifecycle Every event exits as a typed ECS document or as a coded ERR_CEF_* rejection in the dead-letter sink — never silently dropped. ERR_CEF_001 ERR_CEF_002 · date ERR_CEF_003 · ip ERR_CEF_004 · num stray field dynamic:strict ERR_CEF_005 non-fatal · indexes Raw CEF CEF:0|Vendor|… |8| ext blob syslog line 1 Header split 7 pipe fields Vendor → Severity anchored regex 2 Extension parse key=value pairs unescape \= \\ \n 3 Type coerce ip · long · date src · spt · rt 4 Label route cs*/cn* → ECS by *Label 5 Validate ECS template dynamic:strict ECS document source.ip · long @timestamp · date indexable Dead-letter sink Rejected raw payload preserved with its ERR_CEF_* code — quarantine index, Kafka topic, or file. Replay after the forwarder, epoch unit, or label dictionary is corrected.

Prerequisites

  • Python 3.11+ — the implementation uses dataclasses(slots=True), the | union syntax, and enum.StrEnum; the standard library alone covers it (re, ipaddress, datetime), so there is no third-party dependency for the core mapper.
  • A target ECS version pinned in config. Field names move between ECS major versions; pin one (for example ECS 8.11) and validate against the official ECS field reference rather than assuming stability.
  • A device extension dictionary. Before mapping, collect each device’s *Label conventions — cs1Label, cs2Label, cn1Label — because the label, not the slot number, decides the ECS target. Treat this as configuration, not code.
  • A dead-letter sink. A Kafka topic, a quarantine index, or even a file — anywhere a rejected payload lands intact with its ERR_CEF_* code so it can be replayed after the dictionary is corrected.

Production-Ready Implementation

The mapper below is self-contained and runnable. It decodes the CEF header, parses extensions (honouring the backslash escaping CEF mandates for =, \, and newlines inside values), coerces each typed field with ipaddress and datetime, and routes cs*/cn* slots through a label-driven table to stable ECS fields. Every rejection returns a typed MapResult carrying an ERR_CEF_* code instead of raising into the hot path, so a single bad event never stalls the batch.

from __future__ import annotations

import ipaddress
import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import StrEnum
from typing import Any

# CEF header: "CEF:" + 7 pipe-delimited fields, with the extension blob last.
_HEADER = re.compile(
    r"CEF:(?P<cef_version>\d+)\|(?P<vendor>(?:[^|\\]|\\.)*)\|"
    r"(?P<product>(?:[^|\\]|\\.)*)\|(?P<dev_version>(?:[^|\\]|\\.)*)\|"
    r"(?P<class_id>(?:[^|\\]|\\.)*)\|(?P<name>(?:[^|\\]|\\.)*)\|"
    r"(?P<severity>(?:[^|\\]|\\.)*)\|(?P<extensions>.*)$"
)
# Extension tokens: key=value where value runs until the next " key=".
_EXT = re.compile(r"(?P<key>\w+)=(?P<val>(?:[^\\]|\\.)*?)(?=\s+\w+=|$)")


class CefError(StrEnum):
    MALFORMED_HEADER = "ERR_CEF_001"   # header did not match the 7-field grammar
    BAD_TIMESTAMP = "ERR_CEF_002"      # rt/end could not be parsed to a date
    BAD_IP = "ERR_CEF_003"             # src/dst was not a valid IPv4/IPv6 address
    BAD_NUMERIC = "ERR_CEF_004"        # port/severity was non-numeric
    UNMAPPED_LABEL = "ERR_CEF_005"     # cs*/cn* slot had no dictionary entry (non-fatal)


@dataclass(slots=True)
class MapResult:
    """Outcome of one CEF->ECS mapping. ok is False only for fatal errors."""

    ok: bool
    document: dict[str, Any] = field(default_factory=dict)
    errors: list[str] = field(default_factory=list)
    raw: str = ""


# Direct CEF-extension -> ECS-field mapping with a coercion tag.
_FIELD_MAP: dict[str, tuple[str, str]] = {
    "src": ("source.ip", "ip"),
    "dst": ("destination.ip", "ip"),
    "spt": ("source.port", "long"),
    "dpt": ("destination.port", "long"),
    "suser": ("source.user.name", "keyword"),
    "duser": ("destination.user.name", "keyword"),
    "proto": ("network.transport", "keyword"),
    "app": ("network.application", "keyword"),
    "request": ("url.original", "keyword"),
}
# Contextual routing for custom slots: cs*Label value -> ECS target.
_LABEL_MAP: dict[str, str] = {
    "FirewallRule": "rule.name",
    "Rule": "rule.name",
    "User": "user.name",
    "Geo": "source.geo.country_iso_code",
    "ThreatName": "threat.indicator.name",
}


def _unescape(value: str) -> str:
    return re.sub(r"\\(.)", r"\1", value)


def _set(doc: dict[str, Any], dotted: str, value: Any) -> None:
    """Expand a dotted ECS path into the nested document structure."""
    parts = dotted.split(".")
    cursor = doc
    for part in parts[:-1]:
        cursor = cursor.setdefault(part, {})
    cursor[parts[-1]] = value


def _coerce(kind: str, value: str) -> Any:
    if kind == "ip":
        return str(ipaddress.ip_address(value))           # raises ValueError if bad
    if kind == "long":
        return int(value)
    return value


def map_cef_to_ecs(raw: str, labels: dict[str, str] | None = None) -> MapResult:
    result = MapResult(ok=True, raw=raw)
    header = _HEADER.match(raw.strip())
    if not header:
        return MapResult(ok=False, errors=[CefError.MALFORMED_HEADER], raw=raw)

    g = header.groupdict()
    doc = result.document
    _set(doc, "observer.vendor", _unescape(g["vendor"]))
    _set(doc, "observer.product", _unescape(g["product"]))
    _set(doc, "observer.version", _unescape(g["dev_version"]))
    _set(doc, "event.code", _unescape(g["class_id"]))
    _set(doc, "event.action", _unescape(g["name"]))
    _set(doc, "event.kind", "event")

    try:
        _set(doc, "event.severity", int(g["severity"]))
    except ValueError:
        result.errors.append(CefError.BAD_NUMERIC)

    ext = {m["key"]: _unescape(m["val"]) for m in _EXT.finditer(g["extensions"])}
    label_overrides = {k: v for k, v in ext.items() if k.endswith("Label")}

    # Timestamp: CEF rt is epoch milliseconds; emit ISO-8601 UTC.
    if "rt" in ext:
        try:
            ts = datetime.fromtimestamp(int(ext["rt"]) / 1000, tz=timezone.utc)
            _set(doc, "@timestamp", ts.isoformat())
        except (ValueError, OverflowError):
            result.errors.append(CefError.BAD_TIMESTAMP)

    for key, value in ext.items():
        if key.endswith("Label"):
            continue
        if key in _FIELD_MAP:
            target, kind = _FIELD_MAP[key]
            try:
                _set(doc, target, _coerce(kind, value))
            except ValueError:
                result.errors.append(
                    CefError.BAD_IP if kind == "ip" else CefError.BAD_NUMERIC
                )
            continue
        # Custom slot (cs1, cn2, ...): resolve via its *Label.
        if re.fullmatch(r"c[sn]\d", key):
            label = label_overrides.get(f"{key}Label", "")
            target = _LABEL_MAP.get(label) or (labels or {}).get(label)
            if target:
                _set(doc, target, value)
            else:
                result.errors.append(CefError.UNMAPPED_LABEL)

    # Drop nulls and fail only on fatal (non-UNMAPPED) errors.
    fatal = [e for e in result.errors if e != CefError.UNMAPPED_LABEL]
    result.ok = not fatal
    return result


if __name__ == "__main__":
    sample = (
        "CEF:0|VendorX|FW|1.0|1001|Connection Drop|8|"
        "rt=1690000000000 src=10.0.0.5 dst=192.168.1.10 spt=443 dpt=80 "
        "proto=TCP cs1Label=FirewallRule cs1=DROP"
    )
    out = map_cef_to_ecs(sample)
    print(out.ok, out.errors)
    print(out.document)

The parser deliberately decouples fatal failures (a malformed header, a value that cannot be coerced into the type its rule depends on) from non-fatal ones (a custom slot with no dictionary entry). An unmapped cs3 should not discard an otherwise-good event; it should record ERR_CEF_005 and let the document index while you extend the dictionary. This mirrors how normalizing JSON logs from cloud providers treats unknown keys — preserve the event, flag the gap.

If you would rather keep parsing inside the SIEM than run a sidecar, the same contract expresses cleanly as an Elasticsearch ingest pipeline: a grok to split the header, a kv processor (field_split: " ", value_split: "=") for extensions, a date processor with format UNIX_MS onto @timestamp, convert processors to coerce source.port/event.severity, and rename processors with ignore_missing onto source.ip/destination.ip — with on_failure routing to a quarantine index so the dead-letter discipline survives the move.

Error-Code Reference

Code Meaning Action
ERR_CEF_001 The string did not match the 7-field CEF header grammar (truncated transport, an unescaped pipe, or a non-CEF payload). Dead-letter the raw event; check the forwarder for message truncation or a wrong syslog framing mode, then replay.
ERR_CEF_002 rt/end could not be parsed to a timestamp (non-numeric, or seconds where milliseconds were expected). Confirm the device’s epoch unit; add a per-device unit hint to config rather than guessing at ingest time.
ERR_CEF_003 src/dst was not a valid IPv4/IPv6 address — often a hostname leaking into an IP slot. Route hostnames to source.domain instead; do not index them as ip or every range/CIDR query silently misses.
ERR_CEF_004 A port or severity value was non-numeric. Inspect the device profile; some vendors emit textual severities (High) that need a lookup table to the ECS 0–10 scale.
ERR_CEF_005 A cs*/cn* slot had no entry in the label dictionary (non-fatal). Extend the device extension dictionary with the new *Label; the event still indexes, so this is a backlog item, not an outage.

These follow the ERR_CATEGORY_NNN convention used across the pipeline, so CEF rejections slot into the shared error categorization frameworks without a translation layer.

Operational Notes

  • CPU/memory profile. Mapping is a regex match plus a bounded dictionary walk — single-digit microseconds per event and constant memory, so a single core sustains tens of thousands of EPS. The cost is in the regex: anchor it ($) and avoid catastrophic backtracking by keeping the extension token pattern non-greedy with an explicit lookahead, as above.
  • Severity, not risk. Map CEF Severity (0–10) to event.severity only when it is a true ordinal. Vendor “risk” or “confidence” scores belong in event.risk_score (a float), and a textual severity needs a deterministic lookup to the numeric scale before it reaches a range rule — never coerce "High" to a magic number inline.
  • Label dictionaries are per-device. cs1 means a firewall rule on one appliance and a URL category on another. Key the dictionary on (observer.product, slot, label) so two products never collide on slot number, and version it in Git alongside the rest of your config-as-code.
  • Validate downstream. Mapping does not guarantee ECS legality — a typo’d target field indexes a stray key. Run mapped documents through a schema-validation stage or an ECS template with dynamic: strict so an unknown field is rejected loudly, not absorbed.
  • Enrichment readiness. Because source.ip/destination.ip are now real ip values, the document is immediately eligible for MITRE ATT&CK integration tagging and for threat-indicator matching from threat intel feed mapping — both of which compare typed fields, not string blobs.
  • CSV fallbacks. During legacy migrations the same telemetry sometimes arrives as CSV; align those columns to the CEF extension dictionary so positional drift does not silently shift spt into dpt. See CSV ingestion patterns for the quoting and column-ordering rules.

Verification Checklist

FAQ

How do I handle the cs1/cs2 custom fields without hardcoding every device?

Route by the paired *Label, not the slot number. CEF guarantees that cs1Label names what cs1 holds for that device, so resolve cs1 through a dictionary keyed on its label value (FirewallRule -> rule.name). Keep that dictionary in config, keyed on (observer.product, slot, label) so two products can reuse cs1 for different meanings without colliding. A label with no entry should emit ERR_CEF_005 and still index the event — that turns an unknown extension into a backlog item you can fix by editing config, not a dropped event or a code change.

CEF severity is 0–10 but my detection rules use ECS event.severity — what scale should I store?

Store the raw 0–10 ordinal in event.severity as a long only when the vendor genuinely emits an ordinal severity. If the device sends a textual level (Low/High) or a separate “risk”/“confidence” number, do not coerce it inline: map text through a deterministic lookup table to the numeric scale, and put non-ordinal risk scores in event.risk_score (a float). Mixing a confidence score into event.severity breaks every range-based correlation rule that assumes the field is a true severity ordinal.

Where should malformed CEF events go so I don't lose them?

Into a dead-letter sink that preserves the raw string plus its ERR_CEF_* code — a quarantine index, a Kafka topic, or a file. The mapper returns a typed MapResult instead of raising, so one bad event never stalls a batch, and the original payload survives for replay once you fix the root cause (a truncating forwarder for ERR_CEF_001, a wrong epoch unit for ERR_CEF_002). Replaying from the sink after a dictionary fix recovers the events without re-ingesting the entire source.