Correlating Identity and Network Events by Resolved Principal

Network telemetry knows IP addresses; identity telemetry knows users; and the gap between them is where account-takeover investigations stall. This guide implements the time-bounded IP-to-user binding that joins the two, as one precise technique within cross-source event linking and the broader alert correlation rule engines pipeline.

Root-Cause Context

A proxy log says 203.0.113.9 exfiltrated 4 GB to a rare ASN; an identity log says jdoe authenticated from a new device. Only by binding that IP to that user, at that time, does the SOC learn they are the same event — that jdoe’s session exfiltrated the data. The binding is hard because IP-to-user is a temporal relationship: DHCP leases, VPN pool assignments, and NAT gateways mean an address maps to different principals at different moments, so a static IP-to-user table is wrong the moment it is written.

Two failure modes dominate. First, stale binding: joining on an IP without a time bound attributes one user’s network activity to whoever held that address yesterday. Second, NAT collapse: many users behind one egress IP make a naive IP join fuse unrelated people into a phantom shared session. The fix is a lease-style binding derived from authoritative session-start events (VPN connect, 802.1X, DHCP, IdP session establishment) with an explicit validity interval, so a network event is attributed only to the principal who actually held the address at that instant.

Prerequisites

The implementation targets Python 3.11+ and uses only the standard library plus pydantic. It assumes IP addresses and timestamps are normalized upstream by JSON event normalization, and that resolved principals feed the parent cross-source event linking window.

pip install "pydantic>=2.6,<3.0"

Production-Ready Implementation

The binding store holds interval leases per IP; a network event is attributed to the lease whose interval contains its timestamp. Overlapping or absent leases fail closed rather than guessing.

from __future__ import annotations

import bisect
import logging
from dataclasses import dataclass

from pydantic import BaseModel, Field

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


class NetworkEvent(BaseModel):
    event_id: str = Field(min_length=1)
    src_ip: str = Field(min_length=1)
    event_time: float = Field(gt=0)


@dataclass(frozen=True)
class Lease:
    principal: str
    start: float
    end: float          # inclusive upper bound; use +inf for open leases


class IpUserBinder:
    def __init__(self) -> None:
        # Per-IP sorted lease starts for binary-search lookup.
        self._leases: dict[str, list[Lease]] = {}

    def add_lease(self, ip: str, lease: Lease) -> None:
        arr = self._leases.setdefault(ip, [])
        idx = bisect.bisect_left([l.start for l in arr], lease.start)
        arr.insert(idx, lease)

    def resolve(self, ev: NetworkEvent) -> str | None:
        """Return the principal whose lease covers ev.event_time, else None."""
        arr = self._leases.get(ev.src_ip)
        if not arr:
            return None                                  # -> ERR_LINK_061
        # Find the latest lease starting at or before the event time.
        starts = [l.start for l in arr]
        idx = bisect.bisect_right(starts, ev.event_time) - 1
        if idx < 0:
            return None
        lease = arr[idx]
        if lease.start <= ev.event_time <= lease.end:
            return lease.principal
        return None                                      # gap between leases

    def attribute(self, ev: NetworkEvent) -> dict | None:
        principal = self.resolve(ev)
        if principal is None:
            logger.warning("ERR_LINK_061 no binding ip=%s t=%.0f",
                           ev.src_ip, ev.event_time)
            return None
        return {"event_id": ev.event_id, "principal": principal,
                "src_ip": ev.src_ip, "event_time": ev.event_time}


if __name__ == "__main__":
    binder = IpUserBinder()
    binder.add_lease("10.2.0.7", Lease("alice", 1_000_000.0, 1_003_600.0))
    binder.add_lease("10.2.0.7", Lease("bob", 1_003_600.1, 1_007_200.0))
    ev = NetworkEvent(event_id="n1", src_ip="10.2.0.7", event_time=1_002_000.0)
    assert binder.attribute(ev)["principal"] == "alice"

The binary search over lease starts makes attribution O(log n) per event even for an address that has churned through hundreds of leases in a day. Open-ended leases (end = float("inf")) model still-active sessions; they are closed when the corresponding session-end event arrives.

Error-Code Reference

Operational Notes

• Bindings are leases, not a table. Always attribute by the interval that contains the event timestamp; never by “whoever holds this IP now.”
• NAT needs a finer key. Behind a shared egress IP, fall back to a more specific identifier (source port range, x-forwarded-for, or an inner VPN address) or decline to attribute rather than fusing users.
• Watch lease-source lag. If VPN or DHCP logs arrive after the network events they should bind, briefly buffer network events so the lease exists before attribution — the same lateness discipline used in temporal correlation windows.

Verification Checklist

• A network event during alice’s lease attributes to alice, not the prior holder.
• An event one second into bob’s lease attributes to bob (boundary correctness).
• An event in a gap between leases yields ERR_LINK_061, not a wrong attribution.
• Overlapping leases raise ERR_LINK_062 rather than silently picking one.
• Attribution stays O(log n) for an IP with hundreds of daily leases.
• A missing IP or timestamp yields ERR_LINK_063, not a crash.

FAQ

Related

• Cross-Source Event Linking
• Linking Multi-Stage Lateral Movement
• Temporal Correlation Windows
• JSON Event Normalization
• Alert Correlation & Rule Engines