Lateral Movement Alerts from a Newly Observed User

FROM .alerts-security.* METADATA _index

// Lateral Movement related rules
| where kibana.alert.rule.threat.tactic.name is not null and user.id is not null and 
        (to_string(user.id) like "S-1-5-21*" or to_string(user.id) like "S-1-12-*") and
        host.id is not null and KQL("""kibana.alert.rule.threat.tactic.name : "Lateral Movement" """) and
        not KQL("""kibana.alert.rule.tags : "Rule Type: Higher-Order Rule" """)

// aggregate stats by user.id
| stats  Esql.first_time_seen = MIN(@timestamp),
         Esql.alerts_count = count(*),
         Esql.unique_rules_count = COUNT_DISTINCT(kibana.alert.rule.name),
         Esql.unique_count_host_id = COUNT_DISTINCT(host.id),
         Esql.rule_name_values = VALUES(kibana.alert.rule.name),
         Esql.host_id_values = VALUES(host.id),
         Esql.host_ip_values = VALUES(host.ip),
         Esql.source_ip_values = VALUES(source.ip),
         Esql.process_cmd_line = VALUES(process.command_line),
         Esql.tactic_name_values = VALUES(kibana.alert.rule.threat.tactic.name) by user.id, user.name

// at least 2 unique lateral movement detection rules from same user.id and that was first seen in last 5 days
| eval Esql.date_diff = DATE_DIFF("minute", Esql.first_time_seen, now())
| where Esql.unique_rules_count >= 2 and
        // matches are within 10m of the rule execution time to avoid alert duplicates
        Esql.date_diff <= 10
| eval source.ip = MV_FIRST(Esql.source_ip_values),  host.id = MV_FIRST(Esql.host_id_values)
| KEEP Esql.*, user.id, user.name, host.id, source.ip