✄𐘗
text code block:
✄𐘗from logs-network_traffic.dns-*, logs-zeek.dns-*, packetbeat-*
| where
    (
      data_stream.dataset in ("network_traffic.dns", "zeek.dns")
      or event.dataset == "dns"
    )
    and dns.question.name is not null
    and dns.question.registered_domain is not null
| eval
    Esql.client_ip = COALESCE(client.ip, source.ip),
    Esql.dataset = COALESCE(data_stream.dataset, event.dataset),
    Esql.dns_question_name = TO_LOWER(dns.question.name),
    Esql.dns_registered_domain = TO_LOWER(dns.question.registered_domain),
    Esql.dns_question_type = TO_LOWER(dns.question.type),
    Esql.subdomain_length = LENGTH(Esql.dns_question_name) - LENGTH(Esql.dns_registered_domain) - 1
| where
    Esql.client_ip is not null
    and Esql.subdomain_length >= 50
    and (Esql.dns_question_type is null or Esql.dns_question_type != "ptr")
    and not ENDS_WITH(Esql.dns_question_name, ".arpa")
| eval Esql.time_window = DATE_TRUNC(5 minutes, @timestamp)
| stats
    Esql.count_queries = COUNT(*),
    Esql.count_distinct_names = COUNT_DISTINCT(Esql.dns_question_name),
    Esql.max_subdomain_length = MAX(Esql.subdomain_length),
    Esql.avg_subdomain_length = AVG(Esql.subdomain_length),
    Esql.dns_question_type_values = MV_SLICE(VALUES(Esql.dns_question_type), 0, 9),
    Esql.destination_ip_values = MV_SLICE(VALUES(destination.ip), 0, 4),
    Esql.sample_names = MV_SLICE(VALUES(Esql.dns_question_name), 0, 4),
    Esql.dataset_values = MV_SLICE(VALUES(Esql.dataset), 0, 9),
    Esql.observer_name_values = MV_SLICE(VALUES(observer.name), 0, 19),
    Esql.first_seen = MIN(@timestamp),
    Esql.last_seen = MAX(@timestamp)
  by Esql.time_window, Esql.client_ip, Esql.dns_registered_domain
| where Esql.count_queries >= 25 and Esql.count_distinct_names >= 15
| eval Esql.unique_name_ratio = TO_DOUBLE(Esql.count_distinct_names) / Esql.count_queries
| keep Esql.*