✄𐘗
text code block:
✄𐘗FROM logs-okta.system-* METADATA _id, _version, _index
| WHERE event.dataset == "okta.system"
    AND (event.action LIKE "user.authentication.*" OR event.action == "user.session.start")
    AND okta.outcome.reason IN ("INVALID_CREDENTIALS", "LOCKED_OUT")
    AND okta.actor.alternate_id IS NOT NULL

// Create source mapping for analyst context
| EVAL Esql.source_info = CONCAT(
    "{\"ip\":\"", COALESCE(okta.client.ip::STRING, "unknown"),
    "\",\"country\":\"", COALESCE(client.geo.country_name, "unknown"),
    "\",\"asn\":\"", COALESCE(source.as.organization.name, "unknown"),
    "\",\"user_agent\":\"", COALESCE(okta.client.user_agent.raw_user_agent, "unknown"), "\"}"
  )

| STATS
    Esql.unique_source_ips = COUNT_DISTINCT(okta.client.ip),
    Esql.total_attempts = COUNT(*),
    Esql.unique_user_agents = COUNT_DISTINCT(okta.client.user_agent.raw_user_agent),
    Esql.unique_dt_hashes = COUNT_DISTINCT(okta.debug_context.debug_data.dt_hash),
    Esql.unique_asns = COUNT_DISTINCT(source.as.number),
    Esql.unique_countries = COUNT_DISTINCT(client.geo.country_name),
    Esql.first_seen = MIN(@timestamp),
    Esql.last_seen = MAX(@timestamp),
    Esql.source_ip_values = VALUES(okta.client.ip),
    Esql.source_mapping = VALUES(Esql.source_info),
    Esql.event_action_values = VALUES(event.action),
    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),
    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.threat_suspected_values = VALUES(okta.debug_context.debug_data.threat_suspected),
    Esql.risk_level_values = VALUES(okta.debug_context.debug_data.risk_level),
    Esql.risk_reasons_values = VALUES(okta.debug_context.debug_data.risk_reasons)
  BY okta.actor.alternate_id

| EVAL
    Esql.attempts_per_ip = Esql.total_attempts * 1.0 / Esql.unique_source_ips,
    Esql.duration_seconds = DATE_DIFF("seconds", Esql.first_seen, Esql.last_seen)

| WHERE
    Esql.unique_source_ips >= 5
    AND Esql.total_attempts >= 10
    AND (
        Esql.unique_countries >= 2 OR
        Esql.unique_asns >= 3 OR
        Esql.unique_source_ips >= 8 OR
        Esql.unique_user_agents >= 3
    )

| SORT Esql.unique_source_ips DESC
| KEEP Esql.*, okta.actor.alternate_id