LoFP / automated password reset flows where a user fails multiple times then succeeds after resetting their password.

FROM .alerts-security.*, logs-okta.system-* METADATA _id, _version, _index
// Filter for credential attack alerts OR successful Okta authentications
| WHERE
    (
        // Credential attack alerts from the five correlated rules
        kibana.alert.rule.rule_id IN (
            "94e734c0-2cda-11ef-84e1-f661ea17fbce",  // Credential Stuffing
            "42bf698b-4738-445b-8231-c834ddefd8a0",  // Password Spraying
            "23f18264-2d6d-11ef-9413-f661ea17fbce",  // DT Brute Force
            "5889760c-9858-4b4b-879c-e299df493295",  // Distributed Brute Force
            "2d3c27d5-d133-4152-8102-8d051619ec4a"   // Distributed Spray
        )
    )
    OR (
        // Successful Okta authentication events
        event.dataset == "okta.system"
        AND (event.action LIKE "user.authentication.*" OR event.action == "user.session.start")
        AND okta.outcome.result == "SUCCESS"
        AND okta.actor.alternate_id IS NOT NULL
    )
// correlation - alerts may store user/IP in different fields than raw logs
| EVAL
    Esql.user = COALESCE(okta.actor.alternate_id, user.name, user.email),
    Esql.source_ip = COALESCE(okta.client.ip, client.ip, source.ip)
// Must have user identity to correlate
| WHERE Esql.user IS NOT NULL
// Classify events and capture timestamps/IPs by event type
| EVAL
    Esql.is_attack_alert = CASE(
        kibana.alert.rule.rule_id IN (
            "94e734c0-2cda-11ef-84e1-f661ea17fbce",
            "42bf698b-4738-445b-8231-c834ddefd8a0",
            "23f18264-2d6d-11ef-9413-f661ea17fbce",
            "5889760c-9858-4b4b-879c-e299df493295",
            "2d3c27d5-d133-4152-8102-8d051619ec4a"
        ), 1, 0
    ),
    Esql.is_success_login = CASE(
        event.dataset == "okta.system"
        AND okta.outcome.result == "SUCCESS", 1, 0
    ),
    Esql.attack_ip = CASE(
        kibana.alert.rule.rule_id IN (
            "94e734c0-2cda-11ef-84e1-f661ea17fbce",
            "42bf698b-4738-445b-8231-c834ddefd8a0",
            "23f18264-2d6d-11ef-9413-f661ea17fbce",
            "5889760c-9858-4b4b-879c-e299df493295",
            "2d3c27d5-d133-4152-8102-8d051619ec4a"
        ), Esql.source_ip, null
    ),
    Esql.login_ip = CASE(
        event.dataset == "okta.system"
        AND okta.outcome.result == "SUCCESS", Esql.source_ip, null
    ),
    Esql.attack_ts = CASE(
        kibana.alert.rule.rule_id IN (
            "94e734c0-2cda-11ef-84e1-f661ea17fbce",
            "42bf698b-4738-445b-8231-c834ddefd8a0",
            "23f18264-2d6d-11ef-9413-f661ea17fbce",
            "5889760c-9858-4b4b-879c-e299df493295",
            "2d3c27d5-d133-4152-8102-8d051619ec4a"
        ), @timestamp, null
    ),
    Esql.login_ts = CASE(
        event.dataset == "okta.system"
        AND okta.outcome.result == "SUCCESS", @timestamp, null
    )
// Aggregate by user (catches IP rotation: spray from IP A, login from IP B)
| STATS
    Esql.attack_count = SUM(Esql.is_attack_alert),
    Esql.login_count = SUM(Esql.is_success_login),
    Esql.earliest_attack = MIN(Esql.attack_ts),
    Esql.latest_attack = MAX(Esql.attack_ts),
    Esql.earliest_login = MIN(Esql.login_ts),
    Esql.latest_login = MAX(Esql.login_ts),
    Esql.attack_source_ips = VALUES(Esql.attack_ip),
    Esql.login_source_ips = VALUES(Esql.login_ip),
    Esql.all_source_ips = VALUES(Esql.source_ip),
    Esql.alert_rule_ids = VALUES(kibana.alert.rule.rule_id),
    Esql.alert_rule_names = VALUES(kibana.alert.rule.name),
    Esql.event_action_values = VALUES(event.action),
    Esql.geo_country_values = VALUES(client.geo.country_name),
    Esql.geo_city_values = VALUES(client.geo.city_name),
    Esql.source_asn_values = VALUES(source.as.number),
    Esql.source_asn_org_values = VALUES(source.as.organization.name),
    Esql.user_agent_values = VALUES(okta.client.user_agent.raw_user_agent),
    Esql.device_values = VALUES(okta.client.device),
    Esql.is_proxy_values = VALUES(okta.security_context.is_proxy)
  BY Esql.user
// Calculate time gap between latest attack and earliest subsequent login
| EVAL Esql.attack_to_login_minutes = DATE_DIFF("minute", Esql.latest_attack, Esql.earliest_login)
// Correlation: attack BEFORE login + success within reasonable window (3 hours)
| WHERE
    Esql.attack_count > 0
    AND Esql.login_count > 0
    AND Esql.latest_attack < Esql.earliest_login
    AND Esql.attack_to_login_minutes <= 180
| SORT Esql.login_count DESC
| KEEP Esql.*