✄𐘗
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 and
    dns.resolved_ip is not null and
    TO_UPPER(dns.response_code) == "NOERROR" and
    TO_UPPER(dns.question.type) in ("A", "AAAA")
| eval
    Esql.client_ip = COALESCE(client.ip, source.ip),
    Esql.dataset = COALESCE(data_stream.dataset, event.dataset)
| where Esql.client_ip is not null
| mv_expand dns.resolved_ip
| eval Esql.is_private = CIDR_MATCH(
    dns.resolved_ip,
    "0.0.0.0/32",
    "10.0.0.0/8",
    "100.64.0.0/10",
    "127.0.0.0/8",
    "169.254.0.0/16",
    "172.16.0.0/12",
    "192.168.0.0/16",
    "::1/128",
    "fc00::/7",
    "fe80::/10"
  )
| eval
    Esql.private_time = CASE(Esql.is_private, @timestamp, null),
    Esql.public_time = CASE(not Esql.is_private, @timestamp, null),
    Esql.private_event_ttl = CASE(Esql.is_private, MV_MIN(dns.answers.ttl), null),
    Esql.private_ip = CASE(Esql.is_private, dns.resolved_ip, null),
    Esql.public_ip = CASE(not Esql.is_private, dns.resolved_ip, null)
| stats
    Esql.resolved_ip_observation_count = COUNT(*),
    Esql.resolved_ip_count = COUNT_DISTINCT(dns.resolved_ip),
    Esql.first_public_answer = MIN(Esql.public_time),
    Esql.first_private_answer = MIN(Esql.private_time),
    Esql.min_private_event_ttl = MIN(Esql.private_event_ttl),
    Esql.public_ips = MV_SLICE(VALUES(Esql.public_ip), 0, 100),
    Esql.private_ips = MV_SLICE(VALUES(Esql.private_ip), 0, 100),
    Esql.dataset_values = MV_SLICE(VALUES(Esql.dataset), 0, 10),
    Esql.observer_name_values = MV_SLICE(VALUES(observer.name), 0, 20)
  by Esql.client_ip, dns.question.name, dns.question.registered_domain
| eval
    Esql.same_timestamp = Esql.first_public_answer == Esql.first_private_answer,
    Esql.transition_seconds = DATE_DIFF("seconds", Esql.first_public_answer, Esql.first_private_answer)
| where
    Esql.first_public_answer is not null and
    Esql.first_private_answer is not null and
    (
        Esql.same_timestamp or
        (
            Esql.first_public_answer < Esql.first_private_answer and
            Esql.min_private_event_ttl is not null and
            Esql.min_private_event_ttl <= 60 and
            Esql.transition_seconds <= 300
        )
    )
| keep Esql.*, dns.*