runbound 0.5.1

// SPDX-License-Identifier: AGPL-3.0-or-later
// Copyright (C) 2024-2026 RedLemonBe — https://github.com/redlemonbe/Runbound
// Runbound DNS server — drop-in for Unbound.
//
// Architecture:
//   1. Access-control list check (per source IP, from unbound.conf)
//   2. Rate limiting (per source IP token bucket)
//   3. Check local zones (local-data, blacklist, feeds) in memory → instant
//   4. Otherwise → recursive resolver (hickory-resolver)
//
// UDP + TCP on the configured port (default 53).

use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::str::FromStr;
use std::sync::{Arc, OnceLock};
use std::sync::atomic::{AtomicU16, Ordering};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use hickory_proto::op::{Metadata, ResponseCode};
use hickory_proto::rr::{LowerName, Name, RData, Record, RecordType};
use hickory_resolver::{
    TokioResolver,
    config::{ConnectionConfig, NameServerConfig, ResolveHosts, ResolverConfig, ResolverOpts},
    net::runtime::TokioRuntimeProvider,
    net::{DnsError, NetError, NoRecords},
};
use hickory_server::{
    Server,
    zone_handler::MessageResponseBuilder,
    server::{Request, RequestHandler, ResponseHandler, ResponseInfo},
    net::runtime::Time,
};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use arc_swap::ArcSwap;
use dashmap::DashMap;
use tokio::net::{TcpListener, TcpStream, UdpSocket};
use tokio::sync::Semaphore;
use tracing::{debug, error, info, warn};

use crate::config::parser::TlsConfig;
use crate::config::parser::UnboundConfig;
use crate::logbuffer::{LogAction, SharedLogBuffer};
use crate::stats::{Stats, CACHE_HIT_THRESHOLD_US};
use super::acl::{Acl, AclAction, PrivateAddressSet};
use super::local::{LocalZoneSet, ZoneAction};
use super::ratelimit::RateLimiter;

// ── Concurrency cap — prevents OOM under flood ─────────────────────────────
//
// hickory-server spawns one tokio task per incoming DNS request with no
// backpressure. Under a flood (DDoS or perf test) this exhausts RAM.
// A non-blocking try_acquire returns REFUSED instantly without allocating
// any additional memory, so the bound is hard even at line rate.
const MAX_INFLIGHT_REQUESTS: usize = 4_096;

const RATE_LIMIT_QPS_DEFAULT: u64 = 200;

// ── Identity-probe name set (zero-alloc hot path) ──────────────────────────
// Initialised once on first DNS query; compared directly as LowerName.
// Avoids a String allocation per request for the CHAOS identity-probe check.
static IDENTITY_PROBE_NAMES: OnceLock<[LowerName; 4]> = OnceLock::new();

fn identity_probe_names() -> &'static [LowerName; 4] {
    IDENTITY_PROBE_NAMES.get_or_init(|| [
        LowerName::from(Name::from_str("version.bind.").expect("static DNS name")),
        LowerName::from(Name::from_str("hostname.bind.").expect("static DNS name")),
        LowerName::from(Name::from_str("id.server.").expect("static DNS name")),
        LowerName::from(Name::from_str("version.server.").expect("static DNS name")),
    ])
}

// ============================================================
// Handler
// ============================================================

pub struct RunboundHandler {
    pub zones:        Arc<ArcSwap<LocalZoneSet>>,
    resolver:         Arc<ArcSwap<TokioResolver>>,
    rate_limiter:     Arc<RateLimiter>,
    inflight:         Arc<Semaphore>,
    acl:              Arc<Acl>,
    private_addrs:    Arc<PrivateAddressSet>,
    cache_max_ttl:    u32,
    pub stats:        Arc<Stats>,
    pub log_buffer:   SharedLogBuffer,
    /// DNSSEC tracking enabled — mirrors `dnssec-validation: yes` in config.
    dnssec_enabled:   bool,
    dnssec_log_bogus: bool,
}

impl RunboundHandler {
    #[allow(clippy::too_many_arguments)]
    fn new(
        zones:            Arc<ArcSwap<LocalZoneSet>>,
        resolver:         Arc<ArcSwap<TokioResolver>>,
        rate_limiter:     Arc<RateLimiter>,
        acl:              Arc<Acl>,
        private_addrs:    Arc<PrivateAddressSet>,
        cache_max_ttl:    u32,
        stats:            Arc<Stats>,
        log_buffer:       SharedLogBuffer,
        dnssec_enabled:   bool,
        dnssec_log_bogus: bool,
    ) -> Self {
        Self {
            zones, resolver, rate_limiter,
            inflight: Arc::new(Semaphore::new(MAX_INFLIGHT_REQUESTS)),
            acl, private_addrs, cache_max_ttl, stats, log_buffer,
            dnssec_enabled, dnssec_log_bogus,
        }
    }

    /// Record a completed query: latency histogram, log buffer push, tracing log.
    /// Cache metrics (record_forward) and domain counters are updated at call sites.
    #[inline]
    fn record_query(
        &self,
        client:    IpAddr,
        qname:     &hickory_proto::rr::LowerName,
        qtype:     RecordType,
        rcode:     ResponseCode,
        action:    LogAction,
        start:     Instant,
    ) {
        let elapsed    = start.elapsed();
        let elapsed_us = elapsed.as_micros() as u64;
        let elapsed_ms = elapsed.as_millis() as u32;

        self.stats.record_latency_us(elapsed_us);
        if action == LogAction::Local {
            self.stats.inc_local_hits();
        }

        // MED-06: sanitize the DNS name before structured log emission to prevent
        // log injection via control characters embedded in query names.
        // Guard the allocation: skip when log buffer is disabled AND info-level tracing
        // is off (verbosity < 2). Hot path at verbosity:1 with log-retention:0 → zero alloc.
        if self.log_buffer.is_enabled() || tracing::enabled!(tracing::Level::INFO) {
            let safe_name = sanitize_dns_name(qname);
            let client_log = self.log_buffer.push_query(&safe_name, &client, u16::from(qtype), action, elapsed_ms);
            info!(
                client = %client_log,
                name   = %safe_name,
                qtype  = %qtype,
                rcode  = %rcode,
                action = action.as_str(),
                ms     = elapsed_ms,
                "query"
            );
        }
    }
}

impl RunboundHandler {
    /// Attempt to answer from local zones (blacklist, static data, CNAME chains).
    /// Returns `Ok(info)` when a response was sent; `Err(rh)` when no local match
    /// was found and the query should fall through to recursive resolution.
    async fn handle_local_zone<R: ResponseHandler>(
        &self,
        request: &Request,
        mut response_handle: R,
        qname: &LowerName,
        qtype: RecordType,
        client_ip: IpAddr,
        start: Instant,
    ) -> Result<ResponseInfo, R> {
        let zones_snap  = self.zones.load();
        let zone_action = zones_snap.find(qname);

        match zone_action {
            Some(ZoneAction::Refuse) => {
                debug!(name=%sanitize_dns_name(qname), "local-zone REFUSED");
                self.stats.inc_blocked(); self.stats.inc_refused();
                self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Blocked, start);
                return Ok(send_error(request, response_handle, ResponseCode::Refused).await);
            }
            Some(ZoneAction::NxDomain) => {
                debug!(name=%sanitize_dns_name(qname), "local-zone NXDOMAIN");
                self.stats.inc_blocked(); self.stats.inc_nxdomain();
                self.record_query(client_ip, qname, qtype, ResponseCode::NXDomain, LogAction::Blocked, start);
                return Ok(send_error(request, response_handle, ResponseCode::NXDomain).await);
            }
            Some(ZoneAction::Static) | Some(ZoneAction::Redirect) => {
                let records = zones_snap.local_records(qname, qtype);

                if !records.is_empty() {
                    debug!(name=%sanitize_dns_name(qname), count = records.len(), "local-data answer");
                    let mut metadata = Metadata::response_from_request(&request.metadata);
                    metadata.authoritative = true;
                    let builder = MessageResponseBuilder::from_message_request(request);
                    let response = builder.build(
                        metadata, records,
                        std::iter::empty(), std::iter::empty(), std::iter::empty(),
                    );
                    self.record_query(client_ip, qname, qtype, ResponseCode::NoError, LogAction::Local, start);
                    return Ok(response_handle.send_response(response).await
                        .unwrap_or_else(|e| { error!("send: {e}"); servfail_info(request) }));
                }

                // CNAME chain following (RFC 1034 §3.6.2)
                if qtype != RecordType::CNAME {
                    let chain = follow_local_cname(&zones_snap, qname, qtype);
                    if !chain.is_empty() {
                        let mut metadata = Metadata::response_from_request(&request.metadata);
                        metadata.authoritative = true;
                        let builder = MessageResponseBuilder::from_message_request(request);
                        let response = builder.build(
                            metadata, chain.iter(),
                            std::iter::empty(), std::iter::empty(), std::iter::empty(),
                        );
                        self.record_query(client_ip, qname, qtype, ResponseCode::NoError, LogAction::Local, start);
                        return Ok(response_handle.send_response(response).await
                            .unwrap_or_else(|e| { error!("send: {e}"); servfail_info(request) }));
                    }
                }

                // RFC 1035 §3.7 / RFC 2308: NODATA vs NXDOMAIN
                if zones_snap.name_has_records(qname) {
                    debug!(name=%sanitize_dns_name(qname), %qtype, "local-zone NODATA");
                    let mut metadata = Metadata::response_from_request(&request.metadata);
                    metadata.authoritative = true;
                    let builder = MessageResponseBuilder::from_message_request(request);
                    let response = builder.build(
                        metadata, std::iter::empty::<&Record>(),
                        std::iter::empty(), std::iter::empty(), std::iter::empty(),
                    );
                    self.record_query(client_ip, qname, qtype, ResponseCode::NoError, LogAction::Local, start);
                    return Ok(response_handle.send_response(response).await
                        .unwrap_or_else(|e| { error!("send: {e}"); servfail_info(request) }));
                }
                debug!(name=%sanitize_dns_name(qname), "local-zone NXDOMAIN (name not found)");
                self.record_query(client_ip, qname, qtype, ResponseCode::NXDomain, LogAction::Nxdomain, start);
                return Ok(send_error(request, response_handle, ResponseCode::NXDomain).await);
            }
            None => {}
        }
        // Reached only via the None arm — response_handle was not consumed.
        Err(response_handle)
    }

    /// Recursive upstream resolution with DNSSEC validation and rebinding protection.
    async fn resolve_upstream<R: ResponseHandler>(
        &self,
        request: &Request,
        mut response_handle: R,
        qname: &LowerName,
        qtype: RecordType,
        client_ip: IpAddr,
        start: Instant,
    ) -> ResponseInfo {
        match self.resolver.load().lookup(Name::from(qname), qtype).await {
            Ok(lookup) => {
                // DNS rebinding protection: SERVFAIL if any A/AAAA record falls
                // within a configured private-address range (Unbound compatible).
                if !self.private_addrs.is_empty() {
                    for rec in lookup.answers() {
                        let private_ip = match &rec.data {
                            RData::A(a)    => Some(IpAddr::V4(a.0)),
                            RData::AAAA(a) => Some(IpAddr::V6(a.0)),
                            _ => None,
                        };
                        if let Some(ip) = private_ip {
                            if self.private_addrs.contains(ip) {
                                warn!(name=%sanitize_dns_name(qname), %ip, "private-address block → SERVFAIL");
                                self.record_query(client_ip, qname, qtype, ResponseCode::ServFail, LogAction::Servfail, start);
                                return send_error(request, response_handle, ResponseCode::ServFail).await;
                            }
                        }
                    }
                }

                let ttl_cap = self.cache_max_ttl;
                let needs_cap = lookup.answers().iter().any(|r| r.ttl > ttl_cap);
                let capped: Vec<Record>;
                let records: &[Record] = if needs_cap {
                    capped = lookup.answers().iter().map(|r| {
                        if r.ttl > ttl_cap { let mut rc = r.clone(); rc.ttl = ttl_cap; rc }
                        else { r.clone() }
                    }).collect();
                    &capped
                } else {
                    lookup.answers()
                };

                // DNSSEC: check for bogus proof on individual records in success path.
                if self.dnssec_enabled {
                    let has_bogus = records.iter().any(|r| r.proof.is_bogus());
                    if has_bogus {
                        self.stats.inc_dnssec_bogus();
                        if self.dnssec_log_bogus {
                            warn!(name=%sanitize_dns_name(qname), "DNSSEC bogus — SERVFAIL");
                        }
                        self.record_query(client_ip, qname, qtype, ResponseCode::ServFail, LogAction::Servfail, start);
                        return send_error(request, response_handle, ResponseCode::ServFail).await;
                    }
                    let has_rrsig = records.iter().any(|r| r.record_type() == RecordType::RRSIG);
                    if has_rrsig {
                        self.stats.inc_dnssec_secure();
                    } else {
                        self.stats.inc_dnssec_insecure();
                    }
                }

                debug!(name=%sanitize_dns_name(qname), %qtype, count = records.len(), "resolved");
                let mut metadata = Metadata::response_from_request(&request.metadata);
                metadata.recursion_available = true;
                let builder = MessageResponseBuilder::from_message_request(request);
                let response = builder.build(
                    metadata, records.iter(),
                    std::iter::empty(), std::iter::empty(), std::iter::empty(),
                );
                self.stats.inc_forwarded();
                let fwd_us = start.elapsed().as_micros() as u64;
                self.stats.record_forward(fwd_us);
                let fwd_action = if fwd_us < CACHE_HIT_THRESHOLD_US {
                    LogAction::Cached
                } else {
                    LogAction::Forwarded
                };
                self.record_query(client_ip, qname, qtype, ResponseCode::NoError, fwd_action, start);
                response_handle.send_response(response).await
                    .unwrap_or_else(|e| { error!("send: {e}"); servfail_info(request) })
            }
            Err(e) => {
                // DNSSEC bogus via NSEC denial: validated proof the record does not exist.
                let is_dnssec_bogus = self.dnssec_enabled
                    && matches!(&e, NetError::Dns(DnsError::Nsec { proof, .. }) if proof.is_bogus());

                if is_dnssec_bogus {
                    self.stats.inc_dnssec_bogus();
                    if self.dnssec_log_bogus {
                        warn!(name=%sanitize_dns_name(qname), "DNSSEC bogus — SERVFAIL (NSEC denial)");
                    }
                }

                let rcode = match &e {
                    NetError::Dns(DnsError::NoRecordsFound(NoRecords { response_code, .. })) => {
                        debug!(name=%sanitize_dns_name(qname), ?response_code, "no records from resolver");
                        *response_code
                    }
                    _ => {
                        if !is_dnssec_bogus {
                            warn!(name=%sanitize_dns_name(qname), err=%e, "resolver error → SERVFAIL");
                        }
                        ResponseCode::ServFail
                    }
                };
                let err_action = match rcode {
                    ResponseCode::NXDomain => { self.stats.inc_nxdomain(); LogAction::Nxdomain }
                    ResponseCode::ServFail => { self.stats.inc_servfail(); LogAction::Servfail }
                    ResponseCode::Refused  => { self.stats.inc_refused();  LogAction::Refused  }
                    _                      => LogAction::Servfail,
                };
                self.record_query(client_ip, qname, qtype, rcode, err_action, start);
                send_error(request, response_handle, rcode).await
            }
        }
    }
}

#[async_trait]
impl RequestHandler for RunboundHandler {
    async fn handle_request<R: ResponseHandler, T: Time>(
        &self,
        request: &Request,
        response_handle: R,
    ) -> ResponseInfo {
        let start = Instant::now();

        // Require exactly one query — RFC 1035 §4.1.2.
        let Ok(info) = request.request_info() else {
            self.stats.inc_total();
            return servfail_info(request);
        };
        let qname     = info.query.name();
        let qtype     = info.query.query_type();
        let client_ip = info.src.ip();

        self.stats.inc_total();

        // ── 0. Access-control list ──────────────────────────────────────
        match self.acl.check(client_ip) {
            AclAction::Allow  => {}
            AclAction::Deny   => {
                // Silently drop — no response sent, just track the event.
                debug!(%client_ip, name=%sanitize_dns_name(qname), "ACL deny (silent drop)");
                self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Refused, start);
                let mut meta = Metadata::response_from_request(&request.metadata);
                meta.response_code = ResponseCode::Refused;
                return ResponseInfo::from(hickory_proto::op::Header {
                    metadata: meta,
                    counts:   hickory_proto::op::HeaderCounts::default(),
                });
            }
            AclAction::Refuse => {
                debug!(%client_ip, name=%sanitize_dns_name(qname), "ACL refuse");
                self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Refused, start);
                return send_error(request, response_handle, ResponseCode::Refused).await;
            }
        }

        // ── 1. Rate limiting (per source IP) ───────────────────────────
        if !self.rate_limiter.check(client_ip) {
            warn!(%client_ip, "rate limited");
            self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Refused, start);
            return send_error(request, response_handle, ResponseCode::Refused).await;
        }

        // ── 2. Concurrency cap (anti-OOM) ──────────────────────────────
        let _permit = match self.inflight.try_acquire() {
            Ok(p) => p,
            Err(_) => {
                warn!(%client_ip, inflight = MAX_INFLIGHT_REQUESTS, "inflight cap reached — REFUSED");
                self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Refused, start);
                return send_error(request, response_handle, ResponseCode::Refused).await;
            }
        };

        // ── 3a. Block CHAOS class queries (version.bind, hostname.bind) ───
        // CHAOS class (numeric 3) exposes server identity. Compare by wire value
        // to avoid any hickory Unknown(3) vs CH variant mismatch.
        // RFC 5358 §4: responders that do not implement CHAOS SHOULD return NOTIMP.
        if u16::from(info.query.query_class()) == 3 {
            debug!(%client_ip, name=%sanitize_dns_name(qname), "CHAOS class query → NOTIMP");
            self.stats.inc_refused();
            self.record_query(client_ip, qname, qtype, ResponseCode::NotImp, LogAction::Refused, start);
            return send_error(request, response_handle, ResponseCode::NotImp).await;
        }

        // ── 3b. SEC-03: defense-in-depth — block identity-probe names ───────
        // Block well-known identity-probe names by name regardless of query
        // class. hickory may normalise the CHAOS class (numeric 3) to IN
        // before invoking our handler, which bypasses the class check above
        // (observed: version.bind → NOERROR, hostname.bind → NXDOMAIN).
        // Zero allocation: qname is already a LowerName; compared against a
        // static set initialised once on first request.
        if identity_probe_names().iter().any(|n| n == qname) {
            debug!(%client_ip, name=%sanitize_dns_name(qname), "identity probe → REFUSED");
            self.stats.inc_refused();
            self.record_query(client_ip, qname, qtype, ResponseCode::Refused, LogAction::Refused, start);
            return send_error(request, response_handle, ResponseCode::Refused).await;
        }

        // ── 3c. Block ANY queries (RFC 8482 — amplification vector) ────
        if qtype == RecordType::ANY {
            debug!(%client_ip, "ANY query blocked (RFC 8482)");
            self.record_query(client_ip, qname, qtype, ResponseCode::NotImp, LogAction::Refused, start);
            return send_error(request, response_handle, ResponseCode::NotImp).await;
        }

        debug!(%client_ip, name=%sanitize_dns_name(qname), type=%qtype, "DNS query");

        // ── 4. Local zones ──────────────────────────────────────────────
        let response_handle = match self.handle_local_zone(
            request, response_handle, qname, qtype, client_ip, start,
        ).await {
            Ok(info) => return info,
            Err(rh)  => rh,
        };

        // ── 5. Recursive resolution ─────────────────────────────────────
        self.resolve_upstream(request, response_handle, qname, qtype, client_ip, start).await
    }
}

// ============================================================
// Helpers
// ============================================================

/// MED-06: Strip control characters from DNS names before structured log emission.
/// Prevents log injection via carefully crafted query names containing \n, \r, etc.
/// Fast path: if the name contains only printable ASCII (the common case), the
/// String from to_string() is returned as-is — no second allocation.
fn sanitize_dns_name(name: &LowerName) -> String {
    let s = name.to_string();
    if s.bytes().any(|b| !(0x20..0x7f).contains(&b)) {
        s.chars().map(|c| if c.is_ascii() && !c.is_ascii_control() { c } else { '?' }).collect()
    } else {
        s
    }
}

/// Follow CNAME records within local zones, up to 8 hops (prevents loops).
/// Returns a Vec<Record> containing the CNAME chain + final target records,
/// or an empty Vec if there is no CNAME or the chain leads outside local zones.
fn follow_local_cname(
    zones: &super::local::LocalZoneSet,
    start: &LowerName,
    qtype: RecordType,
) -> Vec<Record> {
    let mut chain: Vec<Record> = Vec::with_capacity(8);
    let mut current = start.clone();

    for _ in 0..8 {
        let cnames = zones.local_records(&current, RecordType::CNAME);
        if cnames.is_empty() { break; }
        let cname_rec = (*cnames[0]).clone();
        let next = match &cname_rec.data {
            RData::CNAME(c) => LowerName::from(c.0.clone()),
            _ => break,
        };
        chain.push(cname_rec);
        let resolved: Vec<Record> = zones.local_records(&next, qtype)
            .into_iter().map(|r| (*r).clone()).collect();
        if !resolved.is_empty() {
            chain.extend(resolved);
            return chain;
        }
        current = next;
    }
    // Chain incomplete (target not in local zones) — return nothing;
    // the caller will fall through to NODATA / NXDOMAIN as appropriate.
    Vec::new()
}

/// Construct a ResponseInfo carrying a SERVFAIL code without sending.
/// Used as the send-failure fallback when the socket is already broken.
#[inline]
fn servfail_info(request: &Request) -> ResponseInfo {
    let mut meta = Metadata::response_from_request(&request.metadata);
    meta.response_code = ResponseCode::ServFail;
    ResponseInfo::from(hickory_proto::op::Header {
        metadata: meta,
        counts:   hickory_proto::op::HeaderCounts::default(),
    })
}

/// Send an error response, mirroring the request's EDNS0 OPT record if present.
/// RFC 6891 §7: "If a query included an OPT record, the response MUST include one."
#[inline(always)]
async fn send_error<R: ResponseHandler>(
    request: &Request,
    mut response_handle: R,
    rcode: ResponseCode,
) -> ResponseInfo {
    // `error_msg` mirrors the request's EDNS0 OPT record, satisfying RFC 6891 §7.
    let builder  = MessageResponseBuilder::from_message_request(request);
    let response = builder.error_msg(&request.metadata, rcode);
    response_handle.send_response(response).await
        .unwrap_or_else(|e| {
            error!("send: {e}");
            servfail_info(request)
        })
}

/// Build resolver from forward-zones in unbound.conf, fallback to system resolvers.
fn build_resolver(cfg: &UnboundConfig, cache_size: usize) -> anyhow::Result<TokioResolver> {
    let mut resolver_cfg = ResolverConfig::from_parts(None, vec![], vec![]);

    for fwd in &cfg.forward_zones {
        for addr_str in &fwd.addrs {
            // Supports Unbound addr@port syntax: "1.1.1.1@853"
            let default_port = if fwd.tls { 853u16 } else { 53u16 };
            let (ip_str, port) = if let Some(at) = addr_str.find('@') {
                let p: u16 = addr_str[at + 1..].parse().unwrap_or(default_port);
                (&addr_str[..at], p)
            } else {
                (addr_str.as_str(), default_port)
            };
            if let Ok(ip) = ip_str.parse::<IpAddr>() {
                if fwd.tls {
                    // DNS-over-TLS: encrypted channel, single TCP connection per server.
                    let mut cc = ConnectionConfig::tls(Arc::from(ip.to_string()));
                    cc.port = port;
                    resolver_cfg.add_name_server(NameServerConfig::new(ip, true, vec![cc]));
                } else {
                    // UDP for normal queries (fast path)
                    let mut cc_udp = ConnectionConfig::udp();
                    cc_udp.port = port;
                    // TCP mandatory for DNSSEC: DNSKEY RRsets exceed UDP MTU and
                    // arrive truncated (TC=1). Without TCP, the DNSSEC chain cannot
                    // be completed and every signed domain returns SERVFAIL.
                    let mut cc_tcp = ConnectionConfig::tcp();
                    cc_tcp.port = port;
                    resolver_cfg.add_name_server(NameServerConfig::new(ip, true, vec![cc_udp, cc_tcp]));
                }
            }
        }
    }

    // No forward-zone configured — fall back to Cloudflare.
    // WARNING: This sends all DNS queries to a third-party cloud resolver.
    // For sensitive or nation-state deployments, configure explicit forward-zone
    // blocks with trusted upstream resolvers. This fallback should never trigger
    // in production; it exists only to make Runbound usable out of the box.
    if resolver_cfg.name_servers().is_empty() {
        warn!(
            "No forward-zone configured — falling back to Cloudflare (1.1.1.1). \
             All DNS queries will be sent to a third-party resolver. \
             Add forward-zone blocks to runbound.conf to suppress this warning."
        );
        for ip_str in ["1.1.1.1", "1.0.0.1"] {
            let ip: IpAddr = ip_str.parse().expect("valid Cloudflare IP");
            resolver_cfg.add_name_server(NameServerConfig::new(ip, true, vec![
                ConnectionConfig::udp(),
                ConnectionConfig::tcp(),
            ]));
        }
    }

    let mut opts = ResolverOpts::default();
    opts.recursion_desired = true;
    opts.cache_size = cache_size as u64;
    opts.timeout = Duration::from_secs(3);            // hard timeout per upstream query
    opts.attempts = 2;                                // retry once before SERVFAIL
    // DNSSEC: controlled by `dnssec-validation` directive (default: off for forwarders).
    // Enable only when operating as a full recursive resolver with complete RRSIG chains.
    // Forwarders must leave this off: upstreams strip RRSIGs before forwarding,
    // causing spurious SERVFAIL for every signed domain (gmail.com, google.com…).
    opts.validate = cfg.dnssec_validation;
    opts.use_hosts_file = ResolveHosts::Never;        // don't leak host file data

    TokioResolver::builder_with_config(resolver_cfg, TokioRuntimeProvider::default())
        .with_options(opts)
        .build()
        .map_err(|e| anyhow::anyhow!("resolver build: {e}"))
}

// ============================================================
// Memory pressure guard
// ============================================================

// Check memory every 30 s. On Linux /proc/meminfo is a cheap kernel read.
const MEM_CHECK_SECS:        u64 = 30;
// Scale-up cooldown: do not increase cache more often than every 5 minutes.
const MEM_SCALEUP_COOLDOWN:  u64 = 300;
// Halving cooldown: do not halve more often than once every 5 minutes.
const CACHE_HALVE_COOLDOWN: Duration = Duration::from_secs(300);
// Memory pressure thresholds (used ratio = 1 - MemAvailable/MemTotal):
const MEM_LOW_WATERMARK:    f64 = 0.60; // below → scale up if cache was reduced
const MEM_MOD_WATERMARK:    f64 = 0.70; // [0.70, 0.80) → halve cache
const MEM_HIGH_WATERMARK:   f64 = 0.80; // ≥ 0.80 → recalc + flush rate limiter

/// Compute an appropriate resolver cache size from current available RAM.
///
/// 1 hickory cache entry ≈ 512 bytes.
/// Allocates up to 10 % of MemAvailable, clamped to [512, 65536].
/// Falls back to 8192 when /proc/meminfo is unavailable.
fn cache_size_from_meminfo() -> usize {
    let avail_kb: u64 = std::fs::read_to_string("/proc/meminfo").ok()
        .and_then(|t| t.lines()
            .find(|l| l.starts_with("MemAvailable:"))
            .and_then(|l| l.split_whitespace().nth(1))
            .and_then(|v| v.parse().ok()))
        .unwrap_or(0);
    if avail_kb == 0 {
        return 8192;
    }
    ((avail_kb * 1024 / 10 / 512) as usize).clamp(512, 65536)
}

/// Read /proc/meminfo and return (available_kb, total_kb).
/// Returns None on any parse or I/O error (non-Linux, container without /proc, etc.).
fn read_meminfo() -> Option<(u64, u64)> {
    let text = std::fs::read_to_string("/proc/meminfo").ok()?;
    let (mut total, mut available) = (0u64, 0u64);
    for line in text.lines() {
        if line.starts_with("MemTotal:") {
            total = line.split_whitespace().nth(1)?.parse().ok()?;
        } else if line.starts_with("MemAvailable:") {
            available = line.split_whitespace().nth(1)?.parse().ok()?;
        }
        if total > 0 && available > 0 { break; }
    }
    if total > 0 { Some((available, total)) } else { None }
}

/// Background task: monitors system memory and adjusts the DNS resolver cache size.
///
/// Four operating modes based on memory pressure (used = 1 - MemAvailable/MemTotal):
///   < 60 %  — scale up: restore cache toward optimal size (with cooldown).
///   60–70 % — stable: no action.
///   70–80 % — moderate pressure: halve cache size, floor at cache_min_entries.
///   ≥ 80 %  — high pressure: recalc cache from current RAM + flush rate limiter.
///
/// Three guards prevent infinite cache destruction on memory-constrained systems:
///   1. Floor: never halve below cfg.cache_min_entries (default 2048).
///   2. Cooldown: at most one halving per CACHE_HALVE_COOLDOWN (5 min).
///   3. No-effect detection: if halving does not reduce used_pct by ≥ 5 %,
///      halvings are disabled for this process lifetime with a clear WARN.
///
/// Cache changes take effect by rebuilding the hickory resolver and atomically
/// swapping it via ArcSwap. In-flight queries keep their Arc until completion.
pub async fn memory_guard_loop(
    rate_limiter:         Arc<RateLimiter>,
    resolver:             Arc<ArcSwap<TokioResolver>>,
    cfg:                  Arc<UnboundConfig>,
    stats:                Arc<Stats>,
    initial_cache_size:   usize,
) {
    let mut interval = tokio::time::interval(Duration::from_secs(MEM_CHECK_SECS));
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

    let mut current_cache_size = initial_cache_size;
    let mut last_scale_up = Instant::now()
        .checked_sub(Duration::from_secs(MEM_SCALEUP_COOLDOWN))
        .unwrap_or_else(Instant::now);

    // Halving guards
    let mut last_halved: Instant = Instant::now()
        .checked_sub(CACHE_HALVE_COOLDOWN)
        .unwrap_or_else(Instant::now);
    // used_ratio captured just before the last halving; cleared after one check cycle.
    let mut pct_before_last_halve: Option<f64> = None;
    // Set permanently when halving is proven to have no measurable effect on RSS.
    let mut halving_disabled = false;

    loop {
        interval.tick().await;

        let Some((avail_kb, total_kb)) = read_meminfo() else { continue };
        let used_ratio = 1.0 - (avail_kb as f64 / total_kb as f64);

        // No-effect detection: check whether the previous halving reduced pressure.
        if let Some(pct_before) = pct_before_last_halve.take() {
            if used_ratio >= pct_before - 0.05 {
                halving_disabled = true;
                warn!(
                    pct_before = format!("{:.1}%", pct_before * 100.0),
                    pct_after  = format!("{:.1}%", used_ratio * 100.0),
                    "cache halving has no effect on memory pressure \
                     (pct before={:.1}% after={:.1}%) — cache floor reached, \
                     consider increasing MemoryMax in the service file or reducing other workloads",
                    pct_before * 100.0,
                    used_ratio * 100.0,
                );
            }
        }

        if used_ratio >= MEM_HIGH_WATERMARK {
            // High pressure — recalc from current RAM state, flush rate limiter.
            let new_size = cache_size_from_meminfo();
            match build_resolver(&cfg, new_size) {
                Ok(new_res) => {
                    resolver.store(Arc::new(new_res));
                    stats.reset_cache();
                    let freed = rate_limiter.clear();
                    warn!(
                        used_pct = format!("{:.1}%", used_ratio * 100.0),
                        cache_from = current_cache_size,
                        cache_to   = new_size,
                        freed_buckets = freed,
                        "memory pressure high — cache flushed, resized, rate limiter cleared"
                    );
                    current_cache_size = new_size;
                }
                Err(e) => warn!(%e, "memory guard: resolver rebuild failed (high pressure)"),
            }
        } else if used_ratio >= MEM_MOD_WATERMARK {
            // Moderate pressure — halve cache, floor at cache_min_entries.
            let min = cfg.cache_min_entries;
            if halving_disabled {
                // no-op: halving proven ineffective, avoid log spam
            } else if current_cache_size <= min {
                warn!(
                    cache_size = current_cache_size,
                    cache_min  = min,
                    used_pct   = format!("{:.1}%", used_ratio * 100.0),
                    "cache at minimum size ({}) — memory pressure ignored", min,
                );
            } else if last_halved.elapsed() < CACHE_HALVE_COOLDOWN {
                // cooldown active — skip this cycle silently
            } else {
                let new_size = (current_cache_size / 2).max(min);
                match build_resolver(&cfg, new_size) {
                    Ok(new_res) => {
                        resolver.store(Arc::new(new_res));
                        stats.reset_cache();
                        warn!(
                            used_pct   = format!("{:.1}%", used_ratio * 100.0),
                            cache_from = current_cache_size,
                            cache_to   = new_size,
                            "memory pressure — cache halved"
                        );
                        current_cache_size = new_size;
                        last_halved = Instant::now();
                        pct_before_last_halve = Some(used_ratio);
                    }
                    Err(e) => warn!(%e, "memory guard: resolver rebuild failed (moderate pressure)"),
                }
            }
        } else if used_ratio < MEM_LOW_WATERMARK {
            // Memory freed — scale up toward optimal if cooldown elapsed.
            let optimal = cache_size_from_meminfo();
            let elapsed = last_scale_up.elapsed();
            if optimal > current_cache_size
                && elapsed >= Duration::from_secs(MEM_SCALEUP_COOLDOWN)
            {
                match build_resolver(&cfg, optimal) {
                    Ok(new_res) => {
                        resolver.store(Arc::new(new_res));
                        stats.reset_cache();
                        info!(
                            used_pct   = format!("{:.1}%", used_ratio * 100.0),
                            cache_from = current_cache_size,
                            cache_to   = optimal,
                            "memory pressure resolved — cache scaled up"
                        );
                        current_cache_size = optimal;
                        last_scale_up = Instant::now();
                    }
                    Err(e) => warn!(%e, "memory guard: resolver rebuild failed (scale up)"),
                }
            }
        }
        // 60–70 %: stable band — no action.
    }
}

// ============================================================
// TLS helpers
// ============================================================

/// Load TLS cert+key materials from PEM files. Returns None if not configured.
fn load_tls_materials(tls: &TlsConfig) -> Option<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)> {
    let cert_path = tls.cert_path.as_deref()?;
    let key_path  = tls.key_path.as_deref()?;
    let cert_pem  = std::fs::read(cert_path).ok()?;
    let key_pem   = std::fs::read(key_path).ok()?;
    let certs: Vec<CertificateDer<'static>> = rustls_pemfile::certs(&mut cert_pem.as_slice())
        .flatten()
        .collect();
    let key = rustls_pemfile::private_key(&mut key_pem.as_slice()).ok()??.clone_key();
    if certs.is_empty() { return None; }
    Some((certs, key))
}

/// Build a rustls ServerConfig for a specific DNS-over-TLS protocol.
///
/// * `alpn`      — ALPN token: `b"dot"`, `b"h2"`, or `b"doq"`.
/// * `tls13_only`— true for DoQ (Quinn requires TLS 1.3 exclusively).
/// * `client_ca` — optional path to CA PEM for mTLS (HIGH-08, DoT only).
fn build_tls_config(
    certs:     Vec<CertificateDer<'static>>,
    key:       PrivateKeyDer<'static>,
    alpn:      &[u8],
    tls13_only: bool,
    client_ca: Option<&str>,
) -> anyhow::Result<Arc<rustls::ServerConfig>> {
    let builder = if tls13_only {
        // DoQ requires TLS 1.3 — Quinn rejects configs that allow TLS 1.2.
        rustls::ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
    } else {
        rustls::ServerConfig::builder()
    };

    let mut config = if let Some(ca_path) = client_ca {
        // HIGH-08: mutual TLS for DoT — require a client certificate signed by ca_path.
        let ca_pem = std::fs::read(ca_path)
            .map_err(|e| anyhow::anyhow!("read CA cert {ca_path}: {e}"))?;
        let mut roots = rustls::RootCertStore::empty();
        for cert in rustls_pemfile::certs(&mut ca_pem.as_slice()).flatten() {
            roots.add(cert)
                .map_err(|e| anyhow::anyhow!("load CA cert: {e}"))?;
        }
        let verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(roots))
            .build()
            .map_err(|e| anyhow::anyhow!("mTLS verifier: {e}"))?;
        builder
            .with_client_cert_verifier(verifier)
            .with_single_cert(certs, key)
            .map_err(|e| anyhow::anyhow!("TLS config: {e}"))?
    } else {
        builder
            .with_no_client_auth()
            .with_single_cert(certs, key)
            .map_err(|e| anyhow::anyhow!("TLS config: {e}"))?
    };

    config.alpn_protocols = vec![alpn.to_vec()];
    Ok(Arc::new(config))
}

// SO_REUSEPORT UDP socket: the kernel distributes incoming packets across
// all sockets bound to the same address:port, one per CPU core. Each socket
// is driven by a separate tokio task on a separate thread, giving true
// multi-core parallelism without any userspace load-balancing overhead.
#[cfg(unix)]
fn bind_reuseport_udp(addr: &str) -> anyhow::Result<UdpSocket> {
    use socket2::{Domain, Protocol, Socket, Type};
    let sock_addr: std::net::SocketAddr = addr.parse()
        .map_err(|e| anyhow::anyhow!("parse addr {addr}: {e}"))?;
    let domain = if sock_addr.is_ipv6() { Domain::IPV6 } else { Domain::IPV4 };
    let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))
        .map_err(|e| anyhow::anyhow!("socket new: {e}"))?;
    socket.set_reuse_port(true)
        .map_err(|e| anyhow::anyhow!("SO_REUSEPORT: {e}"))?;
    socket.bind(&sock_addr.into())
        .map_err(|e| anyhow::anyhow!("bind {addr}: {e}"))?;
    socket.set_nonblocking(true)
        .map_err(|e| anyhow::anyhow!("nonblocking: {e}"))?;
    let std_socket: std::net::UdpSocket = socket.into();
    UdpSocket::from_std(std_socket)
        .map_err(|e| anyhow::anyhow!("tokio from_std: {e}"))
}

// ── FIX 6.2: Per-IP TCP connection cap ────────────────────────────────────────

/// Max concurrent TCP DNS connections from a single source IP (or /48 for IPv6).
const TCP_CONN_PER_IP_MAX: u16 = 20;

/// Truncate IPv6 to /48 for TCP connection tracking (consistent with rate limiter).
fn normalize_tcp_ip(ip: IpAddr) -> IpAddr {
    match ip {
        IpAddr::V4(_) => ip,
        IpAddr::V6(v6) => {
            let mut octets = v6.octets();
            octets[6..].fill(0);
            IpAddr::V6(Ipv6Addr::from(octets))
        }
    }
}

struct TcpConnTracker {
    counts:    DashMap<IpAddr, Arc<AtomicU16>, ahash::RandomState>,
    last_warn: DashMap<IpAddr, Instant,        ahash::RandomState>,
}

impl TcpConnTracker {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            counts:    DashMap::with_hasher(ahash::RandomState::default()),
            last_warn: DashMap::with_hasher(ahash::RandomState::default()),
        })
    }

    /// Attempt to claim a connection slot for `ip`.
    /// Returns `true` if allowed, `false` if the per-IP cap is exceeded.
    /// Loopback addresses (127.x and ::1) are always allowed (health checks).
    fn try_acquire(&self, ip: IpAddr) -> bool {
        if matches!(ip, IpAddr::V4(a) if a.is_loopback())
            || matches!(ip, IpAddr::V6(a) if a.is_loopback())
        {
            return true;
        }
        let counter = self.counts
            .entry(ip)
            .or_insert_with(|| Arc::new(AtomicU16::new(0)));
        let prev = counter.fetch_add(1, Ordering::Relaxed);
        if prev >= TCP_CONN_PER_IP_MAX {
            counter.fetch_sub(1, Ordering::Relaxed);
            let mut emit = false;
            self.last_warn
                .entry(ip)
                .and_modify(|t| {
                    if t.elapsed().as_secs() >= 60 {
                        *t = Instant::now();
                        emit = true;
                    }
                })
                .or_insert_with(|| {
                    emit = true;
                    Instant::now()
                });
            if emit {
                warn!(%ip, limit = TCP_CONN_PER_IP_MAX, "TCP per-IP connection cap reached — dropping");
            }
            false
        } else {
            true
        }
    }

    fn release(&self, ip: IpAddr) {
        if let Some(c) = self.counts.get(&ip) {
            c.fetch_sub(1, Ordering::Relaxed);
        }
    }
}

/// Accept-with-limit loop for a public-facing TCP listener.
/// Connections within the per-IP cap are relayed to `relay_addr` (a loopback
/// listener owned by hickory-server) via bidirectional byte copy.
///
/// Trade-off: hickory sees 127.0.0.1 as the source for all relayed TCP
/// connections, so the DNS per-IP rate limiter uses a shared loopback bucket
/// for TCP clients. Acceptable because TCP DNS traffic is inherently low-volume
/// (large responses, DNSSEC chains). The TCP connection cap enforced here
/// prevents the primary DoS vector (FD exhaustion via many idle connections).
async fn run_tcp_with_limit(
    public_tcp:   TcpListener,
    relay_addr:   SocketAddr,
    tracker:      Arc<TcpConnTracker>,
    conn_timeout: Duration,
) {
    loop {
        let (mut client, peer) = match public_tcp.accept().await {
            Ok(x)  => x,
            Err(e) => { warn!(err=%e, "TCP accept error"); continue; }
        };
        // FIX 2 (VUL-NEW-03): check loopback BEFORE normalize so that ::1
        // is not collapsed to :: (an unrelated /48 prefix) by normalize_tcp_ip.
        let src_ip = if peer.ip().is_loopback() {
            peer.ip()
        } else {
            normalize_tcp_ip(peer.ip())
        };
        if !tracker.try_acquire(src_ip) {
            // Over limit: drop immediately (TcpStream closed on drop → TCP FIN/RST)
            continue;
        }
        let tracker2 = Arc::clone(&tracker);
        tokio::spawn(async move {
            let r = tokio::time::timeout(conn_timeout, async {
                let mut relay = TcpStream::connect(relay_addr).await?;
                tokio::io::copy_bidirectional(&mut client, &mut relay).await?;
                Ok::<_, std::io::Error>(())
            })
            .await;
            if let Ok(Err(e)) = r {
                debug!(err=%e, %src_ip, "TCP relay error");
            }
            tracker2.release(src_ip);
        });
    }
}

pub async fn run_dns_server(
    cfg:          &UnboundConfig,
    zones:        Arc<ArcSwap<LocalZoneSet>>,
    rate_limiter: Arc<RateLimiter>,
    acl:          Arc<Acl>,
    stats:        Arc<Stats>,
    log_buffer:   SharedLogBuffer,
) -> anyhow::Result<()> {
    let tls_cfg = &cfg.tls;
    let rps = cfg.rate_limit.unwrap_or(RATE_LIMIT_QPS_DEFAULT);
    if rps == 0 {
        info!("rate limiting disabled (rate-limit: 0)");
    } else {
        info!(rps, burst = rps * 2, "DNS rate limiter configured");
    }

    let initial_cache_size = cache_size_from_meminfo();
    info!(cache_size = initial_cache_size, "cache size auto-sized from MemAvailable");
    let resolver = Arc::new(ArcSwap::new(Arc::new(build_resolver(cfg, initial_cache_size)?)));

    // Spawn memory pressure guard — monitors /proc/meminfo every 30 s and
    // adjusts the DNS cache size and flushes the rate limiter under pressure.
    {
        let rl       = Arc::clone(&rate_limiter);
        let res      = Arc::clone(&resolver);
        let cfg_arc  = Arc::new(cfg.clone());
        let stats_mg = Arc::clone(&stats);
        tokio::spawn(async move {
            memory_guard_loop(rl, res, cfg_arc, stats_mg, initial_cache_size).await
        });
    }

    if acl.is_empty() {
        info!("access-control: no rules — all clients allowed (add access-control directives to restrict)");
    } else {
        info!(rules = acl.len(), "access-control: ACL loaded");
    }

    let cache_max_ttl = cfg.cache_max_ttl.unwrap_or(86400);
    info!(cache_max_ttl, "TTL cap configured");

    let private_addrs = Arc::new(PrivateAddressSet::from_config(&cfg.private_addresses));
    if !private_addrs.is_empty() {
        info!(count = cfg.private_addresses.len(), "private-address: DNS rebinding protection active");
    }

    let handler = RunboundHandler::new(
        Arc::clone(&zones), resolver, rate_limiter, acl, private_addrs, cache_max_ttl, stats, log_buffer,
        cfg.dnssec_validation, cfg.dnssec_log_bogus,
    );
    let mut server = Server::new(handler);

    let ncpus = crate::cpu::physical_cores().len().max(1);

    let port = cfg.port;
    let interfaces: Vec<String> = if cfg.interfaces.is_empty() {
        vec!["0.0.0.0".into()]
    } else {
        cfg.interfaces.clone()
    };

    // FIX 6.2: shared per-IP TCP connection tracker (across all interfaces)
    let tcp_tracker = TcpConnTracker::new();
    const TCP_SESSION_TIMEOUT: Duration = Duration::from_secs(30);

    for iface in &interfaces {
        let udp_addr = format!("{}:{}", iface, port);
        let tcp_addr = format!("{}:{}", iface, port);

        // Bind one UDP socket per CPU with SO_REUSEPORT.
        // The kernel distributes packets across sockets via a flow hash,
        // giving near-linear QPS scaling with core count.
        #[cfg(unix)]
        for i in 0..ncpus {
            let udp = bind_reuseport_udp(&udp_addr)
                .map_err(|e| anyhow::anyhow!("UDP SO_REUSEPORT socket {i}: {e}"))?;
            server.register_socket(udp);
        }
        #[cfg(not(unix))]
        {
            let udp = UdpSocket::bind(&udp_addr).await
                .map_err(|e| anyhow::anyhow!("UDP bind {udp_addr}: {e}"))?;
            server.register_socket(udp);
        }
        info!(addr=%udp_addr, sockets=ncpus, "DNS UDP listening (SO_REUSEPORT)");

        // FIX 6.2: public-facing TCP listener feeds our per-IP accept gate.
        // hickory-server gets a loopback relay listener so its internal accept
        // loop never sees connections from over-limit source IPs.
        let public_tcp = TcpListener::bind(&tcp_addr).await
            .map_err(|e| anyhow::anyhow!("TCP bind {tcp_addr}: {e}"))?;
        // Relay listener: loopback, ephemeral port — hickory owns this listener.
        let relay_tcp = TcpListener::bind("127.0.0.1:0").await
            .map_err(|e| anyhow::anyhow!("TCP relay bind: {e}"))?;
        let relay_addr = relay_tcp.local_addr()
            .map_err(|e| anyhow::anyhow!("TCP relay local_addr: {e}"))?;
        info!(addr=%tcp_addr, "DNS TCP listening (per-IP cap: {} conns)", TCP_CONN_PER_IP_MAX);
        // 30s idle timeout — enough for slow DNSSEC responses while limiting FD exhaustion.
        // 4096-byte response buffer fits any DNS response (max UDP payload is 65535 bytes but
        // typical responses are well under 4 KiB; EDNS0 handles larger).
        server.register_listener(relay_tcp, TCP_SESSION_TIMEOUT, 4096);

        let tracker2 = Arc::clone(&tcp_tracker);
        tokio::spawn(run_tcp_with_limit(public_tcp, relay_addr, tracker2, TCP_SESSION_TIMEOUT));
    }

    // ── DoT / DoH / DoQ (RFC 7858 / 8484 / 9250) ───────────────────────────
    if let Some((certs, key)) = load_tls_materials(tls_cfg) {
        let dot_port = tls_cfg.dot_port.unwrap_or(853);
        let doh_port = tls_cfg.doh_port.unwrap_or(443);
        let doq_port = tls_cfg.doq_port.unwrap_or(853);
        let hostname = tls_cfg.hostname.clone()
            .unwrap_or_else(|| "runbound.local".to_string());

        // Build one ServerConfig per protocol (each needs its own ALPN token).
        // PrivateKeyDer does not implement Clone; use clone_key() for each copy.
        let dot_config = build_tls_config(
            certs.clone(), key.clone_key(),
            b"dot", false,
            tls_cfg.dot_client_auth_ca.as_deref(),
        ).map_err(|e| anyhow::anyhow!("DoT TLS config: {e}"))?;

        let doh_config = build_tls_config(
            certs.clone(), key.clone_key(),
            b"h2", false, None,
        ).map_err(|e| anyhow::anyhow!("DoH TLS config: {e}"))?;

        // DoQ requires TLS 1.3 exclusively (Quinn constraint).
        let doq_config = build_tls_config(
            certs, key,
            b"doq", true, None,
        ).map_err(|e| anyhow::anyhow!("DoQ TLS config: {e}"))?;

        for iface in &interfaces {
            // DNS-over-TLS (port 853 TCP)
            // FIX 1 (VUL-NEW-01): public listener → run_tcp_with_limit → loopback relay → hickory.
            // Same TcpConnTracker as DNS/TCP; DoT now shares the per-IP cap of 20 connections.
            let dot_addr = format!("{}:{}", iface, dot_port);
            match TcpListener::bind(&dot_addr).await {
                Ok(public_dot) => {
                    let relay_dot = TcpListener::bind("127.0.0.1:0").await
                        .map_err(|e| anyhow::anyhow!("DoT relay bind: {e}"))?;
                    let relay_dot_addr = relay_dot.local_addr()
                        .map_err(|e| anyhow::anyhow!("DoT relay local_addr: {e}"))?;
                    info!(addr=%dot_addr, mtls=tls_cfg.dot_client_auth_ca.is_some(), "DoT (DNS-over-TLS) listening — RFC 7858");
                    server.register_tls_listener_with_tls_config(relay_dot, Duration::from_secs(30), Arc::clone(&dot_config))
                        .map_err(|e| anyhow::anyhow!("DoT register: {e}"))?;
                    let tracker_dot = Arc::clone(&tcp_tracker);
                    tokio::spawn(run_tcp_with_limit(public_dot, relay_dot_addr, tracker_dot, TCP_SESSION_TIMEOUT));
                }
                Err(e) => warn!(addr=%dot_addr, err=%e, "DoT bind failed — skipping"),
            }

            // DNS-over-HTTPS (port 443 TCP)
            // FIX 1 (VUL-NEW-01): same relay pattern as DoT above.
            let doh_addr = format!("{}:{}", iface, doh_port);
            match TcpListener::bind(&doh_addr).await {
                Ok(public_doh) => {
                    let relay_doh = TcpListener::bind("127.0.0.1:0").await
                        .map_err(|e| anyhow::anyhow!("DoH relay bind: {e}"))?;
                    let relay_doh_addr = relay_doh.local_addr()
                        .map_err(|e| anyhow::anyhow!("DoH relay local_addr: {e}"))?;
                    info!(addr=%doh_addr, "DoH (DNS-over-HTTPS) listening — RFC 8484");
                    server.register_https_listener_with_tls_config(
                        relay_doh, Duration::from_secs(30),
                        Arc::clone(&doh_config),
                        Some(hostname.clone()),
                        "/dns-query".to_string(),
                    ).map_err(|e| anyhow::anyhow!("DoH register: {e}"))?;
                    let tracker_doh = Arc::clone(&tcp_tracker);
                    tokio::spawn(run_tcp_with_limit(public_doh, relay_doh_addr, tracker_doh, TCP_SESSION_TIMEOUT));
                }
                Err(e) => warn!(addr=%doh_addr, err=%e, "DoH bind failed — skipping"),
            }

            // DNS-over-QUIC (port 853 UDP)
            let doq_addr = format!("{}:{}", iface, doq_port);
            match UdpSocket::bind(&doq_addr).await {
                Ok(udp) => {
                    info!(addr=%doq_addr, "DoQ (DNS-over-QUIC) listening — RFC 9250");
                    server.register_quic_listener_and_tls_config(udp, Duration::from_secs(30), Arc::clone(&doq_config))
                        .map_err(|e| anyhow::anyhow!("DoQ register: {e}"))?;
                }
                Err(e) => warn!(addr=%doq_addr, err=%e, "DoQ bind failed — skipping"),
            }
        }
    } else {
        info!("TLS not configured — DoT/DoH/DoQ disabled (add tls-service-pem + tls-service-key to enable)");
    }

    info!("Runbound ready — RFC 1034/1035/2782/4033/6891/7858/8484/9250");
    server.block_until_done().await
        .map_err(|e| anyhow::anyhow!("Server error: {e}"))
}