125 lines
5.0 KiB
Python
125 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from collections import Counter
|
|
from collections.abc import Iterable
|
|
|
|
from .models import LogEvent
|
|
from .normalization import canonical_value
|
|
from .query_details import event_query_details
|
|
|
|
|
|
ENTITY_FIELDS: dict[str, tuple[str, ...]] = {
|
|
"ip": ("srcip", "src_ip", "source_ip", "source.ip", "client_ip", "client.ip", "remote_addr", "remote_ip", "ip", "ipaddress", "winlog_event_data_ipaddress", "winlog.event_data.ipaddress", "winlog.event_data.ip_address", "event_data_ipaddress"),
|
|
"user": ("username", "user", "user.name", "user_name", "account", "account_name", "targetusername", "subjectusername", "xauthuser", "winlog_event_data_targetusername", "winlog.event_data.targetusername", "winlog_event_data_subjectusername", "winlog.event_data.subjectusername"),
|
|
"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:
|
|
ipaddress.ip_address(value)
|
|
return "ip"
|
|
except ValueError:
|
|
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]] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
for kind, fields in ENTITY_FIELDS.items():
|
|
for field in fields:
|
|
value = str(event.fields.get(field, "")).strip()
|
|
if not value or value in {"-", "unknown", "n/a"}:
|
|
continue
|
|
key = (kind, value.lower() if kind != "ip" else value)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
identities.append({"entity": value, "entity_type": kind, "field": field})
|
|
return identities
|
|
|
|
|
|
def profile_entity(event: LogEvent, field: str) -> str:
|
|
return str(event.fields.get(field.lower(), "")).strip()
|
|
|
|
|
|
def profile_entities(event: LogEvent, profile: object) -> tuple[str, ...]:
|
|
fields = tuple(str(field) for field in getattr(profile, "entity_fields", ()) if field) or (str(getattr(profile, "entity_field", "")),)
|
|
values = []
|
|
for field in fields:
|
|
value = profile_entity(event, field)
|
|
if value and value not in values:
|
|
values.append(value)
|
|
return tuple(values)
|
|
|
|
|
|
def _timeline_sample(event: LogEvent) -> dict[str, object]:
|
|
query_details = event_query_details(event)
|
|
return {
|
|
"stream": event.fields.get("fgai_stream", "local_syslog"),
|
|
"timestamp": event.fields.get("eventtime", event.fields.get("timestamp", "")),
|
|
"type": canonical_value(event.fields, "type"),
|
|
"subtype": event.subtype,
|
|
"action": event.action,
|
|
"severity": event.severity,
|
|
"destination": event.dst_ip or canonical_value(event.fields, "context"),
|
|
"service": canonical_value(event.fields, "service"),
|
|
"context": canonical_value(event.fields, "context")[:240],
|
|
"query_details": query_details,
|
|
"graylog_query": query_details["query"],
|
|
}
|
|
|
|
|
|
def sample_timeline(events: Iterable[LogEvent], *, limit: int = 20) -> list[dict[str, object]]:
|
|
samples = [_timeline_sample(event) for event in events]
|
|
return sorted(samples, key=lambda item: item["timestamp"])[-limit:]
|