✄𐘗
text code block:
✄𐘗from logs-network_traffic.sip-*, packetbeat-* metadata _source
| eval
    Esql.method = TO_UPPER(COALESCE(
        JSON_EXTRACT(_source, "network_traffic.sip.cseq.method"),
        JSON_EXTRACT(_source, "sip.cseq.method"),
        JSON_EXTRACT(_source, "network_traffic.sip.method"),
        JSON_EXTRACT(_source, "sip.method")
    )),
    Esql.sip_type = TO_LOWER(COALESCE(
        JSON_EXTRACT(_source, "network_traffic.sip.type"),
        JSON_EXTRACT(_source, "sip.type")
    )),
    Esql.code = COALESCE(
        JSON_EXTRACT(_source, "network_traffic.sip.code"),
        JSON_EXTRACT(_source, "sip.code")
    ),
    Esql.client_ip = COALESCE(client.ip, destination.ip),
    Esql.server_ip = COALESCE(server.ip, source.ip),
    Esql.extension = COALESCE(
        JSON_EXTRACT(_source, "network_traffic.sip.to.uri.username"),
        JSON_EXTRACT(_source, "sip.to.uri.username")
    )
| where
    Esql.method == "REGISTER" and
    Esql.sip_type == "response" and
    Esql.code in ("401", "403", "407") and
    Esql.client_ip is not null and
    Esql.server_ip is not null and
    Esql.extension is not null
| eval Esql.time_window = DATE_TRUNC(5 minutes, @timestamp)
| stats
    Esql.failures_per_extension = COUNT(*)
  by Esql.time_window, Esql.client_ip, Esql.server_ip, Esql.extension
| eval Esql.repeated_extension = CASE(Esql.failures_per_extension >= 2, 1, 0)
| stats
    Esql.total_failures = SUM(Esql.failures_per_extension),
    Esql.max_failures_per_extension = MAX(Esql.failures_per_extension),
    Esql.distinct_extensions = COUNT_DISTINCT(Esql.extension),
    Esql.repeated_extensions = SUM(Esql.repeated_extension),
    Esql.sample_extensions = MV_SLICE(VALUES(Esql.extension), 0, 20)
  by Esql.time_window, Esql.client_ip, Esql.server_ip
| where
    Esql.max_failures_per_extension >= 10 or
    (
      Esql.total_failures >= 25 and
      Esql.distinct_extensions >= 5 and
      Esql.repeated_extensions >= 5
    )
| keep Esql.*