Correlated Alerts on Similar User Identities

from .alerts-security.*

// truncate timestamp to 5-minute window
| eval Esql.time_window_date_trunc = date_trunc(5 minutes, @timestamp)

// high severity alerts excluding system standard user.ids
| where kibana.alert.rule.name is not null and user.name is not null and kibana.alert.risk_score >= 73 and kibana.alert.workflow_status == "open" and 
        not kibana.alert.rule.type in ("threat_match", "machine_learning") and
        not user.id in ("S-1-5-18", "S-1-5-19", "S-1-5-20", "0")

// group alerts by short time window and extract values of interest for alert triage
| stats Esql.event_module_distinct_count = COUNT_DISTINCT(event.module),
        Esql.user_name_distinct_count = COUNT_DISTINCT(user.name),
        Esql.rule_name_distinct_count = COUNT_DISTINCT(kibana.alert.rule.name),
        Esql.event_category_distinct_count = COUNT_DISTINCT(event.category),
        Esql.rule_risk_score_distinct_count = COUNT_DISTINCT(kibana.alert.risk_score),
        Esql.event_module_values = VALUES(event.module),
        Esql.rule_name_values = VALUES(kibana.alert.rule.name),
        Esql.message_values = VALUES(message),
        Esql.event_category_values = VALUES(event.category),
        Esql.event_action_values = VALUES(event.action),
        Esql.source_ip_values = VALUES(source.ip),
        Esql.destination_ip_values = VALUES(destination.ip),
        Esql.host_id_values = VALUES(host.id),
        Esql.agent_id_values = VALUES(agent.id),
        Esql.rule_severity_values = VALUES(kibana.alert.risk_score),
        Esql.user_name_values = VALUES(user.name) by Esql.time_window_date_trunc

// filter for alerts from different integrations with unique categories
| where Esql.event_module_distinct_count >= 2 and Esql.user_name_distinct_count >= 2 and Esql.event_category_distinct_count >= 2

// build context for LLM analysis
| eval users_list = MV_CONCAT(Esql.user_name_values, ",")

// LLM analysis
| eval instructions = "Analyze the provided user names and return a boolean value true if at least 2 of them are similar and they may belong to the same human identity or false if not, do not compare user names that may look like service accounts. If the list of users has more than 2 users and only 2 of them are similar consider this as true. Structure the output as follows: verdict=<verdict> confidence=<score between 0.0 and 1.0> summary=<short reason max 500 words> without any other response statements on a single line."
| eval prompt = CONCAT("User identities extracted from different alerts: ", users_list, instructions)
| COMPLETION triage_result = prompt WITH { "inference_id": ".anthropic-claude-4.6-sonnet-completion"}

// parse LLM response
| DISSECT triage_result """verdict=%{Esql.verdict} confidence=%{Esql.confidence} summary=%{Esql.summary}"""

// filter for similar user values
| where TO_LOWER(Esql.verdict) == "true"
| keep Esql.*