diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index dbaa583..8031b26 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -65,7 +65,11 @@ Use `profile` to derive IPv4 DNS from the profile or `explicit` with total load, while `max_concurrent_queries_per_client` bounds any single client. Both matter: a global bound alone lets one client recovering from an outage consume every permit and deny service to every other enrolled client. Refusals -are counted separately per limit in `egressy_dns_queries_refused_total`. +are counted separately per limit in `egressy_dns_queries_refused_total`, and +are answered REFUSED rather than dropped, so a refused client fails now instead +of waiting out its own resolver timeout and retrying into the same limit. A +query the gateway tried and could not answer is answered SERVFAIL on the same +reasoning. `local_zones.enabled` answers single-label enrolled-bridge container names from Docker discovery instead of forwarding them; see `docs/NETWORKING.md` for the constraints that keep it from shadowing public names or leaking internal diff --git a/src/dns.rs b/src/dns.rs index d41480d..8c17da6 100644 --- a/src/dns.rs +++ b/src/dns.rs @@ -10,6 +10,7 @@ use std::{ }; use anyhow::{bail, Context}; +use hickory_proto::op::{Message, MessageType, ResponseCode}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::{TcpListener, TcpStream, UdpSocket}, @@ -406,6 +407,10 @@ async fn serve_udp( Err(refusal) => { record_refusal(refusal); warn!(%client, ?refusal, "DNS query refused by concurrency limit"); + // REFUSED, not silence: the gateway declined to try, which is a + // condition the client can act on now rather than in five + // seconds' time. + answer_with_rcode(&listener, client, &query, ResponseCode::Refused).await; continue; } }; @@ -414,6 +419,7 @@ async fn serve_udp( let Some(upstream) = *settings.upstream.borrow() else { observe_udp_health(settings.publisher.as_ref(), &health, false).await; warn!(%client, "DNS upstream is not configured"); + answer_with_rcode(&listener, client, &query, ResponseCode::ServFail).await; return; }; match forward_query(&query, upstream, settings.timeout, settings.udp_attempts).await { @@ -444,7 +450,9 @@ async fn serve_udp( } Err(error) => { observe_udp_health(settings.publisher.as_ref(), &health, false).await; - warn!(%client, %error, "DNS forwarding failed") + warn!(%client, %error, "DNS forwarding failed"); + // SERVFAIL: the gateway tried and could not get an answer. + answer_with_rcode(&listener, client, &query, ResponseCode::ServFail).await; } } }); @@ -473,6 +481,15 @@ async fn serve_tcp( Err(refusal) => { record_refusal(refusal); warn!(%address, ?refusal, "DNS query refused by concurrency limit"); + // Refusing costs a read and a write, not a permit: the point of + // saying REFUSED is that the client stops waiting, and a client + // that has already opened a connection is waiting. + let request_timeout = settings.timeout; + workers.spawn(async move { + if let Err(error) = refuse_tcp_client(&mut client, request_timeout).await { + debug!(%address, %error, "refusing a TCP DNS query failed"); + } + }); continue; } }; @@ -492,12 +509,37 @@ async fn handle_tcp_client(client: &mut TcpStream, settings: &Settings) -> anyho { return write_tcp_message(client, &response, settings.timeout).await; } + match forward_over_tcp(&query, settings).await { + Ok(response) => write_tcp_message(client, &response, settings.timeout).await, + Err(error) => { + // Say SERVFAIL before dropping the connection, so the client fails + // this query rather than reading the close as a transport problem + // and retrying the whole exchange. + if let Some(response) = rcode_response(&query, ResponseCode::ServFail) { + let _ = write_tcp_message(client, &response, settings.timeout).await; + } + Err(error) + } + } +} + +async fn forward_over_tcp(query: &[u8], settings: &Settings) -> anyhow::Result> { let upstream = (*settings.upstream.borrow()).context("DNS upstream is not configured")?; - let response = tcp_exchange(&query, upstream, settings.timeout).await?; + let response = tcp_exchange(query, upstream, settings.timeout).await?; observe_tcp_success(settings.publisher.as_ref()).await; - remember_answer(settings, &query, &response, upstream); - write_tcp_message(client, &response, settings.timeout).await?; - Ok(()) + remember_answer(settings, query, &response, upstream); + Ok(response) +} + +/// Read a query only far enough to answer it REFUSED. +async fn refuse_tcp_client( + client: &mut TcpStream, + request_timeout: Duration, +) -> anyhow::Result<()> { + let query = read_tcp_message(client, request_timeout).await?; + let response = + rcode_response(&query, ResponseCode::Refused).context("query could not be refused")?; + write_tcp_message(client, &response, request_timeout).await } async fn forward_query( @@ -702,6 +744,50 @@ fn is_truncated(message: &[u8]) -> anyhow::Result { Ok(message[2] & 0x02 != 0) } +/// A response carrying nothing but an rcode, echoing the question asked. +/// +/// Silence is the one answer a resolver client cannot act on: it cannot tell a +/// refusal from a gateway that has gone away, so it waits out its own timeout +/// — 5s with two attempts per nameserver on glibc — and retries. Under load +/// those retries arrive as a second wave against the capacity that refused the +/// first, which is how an upstream hiccup became sustained refusal. An rcode +/// fails the client immediately and lets it apply its own backoff. +/// +/// `None` for anything unparseable: there is no question to echo and no +/// transaction to answer, so there is nothing truthful to send. +fn rcode_response(query: &[u8], code: ResponseCode) -> Option> { + let request = Message::from_vec(query).ok()?; + if request.metadata.message_type != MessageType::Query { + return None; + } + let mut response = Message::response(request.metadata.id, request.metadata.op_code); + response.metadata.response_code = code; + response.metadata.recursion_desired = request.metadata.recursion_desired; + response.metadata.recursion_available = true; + response.queries = request.queries.clone(); + // RFC 6891: a responder that was sent an OPT record includes one in reply. + if let Some(edns) = request.edns.clone() { + response.set_edns(edns); + } + response.to_vec().ok() +} + +/// Answer a UDP query with an rcode, or say why it could not be answered. +async fn answer_with_rcode( + listener: &UdpSocket, + client: SocketAddr, + query: &[u8], + code: ResponseCode, +) { + let Some(response) = rcode_response(query, code) else { + debug!(%client, ?code, "DNS query could not be answered with an rcode"); + return; + }; + if let Err(error) = listener.send_to(&response, client).await { + warn!(%client, %error, ?code, "sending DNS rcode response failed"); + } +} + #[cfg(test)] mod tests { use super::*; @@ -812,6 +898,90 @@ mod tests { assert!(validate_response(&query, &mismatched).is_err()); } + use hickory_proto::{ + op::{OpCode, Query}, + rr::{Name, RecordType}, + }; + + fn dns_query(name: &str) -> Vec { + let mut message = Message::new(4242, MessageType::Query, OpCode::Query); + message.metadata.recursion_desired = true; + message.add_query(Query::query(Name::from_ascii(name).unwrap(), RecordType::A)); + message.to_vec().unwrap() + } + + #[test] + fn an_rcode_response_answers_the_question_it_was_asked() { + let query = dns_query("a.test."); + let bytes = rcode_response(&query, ResponseCode::Refused).unwrap(); + let response = Message::from_vec(&bytes).unwrap(); + assert_eq!(response.metadata.id, 4242); + assert_eq!(response.metadata.message_type, MessageType::Response); + assert_eq!(response.metadata.response_code, ResponseCode::Refused); + assert!(response.metadata.recursion_desired); + assert!(response.metadata.recursion_available); + assert_eq!(response.queries.len(), 1); + assert_eq!(response.queries[0].name().to_ascii(), "a.test."); + // Validation the forwarder applies to upstream responses holds here + // too: same transaction, response bit set. + validate_response(&query, &bytes).unwrap(); + } + + #[test] + fn both_failure_rcodes_are_distinguishable_to_the_client() { + let query = dns_query("a.test."); + let refused = Message::from_vec(&rcode_response(&query, ResponseCode::Refused).unwrap()) + .unwrap() + .metadata + .response_code; + let failed = Message::from_vec(&rcode_response(&query, ResponseCode::ServFail).unwrap()) + .unwrap() + .metadata + .response_code; + // "I declined to try" and "I tried and could not" are different + // conditions and the client backs off differently for each. + assert_eq!(refused, ResponseCode::Refused); + assert_eq!(failed, ResponseCode::ServFail); + } + + #[test] + fn nothing_is_sent_for_a_message_that_cannot_be_answered() { + // No question to echo and no transaction to answer. + assert!(rcode_response(b"nonsense", ResponseCode::ServFail).is_none()); + let mut response = Message::response(1, OpCode::Query); + response.add_query(Query::query( + Name::from_ascii("a.test.").unwrap(), + RecordType::A, + )); + assert!(rcode_response(&response.to_vec().unwrap(), ResponseCode::ServFail).is_none()); + } + + #[tokio::test] + async fn a_refused_query_is_answered_rather_than_dropped() { + // The regression that mattered: a client that gets silence waits out + // its own resolver timeout and retries into the capacity that just + // refused it. + let gateway = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let query = dns_query("a.test."); + answer_with_rcode( + &gateway, + client.local_addr().unwrap(), + &query, + ResponseCode::Refused, + ) + .await; + let mut buffer = vec![0_u8; MAX_DNS_MESSAGE]; + let length = timeout(Duration::from_secs(5), client.recv(&mut buffer)) + .await + .expect("the gateway answered") + .unwrap(); + buffer.truncate(length); + let response = Message::from_vec(&buffer).unwrap(); + assert_eq!(response.metadata.response_code, ResponseCode::Refused); + assert_eq!(response.metadata.id, 4242); + } + #[test] fn dns_health_requires_consecutive_failures_and_recovery_successes() { let mut health = DnsHealthState::new(3, 2);