Entity labels now resolve like this for IP-based entities:

This commit is contained in:
larssand
2026-07-02 20:30:33 +02:00
parent 48b56852cf
commit 5e14104f0a
5 changed files with 78 additions and 49 deletions

View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import ipaddress
from collections import Counter
from collections.abc import Iterable
from .models import LogEvent
@@ -14,6 +15,26 @@ ENTITY_FIELDS: dict[str, tuple[str, ...]] = {
"host": ("hostname", "host", "host.name", "computer", "computer_name", "workstation", "device_name", "winlog_computer_name", "winlog.computer_name", "agent.hostname", "agent_name"),
}
DISPLAY_FIELD_GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
("host", ENTITY_FIELDS["host"]),
("user", ENTITY_FIELDS["user"]),
(
"service",
(
"service.name",
"service_name",
"process.name",
"process.executable",
"winlog.provider_name",
"event.provider",
"application",
"application_name",
"appname",
"app",
),
),
)
def entity_type(value: str) -> str:
try:
@@ -23,6 +44,34 @@ def entity_type(value: str) -> str:
return "entity"
def _related_value(events: Iterable[LogEvent], fields: tuple[str, ...], *, exclude: str = "") -> str:
counts: Counter[str] = Counter()
excluded = exclude.lower()
for event in events:
for field in fields:
value = str(event.fields.get(field, "")).strip()
if not value or value.lower() in {"-", "unknown", "n/a", "none", "null"}:
continue
if excluded and value.lower() == excluded:
continue
counts[value] += 1
if not counts:
return ""
return sorted(counts.items(), key=lambda item: (item[1], len(item[0]) <= 64), reverse=True)[0][0]
def entity_display(kind: str, entity: str, events: Iterable[LogEvent]) -> dict[str, str]:
"""Return a human label for an entity without changing its stable correlation key."""
if kind != "ip":
return {"entity_display": entity, "entity_label": entity, "entity_detail": ""}
event_list = list(events)
for _group, fields in DISPLAY_FIELD_GROUPS:
value = _related_value(event_list, fields, exclude=entity)
if value:
return {"entity_display": value, "entity_label": f"{value} ({entity})", "entity_detail": entity}
return {"entity_display": entity, "entity_label": entity, "entity_detail": ""}
def event_entities(event: LogEvent) -> list[dict[str, str]]:
"""Return normalized identities shared across network, endpoint, DNS, and web logs."""
identities: list[dict[str, str]] = []