✄𐘗
text code block:
✄𐘗FROM .alerts-security.*
| WHERE event.kind == "signal"
    AND data_stream.namespace IS NOT NULL
    AND (
      (kibana.alert.rule.rule_id == "dfe3f626-4224-417e-aff1-8ef9a72c3191" AND observer.ip IS NOT NULL)
      OR
      (source.ip IS NOT NULL
        AND kibana.alert.rule.name IS NOT NULL
        AND kibana.alert.rule.rule_id IS NOT NULL
        AND kibana.alert.rule.rule_id != "dfe3f626-4224-417e-aff1-8ef9a72c3191"
        AND (kibana.alert.risk_score >= 47 OR kibana.alert.severity IN ("medium", "high", "critical"))
        AND KQL("""NOT kibana.alert.building_block_type : *""")
        AND NOT kibana.alert.rule.type IN ("machine_learning", "new_terms", "threat_match")
        AND NOT kibana.alert.rule.name LIKE "Deprecated - *"
        AND NOT KQL("""kibana.alert.rule.tags : "Rule Type: Higher-Order Rule" """))
    )
| EVAL
    Esql.is_exporter_alert = kibana.alert.rule.rule_id == "dfe3f626-4224-417e-aff1-8ef9a72c3191",
    Esql.is_suspicious_source_alert = kibana.alert.rule.rule_id != "dfe3f626-4224-417e-aff1-8ef9a72c3191",
    Esql.correlation_ip = CASE(Esql.is_exporter_alert, observer.ip, source.ip),
    Esql.exporter_alert_timestamp = CASE(Esql.is_exporter_alert, @timestamp, null),
    Esql.suspicious_source_alert_timestamp = CASE(Esql.is_suspicious_source_alert, @timestamp, null),
    Esql.suspicious_rule_name = CASE(Esql.is_suspicious_source_alert, kibana.alert.rule.name, null),
    Esql.suspicious_rule_id = CASE(Esql.is_suspicious_source_alert, kibana.alert.rule.rule_id, null)
| WHERE Esql.correlation_ip IS NOT NULL
| STATS
    Esql.exporter_alert_count = SUM(CASE(Esql.is_exporter_alert, 1, 0)),
    Esql.suspicious_source_alert_count = SUM(CASE(Esql.is_suspicious_source_alert, 1, 0)),
    observer.ip = MAX(CASE(Esql.is_exporter_alert, Esql.correlation_ip, null)),
    source.ip = MAX(CASE(Esql.is_suspicious_source_alert, Esql.correlation_ip, null)),
    Esql.exporter_alert_timestamp = MIN(Esql.exporter_alert_timestamp),
    Esql.suspicious_source_alert_timestamp = MAX(Esql.suspicious_source_alert_timestamp),
    Esql.suspicious_rule_name_values = VALUES(Esql.suspicious_rule_name),
    Esql.suspicious_rule_id_values = VALUES(Esql.suspicious_rule_id)
  BY data_stream.namespace, Esql.correlation_ip
| EVAL Esql.time_diff_seconds = DATE_DIFF(
    "second", Esql.exporter_alert_timestamp, Esql.suspicious_source_alert_timestamp
  )
| WHERE Esql.exporter_alert_count > 0
    AND Esql.suspicious_source_alert_count > 0
    AND Esql.time_diff_seconds >= 0
    AND Esql.time_diff_seconds <= 1800
| KEEP
    data_stream.namespace,
    observer.ip,
    source.ip,
    Esql.correlation_ip,
    Esql.exporter_alert_count,
    Esql.suspicious_source_alert_count,
    Esql.exporter_alert_timestamp,
    Esql.suspicious_source_alert_timestamp,
    Esql.suspicious_rule_name_values,
    Esql.suspicious_rule_id_values,
    Esql.time_diff_seconds