Lateral Movement Alerts from a Newly Observed Source Address

FROM .alerts-security.* METADATA _index

// Lateral Movement related rules with fields of interest
| where kibana.alert.rule.threat.tactic.name is not null and
        source.ip IS NOT NULL and destination.ip is not null and
        host.id is not null and KQL("""kibana.alert.rule.threat.tactic.name : "Lateral Movement" and not kibana.alert.rule.tags : "Rule Type: Higher-Order Rule" """)

// aggregate stats by source.ip
| 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.user_name_values = VALUES(user.name),
         Esql.host_id_values = VALUES(host.id),
         Esql.host_ip_values = VALUES(host.ip),
         Esql.tactic_name_values = VALUES(kibana.alert.rule.threat.tactic.name) by source.ip

// values we will need for next filter
| eval isLocal = locate(MV_CONCAT(to_string(Esql.host_ip_values), ","), to_string(source.ip)),
       Esql.date_diff = DATE_DIFF("minute", Esql.first_time_seen, now())

// at least 2 unique rules from same source.ip and that was first seen in last 5 days
| where Esql.unique_rules_count >= 2 and
        // matches are within 10m of the rule execution time to avoid alert duplicates
        Esql.date_diff <= 10 and
        // make sure source.ip is not equal to host.ip
        not isLocal > 0 and
        // reduce noise from SCCM, Nessus and alike
        Esql.unique_count_host_id <= 3 and Esql.alerts_count <= 20
| eval host.id = MV_FIRST(Esql.host_id_values), user.name = MV_FIRST(Esql.user_name_values)
| KEEP Esql.*, source.ip, host.id, user.name