✄𐘗
text code block:
✄𐘗from logs-o365.audit-*, logs-azure.signinlogs-*, .alerts-security.*
// query runs every 1 hour looking for activities occurred during last 8 hours to match on disparate events
| where @timestamp > now() - 8 hours
// filter for azure or m365 sign-in and external alerts with source.ip not null
| where to_ip(source.ip) is not null
  and (event.dataset in ("o365.audit", "azure.signinlogs") or kibana.alert.rule.rule_id == "eb079c62-4481-4d6e-9643-3ca499df7aaa")
  and not cidr_match(
    to_ip(source.ip),
    "10.0.0.0/8", "127.0.0.0/8", "169.254.0.0/16", "172.16.0.0/12", "192.0.0.0/24", "192.0.0.0/29",
    "192.0.0.8/32", "192.0.0.9/32", "192.0.0.10/32", "192.0.0.170/32", "192.0.0.171/32", "192.0.2.0/24",
    "192.31.196.0/24", "192.52.193.0/24", "192.168.0.0/16", "192.88.99.0/24", "224.0.0.0/4",
    "100.64.0.0/10", "192.175.48.0/24", "198.18.0.0/15", "198.51.100.0/24", "203.0.113.0/24",
    "240.0.0.0/4", "::1", "FE80::/10", "FF00::/8"
  )

// capture relevant raw fields
| keep source.ip, event.action, event.outcome, event.dataset, kibana.alert.rule.rule_id, event.category

// classify each source ip based on alert type
| eval
  Esql.source_ip_mail_access_case = case(event.dataset == "o365.audit" and event.action == "MailItemsAccessed" and event.outcome == "success", to_ip(source.ip), null),
  Esql.source_ip_azure_signin_case = case(event.dataset == "azure.signinlogs" and event.outcome == "success", to_ip(source.ip), null),
  Esql.source_ip_network_alert_case = case(kibana.alert.rule.rule_id == "eb079c62-4481-4d6e-9643-3ca499df7aaa" and not event.dataset in ("o365.audit", "azure.signinlogs"), to_ip(source.ip), null)

// aggregate by source ip
| stats
    Esql.event_count = count(*),
    Esql.source_ip_mail_access_case_count_distinct = count_distinct(Esql.source_ip_mail_access_case),
    Esql.source_ip_azure_signin_case_count_distinct = count_distinct(Esql.source_ip_azure_signin_case),
    Esql.source_ip_network_alert_case_count_distinct = count_distinct(Esql.source_ip_network_alert_case),
    Esql.event_dataset_count_distinct = count_distinct(event.dataset),
    Esql.event_dataset_values = values(event.dataset),
    Esql.kibana_alert_rule_id_values = values(kibana.alert.rule.rule_id),
    Esql.event_category_values = values(event.category)
  by Esql.source_ip = to_ip(source.ip)

// correlation condition
| where
  Esql.source_ip_network_alert_case_count_distinct > 0
  and Esql.event_dataset_count_distinct >= 2
  and (Esql.source_ip_mail_access_case_count_distinct > 0 or Esql.source_ip_azure_signin_case_count_distinct > 0)
  and Esql.event_count <= 100