Linking Multi-Stage Lateral Movement Across Telemetry

Lateral movement is invisible to any single detection because each hop looks individually benign — a valid login here, a remote-service start there, a privilege grant somewhere else. This guide implements the stage-linking logic that reconstructs the whole chain, as one precise technique within cross-source event linking and the broader alert correlation rule engines pipeline.

Root-Cause Context

Lateral movement fragments across telemetry boundaries by design. An adversary using stolen credentials (T1078 Valid Accounts) authenticates to a jump host, starts a remote service or remote desktop session (T1021 Remote Services) to reach a second host, then escalates (TA0004 Privilege Escalation) — and each of those actions is emitted by a different system: the identity provider, the endpoint EDR, and the domain controller. In isolation, a single successful login is the most normal event in the enterprise. The malicious signal exists only in the sequence and the shared principal across sources, which is exactly what a per-source rule cannot see.

Three factors defeat naive detection. First, the stages share no verbatim field — the IdP names the actor one way, the EDR another, the DC a third — so they must be resolved to a common entity before they can be joined. Second, the hops are ordered in time but arrive out of order and across skewed clocks. Third, the chain crosses hosts, so linking on a single host id fractures it; the join key must follow the principal (and its acquired identities) as it moves. Getting this right turns three unremarkable events into one high-confidence lateral-movement incident.

Prerequisites

The implementation targets Python 3.11+ and uses only the standard library plus pydantic for the event contract. It assumes events are already normalized to the site’s ECS-aligned schema by JSON event normalization and resolved to a principal key as described in the parent cross-source event linking guide.

pip install "pydantic>=2.6,<3.0"

The key configuration assumption is a dwell-time-sized window: lateral movement can span hours, so the window here defaults to two hours, far longer than the minutes-scale window used for credential replay.

Production-Ready Implementation

The engine tracks an ordered set of ATT&CK stages per principal and fires when a configured minimum-length ordered subsequence of the lateral-movement kill chain is observed within the window.

from __future__ import annotations

import logging
from dataclasses import dataclass, field

from pydantic import BaseModel, Field

logger = logging.getLogger("soc.correlation.lateral")

# Ordered stages of the lateral-movement chain, by ATT&CK id.
CHAIN = ["T1078", "T1021", "TA0004", "T1543"]


class StageEvent(BaseModel):
    event_id: str = Field(min_length=1)
    event_time: float = Field(gt=0)
    principal_key: str = Field(min_length=1)   # resolved upstream
    host_id: str = Field(min_length=1)
    technique_id: str = Field(min_length=1)


@dataclass
class ChainState:
    events: list[StageEvent] = field(default_factory=list)


class LateralMovementLinker:
    def __init__(self, window_seconds: float = 7200.0, min_stages: int = 3) -> None:
        self.window = window_seconds
        self.min_stages = min_stages
        self._state: dict[str, ChainState] = {}

    def admit(self, ev: StageEvent, now: float) -> str:
        if ev.technique_id not in CHAIN:
            return "ERR_LINK_053"                      # not a chain stage
        st = self._state.setdefault(ev.principal_key, ChainState())
        cutoff = now - self.window
        st.events = [e for e in st.events if e.event_time > cutoff]
        st.events.append(ev)
        return "OK"

    def detected_chain(self, principal_key: str) -> list[StageEvent] | None:
        """Return the ordered stage events if an ordered subsequence >= min_stages."""
        st = self._state.get(principal_key)
        if not st:
            return None
        ordered = sorted(st.events, key=lambda e: e.event_time)
        seq: list[StageEvent] = []
        chain_pos = 0
        for ev in ordered:
            # Advance through CHAIN in order; skip repeats and out-of-chain stages.
            idx = CHAIN.index(ev.technique_id)
            if idx >= chain_pos:
                seq.append(ev)
                chain_pos = idx + 1
        distinct_stages = {e.technique_id for e in seq}
        distinct_hosts = {e.host_id for e in seq}
        if len(distinct_stages) >= self.min_stages and len(distinct_hosts) >= 2:
            logger.info(
                '{"msg":"lateral_movement","principal":"%s","stages":%d,"hosts":%d}',
                principal_key, len(distinct_stages), len(distinct_hosts),
            )
            return seq
        return None


if __name__ == "__main__":
    linker = LateralMovementLinker()
    base = 1_000_000.0
    chain = [
        StageEvent(event_id="e1", event_time=base, principal_key="P1",
                   host_id="JUMP-1", technique_id="T1078"),
        StageEvent(event_id="e2", event_time=base + 600, principal_key="P1",
                   host_id="JUMP-1", technique_id="T1021"),
        StageEvent(event_id="e3", event_time=base + 1800, principal_key="P1",
                   host_id="SRV-9", technique_id="TA0004"),
    ]
    for e in chain:
        linker.admit(e, now=base + 1800)
    assert linker.detected_chain("P1") is not None

The two-host requirement (distinct_hosts >= 2) is what distinguishes lateral movement from a single-box compromise: the chain must actually move. The ordered-subsequence check tolerates interleaved benign events and repeated stages without requiring the stages to be strictly adjacent.

Error-Code Reference

Operational Notes

• Window sizing dominates memory. A two-hour window holding a handful of stage events per principal is cheap, but broad service accounts that touch every host inflate one key — cap per-principal cardinality and exclude known automation identities.
• Order matters more than adjacency. Require the stages in ATT&CK order but allow gaps, so an attacker cannot break the rule by interleaving noise between hops.
• Feed the match forward. A detected chain is a high-confidence session; emit it with all contributing technique_ids so dynamic severity scoring can apply a convergence multiplier and SOAR can act.

Verification Checklist

• A three-stage chain (T1078 → T1021 → TA0004) across two hosts fires within the window.
• The same three stages on a single host do not fire (no actual movement).
• Stages arriving in reverse order do not produce a false chain.
• A stage arriving after the window closes yields ERR_LINK_011 and does not extend a prior chain.
• Interleaved benign events between stages do not prevent the match.
• An unresolved principal yields ERR_LINK_054, not a crash.

FAQ

Related

• Cross-Source Event Linking
• Correlating Identity and Network Events
• Dynamic Severity Scoring
• Temporal Correlation Windows
• Alert Correlation & Rule Engines