Elastic ECS vs OSSEM Field Mapping: A Practitioner Crosswalk

The gap between ECS and OSSEM is not conceptual — both describe the same security events — it is a hundred small naming and typing differences that break a query the moment it crosses from one to the other. This guide is the concrete field-by-field crosswalk and a bidirectional mapping module, as one precise technique within schema mapping comparisons and the broader SOC log architecture & taxonomy discipline.

Root-Cause Context

ECS and OSSEM were designed by different communities for different primary uses, so the same concept lands under different names and, occasionally, different structures. A user’s account name is user.name in ECS and SubjectUserName (or TargetUserName, context-dependent) in OSSEM; a source address is source.ip versus SrcIpAddr; a process command line happens to agree (process.command_line vs CommandLine in the raw event). Most differences are pure renames, but a few are genuine structural conflicts — where ECS nests a field that OSSEM keeps flat, or where an enumeration’s allowed values differ.

The practical failure is a hunt written against OSSEM field names silently returning nothing when run over ECS-normalized data, because SrcIpAddr simply does not exist there. A hard-coded, one-directional translation buried in a parser cannot serve both a SIEM and a hunt notebook. The fix is an explicit, versioned crosswalk that both directions consult, with the genuine conflicts handled by named policy rather than by whichever translation happened to run first.

Prerequisites

The crosswalk targets Python 3.11+ and uses only the standard library plus pydantic. The map is version-controlled data and plugs into the parent schema mapping comparisons layer.

pip install "pydantic>=2.6,<3.0"

Production-Ready Implementation

The crosswalk is one authoritative map; both directions derive from it, so ECS-to-OSSEM and OSSEM-to-ECS can never disagree. Conflicting enumerations are resolved by an explicit policy.

from __future__ import annotations

import logging
from typing import Optional

logger = logging.getLogger("soc.taxonomy.crosswalk")

# One authoritative ECS -> OSSEM crosswalk (version-controlled).
ECS_TO_OSSEM: dict[str, str] = {
    "@timestamp": "TimeCreated",
    "source.ip": "SrcIpAddr",
    "destination.ip": "DstIpAddr",
    "user.name": "SubjectUserName",
    "process.command_line": "CommandLine",
    "process.pid": "ProcessId",
    "event.action": "EventID",
    "network.direction": "Direction",
    "host.name": "Hostname",
}
# Derived inverse — never hand-maintained, so the two cannot drift.
OSSEM_TO_ECS: dict[str, str] = {v: k for k, v in ECS_TO_OSSEM.items()}

# Enumeration conflict policy: ECS network.direction allowed values.
_ECS_DIRECTION = {"inbound", "outbound", "internal", "external", "unknown"}


def translate(fields: dict, to: str) -> tuple[dict, list[str]]:
    """Translate a field dict between 'ecs' and 'ossem'. Returns (out, unmapped)."""
    table = ECS_TO_OSSEM if to == "ossem" else OSSEM_TO_ECS
    out: dict = {}
    unmapped: list[str] = []
    for key, value in fields.items():
        target = table.get(key)
        if target is None:
            unmapped.append(key)                     # ERR_MAP_201 candidate
            out[key] = value                         # pass through, do not drop
            continue
        out[target] = value
    return out, unmapped


def normalize_direction(value: str) -> str:
    """Resolve a direction value onto the ECS enumeration; never crash."""
    v = value.strip().lower()
    alias = {"in": "inbound", "ingress": "inbound",
             "out": "outbound", "egress": "outbound"}
    resolved = alias.get(v, v)
    if resolved not in _ECS_DIRECTION:
        logger.warning("ERR_MAP_203 unknown direction=%r -> unknown", value)
        return "unknown"
    return resolved


if __name__ == "__main__":
    ecs_event = {"user.name": "jdoe", "source.ip": "10.0.0.9", "event.action": "4624"}
    ossem, unmapped = translate(ecs_event, to="ossem")
    assert ossem["SubjectUserName"] == "jdoe"
    assert ossem["SrcIpAddr"] == "10.0.0.9"
    back, _ = translate(ossem, to="ecs")
    assert back["user.name"] == "jdoe"               # round-trips exactly

Deriving OSSEM_TO_ECS as the inverse of one map is the crucial move: there is a single authoritative crosswalk, so the two translation directions cannot diverge the way two hand-maintained maps inevitably would. Unmapped fields pass through rather than being dropped, so a field the crosswalk does not yet know about is preserved for review, not lost.

Error-Code Reference

Operational Notes

• Maintain one authoritative direction; derive the inverse. Never hand-maintain both ECS-to-OSSEM and OSSEM-to-ECS; invert one map so they cannot drift.
• Pass unmapped fields through, never drop them. A field the crosswalk lacks is a coverage gap to fix, not a value to discard — dropping it is a silent detection loss.
• Handle context-dependent OSSEM names explicitly. SubjectUserName versus TargetUserName depends on event semantics; encode that as a rule (ERR_MAP_207), not a guess.
• Normalize enumerations, do not just rename them. Direction, action, and outcome values differ in allowed sets between the schemas; resolve them onto the canonical enumeration with a policy that defaults unknowns to a counted bucket.

Verification Checklist

• An ECS event translated to OSSEM and back round-trips to identical field names and values.
• source.ip and user.name map to SrcIpAddr and SubjectUserName respectively.
• A field absent from the crosswalk passes through and is counted, not dropped.
• An unknown direction value resolves to unknown and emits ERR_MAP_203.
• The inverse map is derived, not hand-maintained (changing the forward map updates both).
• Context-dependent user fields resolve by an explicit rule, not by position.

FAQ

Related

• Schema Mapping Comparisons
• JSON Event Normalization
• Threat Intel Feed Mapping
• Schema Validation Pipelines
• SOC Log Architecture & Taxonomy