Suspected Lateral Movement from Compromised Host

from .alerts-security.*

// any alerts excluding deprecated, low severity and threat_match rules
| where kibana.alert.rule.name is not null and kibana.alert.risk_score > 21 and
        kibana.alert.rule.type != "threat_match" and
        not kibana.alert.rule.name like "Deprecated - *" and
        not KQL("""kibana.alert.rule.tags : "Rule Type: Higher-Order Rule" """) and
        not kibana.alert.rule.name in ("Abnormally Large DNS Response", "Web Application Suspicious Activity: No User Agent")

// alerts with existing source.ip or host.ip
| eval alert_source_ip = CASE(source.ip is not null, source.ip, null),
       alert_host_ip = CASE(host.ip is not null and source.ip is null, host.ip, null)
| eval Esql.source_ip = COALESCE(alert_source_ip, alert_host_ip)
| where Esql.source_ip is not null and Esql.source_ip != "127.0.0.1" and Esql.source_ip != "::1"
| stats Esql.alerts_count = COUNT(*),
        Esql.event_module_distinct_count = COUNT_DISTINCT(event.module),
        Esql.host_id_distinct_count = COUNT_DISTINCT(host.id),
        Esql.rule_name_distinct_count = COUNT_DISTINCT(kibana.alert.rule.name),
        Esql.event_module_values = VALUES(event.module),
        Esql.message_values = VALUES(message),
        Esql.rule_name = VALUES(kibana.alert.rule.name),
        Esql.event_action_values = VALUES(event.action),
        Esql.event_category_values = VALUES(event.category),
        Esql.process_executable_values = VALUES(process.executable),
        Esql.process_cmdline_values = VALUES(process.command_line),
        Esql.file_path_values = VALUES(file.path),
        Esql.host_id_values = VALUES(host.id),
        Esql.host_ip_values = VALUES(host.ip),
        Esql.destination_ip_values = VALUES(destination.ip),
        Esql.user_name_values = VALUES(user.name),
        SRC_IP = VALUES(source.ip)
        by Esql.source_ip

// filter for different alerts from multiple hosts and where the host.ip of one alert matches the source.ip of the other alert
| eval concat_ip_values = MV_CONCAT(TO_STRING(Esql.host_ip_values), ",")
| eval host_ip_equal_to_source_ip =LOCATE(concat_ip_values, TO_STRING(Esql.source_ip))
| where Esql.rule_name_distinct_count >= 2 and Esql.host_id_distinct_count >= 2 and host_ip_equal_to_source_ip > 0 and SRC_IP is not null and Esql.alerts_count <= 100

// Move single values to their corresponding ECS fields for alerts exclusion
| eval source.ip = mv_min(Esql.source_ip),
       host.id = mv_min(Esql.host_id_values)
| KEEP Esql.*, source.ip, host.id