Elastic Defend and Network Security Alerts Correlation

FROM logs-* metadata _id
| WHERE
        // Elastic Defend Alerts
        (event.module == "endpoint" and data_stream.dataset == "endpoint.alerts") or

        // PANW suspicious events
        (data_stream.dataset == "panw.panos" and
         event.action in ("virus_detected", "wildfire_virus_detected", "c2_communication", "spyware_detected", "large_upload", "denied", "exploit_detected")) or

        // Fortigate suspicious events
        (data_stream.dataset == "fortinet_fortigate.log" and
         (event.action in ("outbreak-prevention", "infected", "blocked") or message like "backdoor*" or message like "Proxy*" or message like "anomaly*" or message like "P2P*" or message like "misc*" or message like "DNS.Over.HTTPS" or message like "Remote.Access")) or

        // Suricata
        (data_stream.dataset == "suricata.eve" and message in ("Command and Control Traffic", "Potentially Bad Traffic", "A Network Trojan was detected", "Detection of a Network Scan", "Domain Observed Used for C2 Detected", "Malware Command and Control Activity Detected"))

// extract source.ip from PANW or Fortigate events and host.ip from Elastic Defend alert
| eval fw_alert_source_ip = CASE(data_stream.dataset in ("panw.panos", "fortinet_fortigate.log"), source.ip, null),
      elastic_defend_alert_host_ip = CASE(event.module == "endpoint" and data_stream.dataset == "endpoint.alerts", host.ip, null)
| eval Esql.source_ip = COALESCE(fw_alert_source_ip, elastic_defend_alert_host_ip)
| where Esql.source_ip is not null

// group by host_source_ip shared between FG/PANW and Elastic Defend
| stats Esql.alerts_count = COUNT(*),
        Esql.event_module_distinct_count = COUNT_DISTINCT(event.module),
        Esql.message_values_distinct_count = COUNT_DISTINCT(message),
        Esql.event_module_values = VALUES(event.module),
        Esql.message_values = VALUES(message),
        Esql.event_action_values = VALUES(event.action),
        Esql.process_executable_values = VALUES(process.executable),
        Esql.process_hash_sha256_values = VALUES(process.hash.sha256),
        Esql.process_cmdline_values = VALUES(process.command_line),
        Esql.file_path_values = VALUES(file.path),
        Esql.file_hash_sha256_values = VALUES(file.hash.sha256),
        Esql.host_id_values = VALUES(host.id),
        Esql.user_name_values = VALUES(user.name),
        Esql.destination_ip_values = VALUES(destination.ip)
        by Esql.source_ip
| where Esql.event_module_distinct_count >= 2 
AND Esql.message_values_distinct_count >= 2
| eval concat_module_values = MV_CONCAT(Esql.event_module_values, ",")
// Make sure an endpoint alert is present along one of the network ones
| where concat_module_values like "*endpoint*"

// 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),
       user.name = mv_min(Esql.user_name_values)
| keep source.ip, host.id, user.name, Esql.*