LoFP / cdn, cloud load-balancer, software-update, and telemetry hostnames can be long and change often. recursive resolvers, forwarders, nat gateways, and localhost dns listeners can also combine queries from many endpoints under one client address. dns research, testing, and security tools can generate null or unusually long txt queries. validate the apex domain and whether the source is an endpoint before treating the activity as tunneling.

from logs-network_traffic.dns-*, logs-fortinet_fortigate.log-*, logs-zeek.dns-*, packetbeat-*
| where
    (
      data_stream.dataset in ("network_traffic.dns", "fortinet_fortigate.log", "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.question_name_length = LENGTH(Esql.dns_question_name),
    Esql.subdomain_length = LENGTH(Esql.dns_question_name) - LENGTH(Esql.dns_registered_domain) - 1
| where
    Esql.client_ip is not null
    and (Esql.dns_question_type is null or Esql.dns_question_type != "ptr")
    and not ENDS_WITH(Esql.dns_question_name, ".arpa")
    and (
      Esql.dns_question_type == "null"
      or Esql.subdomain_length >= 50
      or (Esql.dns_question_type == "txt" and Esql.question_name_length > 100)
    )
| eval
    Esql.long_subdomain_query_name = CASE(
      Esql.subdomain_length >= 50,
      Esql.dns_question_name,
      null
    ),
    Esql.null_query_name = CASE(
      Esql.dns_question_type == "null",
      Esql.dns_question_name,
      null
    ),
    Esql.long_txt_query_name = CASE(
      Esql.dns_question_type == "txt" and Esql.question_name_length > 100,
      Esql.dns_question_name,
      null
    )
| eval Esql.time_window = DATE_TRUNC(5 minutes, @timestamp)
| stats
    Esql.total_qualifying_queries = COUNT(*),
    Esql.count_queries = COUNT(Esql.long_subdomain_query_name),
    Esql.count_distinct_names = COUNT_DISTINCT(Esql.long_subdomain_query_name),
    Esql.null_query_count = COUNT(Esql.null_query_name),
    Esql.null_distinct_names = COUNT_DISTINCT(Esql.null_query_name),
    Esql.long_txt_query_count = COUNT(Esql.long_txt_query_name),
    Esql.long_txt_distinct_names = COUNT_DISTINCT(Esql.long_txt_query_name),
    Esql.max_question_name_length = MAX(Esql.question_name_length),
    Esql.avg_question_name_length = AVG(Esql.question_name_length),
    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
| eval Esql.unique_name_ratio = CASE(
    Esql.count_queries > 0,
    TO_DOUBLE(Esql.count_distinct_names) / Esql.count_queries,
    null
  )
| where
    (
      Esql.count_queries >= 25
      and Esql.count_distinct_names >= 15
    )
    or (
      Esql.null_query_count >= 3
      and Esql.null_distinct_names >= 2
    )
    or (
      Esql.long_txt_query_count >= 5
      and Esql.long_txt_distinct_names >= 5
    )
| keep Esql.*