use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use futures::future::join_all;
use tokio::sync::Semaphore;
use tracing::{debug, instrument};
use super::analysis::{
analyze_results, build_nameserver_consensus, build_nameserver_inconsistencies, PerVantage,
};
use super::servers::default_dns_servers;
use super::types::{DnsServer, NameserverDetails, PropagationResult, ServerResult};
use crate::dns::query::{DnsQueryResult, DnsStatus};
use crate::dns::records::{RecordData, RecordType};
use crate::dns::resolver::{DnsResolver, ServerReply};
use crate::dns::DEFAULT_DNS_TIMEOUT;
use crate::error::Result;
const MAX_CONCURRENT_NS_LOOKUPS: usize = 50;
const MAX_ENRICHED_NAMESERVERS: usize = 13;
fn per_server_budget(query_timeout: Duration) -> Duration {
query_timeout
.saturating_mul(2)
.saturating_add(Duration::from_secs(1))
}
fn nameservers_to_enrich(results: &[ServerResult]) -> Vec<String> {
let mut reported: HashMap<String, usize> = HashMap::new();
for sr in results {
let names: HashSet<String> = sr
.records
.iter()
.filter_map(|r| match &r.data {
RecordData::NS { nameserver } => Some(nameserver.to_ascii_lowercase()),
_ => None,
})
.collect();
for name in names {
*reported.entry(name).or_insert(0) += 1;
}
}
let mut names: Vec<(String, usize)> = reported.into_iter().collect();
names.sort_by(|(a, a_n), (b, b_n)| b_n.cmp(a_n).then_with(|| a.cmp(b)));
if names.len() > MAX_ENRICHED_NAMESERVERS {
debug!(
nameservers = names.len(),
enriched = MAX_ENRICHED_NAMESERVERS,
"Too many nameserver hostnames; enriching only the most-reported"
);
}
names
.into_iter()
.take(MAX_ENRICHED_NAMESERVERS)
.map(|(name, _)| name)
.collect()
}
#[derive(Debug, Clone)]
pub struct PropagationChecker {
resolver: DnsResolver,
servers: Vec<DnsServer>,
query_timeout: Duration,
}
impl Default for PropagationChecker {
fn default() -> Self {
Self::new()
}
}
impl PropagationChecker {
pub fn new() -> Self {
Self {
resolver: DnsResolver::new(),
servers: default_dns_servers().to_vec(),
query_timeout: DEFAULT_DNS_TIMEOUT,
}
}
pub fn from_config(config: &crate::config::SeerConfig) -> Self {
Self::new().with_timeout(config.dns_timeout())
}
pub fn with_servers(mut self, servers: Vec<DnsServer>) -> Self {
self.servers = servers;
self
}
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.resolver = DnsResolver::new().with_timeout(timeout);
self.query_timeout = timeout;
self
}
#[instrument(skip(self), fields(domain = %domain, record_type = %record_type))]
pub async fn check(&self, domain: &str, record_type: RecordType) -> Result<PropagationResult> {
let domain = crate::dns::resolver::prepare_query(domain, record_type)?;
if record_type == RecordType::SRV
&& crate::dns::resolver::parse_srv_query(&domain).is_none()
{
return Err(crate::dns::resolver::srv_format_error());
}
debug!(servers = self.servers.len(), "Starting propagation check");
let budget = per_server_budget(self.query_timeout);
let futures: Vec<_> = self
.servers
.iter()
.map(|server| self.query_server_bounded(&domain, record_type, server.clone(), budget))
.collect();
let results = join_all(futures).await;
let servers_checked = results.len();
let servers_responding = results.iter().filter(|r| r.success).count();
let outcome = analyze_results(&results, record_type);
let nameserver_details = if record_type == RecordType::NS {
self.resolve_nameserver_details(&results, budget).await
} else {
None
};
Ok(PropagationResult {
domain,
record_type,
servers_checked,
servers_responding,
propagation_percentage: outcome.propagation_percentage,
results,
consensus_values: outcome.consensus_values,
inconsistencies: outcome.inconsistencies,
unreachable_servers: outcome.unreachable_servers,
dnssec_validated: false,
nameserver_details,
})
}
async fn resolve_nameserver_details(
&self,
results: &[ServerResult],
budget: Duration,
) -> Option<NameserverDetails> {
let enriched = nameservers_to_enrich(results);
if enriched.is_empty() {
return None;
}
let sem = Arc::new(Semaphore::new(MAX_CONCURRENT_NS_LOOKUPS));
let deadline = tokio::time::Instant::now() + budget;
let mut tasks = Vec::new();
for sr in results.iter().filter(|sr| sr.success) {
for ns in &enriched {
let resolver = self.resolver.clone();
let server_ip = sr.server.ip.clone();
let ns = ns.clone();
let sem = sem.clone();
tasks.push(async move {
let lookups = async {
let _permit = sem.acquire().await.ok();
tokio::join!(
resolver.resolve(&ns, RecordType::A, Some(&server_ip)),
resolver.resolve(&ns, RecordType::AAAA, Some(&server_ip)),
)
};
let ips = match tokio::time::timeout_at(deadline, lookups).await {
Ok((a, aaaa)) => vantage_addresses(a, aaaa),
Err(_) => None,
};
(server_ip, ns, ips)
});
}
}
let outputs = join_all(tasks).await;
let mut per_vantage: PerVantage = HashMap::new();
for (server_ip, ns, ips) in outputs {
if let Some(ips) = ips {
per_vantage.entry(server_ip).or_default().insert(ns, ips);
}
}
let consensus = build_nameserver_consensus(results, &per_vantage, &enriched);
let inconsistencies = build_nameserver_inconsistencies(results, &per_vantage, &consensus);
Some(NameserverDetails {
consensus,
per_vantage,
inconsistencies,
})
}
#[cfg(test)]
fn with_resolver(mut self, resolver: DnsResolver, timeout: Duration) -> Self {
self.resolver = resolver;
self.query_timeout = timeout;
self
}
#[cfg(test)]
fn empty_for_tests() -> Self {
Self::new().with_servers(Vec::new())
}
async fn query_server_bounded(
&self,
domain: &str,
record_type: RecordType,
server: DnsServer,
budget: Duration,
) -> ServerResult {
let start = Instant::now();
match tokio::time::timeout(
budget,
self.query_server(domain, record_type, server.clone()),
)
.await
{
Ok(result) => result,
Err(_) => {
debug!(
server = %server.name,
budget_secs = budget.as_secs(),
"Server exceeded per-server budget; recording as unreachable"
);
ServerResult {
server,
records: vec![],
response_time_ms: start.elapsed().as_millis() as u64,
success: false,
error: Some("timed out".to_string()),
status: None,
}
}
}
}
async fn query_server(
&self,
domain: &str,
record_type: RecordType,
server: DnsServer,
) -> ServerResult {
let start = Instant::now();
let reply = self
.resolver
.query_server(domain, record_type, &server.ip)
.await;
let response_time_ms = start.elapsed().as_millis() as u64;
let (records, status, error) = match reply {
Ok(ServerReply::Response(response)) => read_reply(response, record_type),
Ok(ServerReply::Silent(reason)) => (vec![], None, Some(reason)),
Err(e) => {
debug!(server = %server.name, error = %e, "Server query failed");
(vec![], None, Some(e.sanitized_message()))
}
};
debug!(
server = %server.name,
records = records.len(),
status = ?status,
error = ?error,
time_ms = response_time_ms,
"Server replied"
);
ServerResult {
server,
success: error.is_none(),
records,
response_time_ms,
error,
status,
}
}
}
fn vantage_addresses(
a: Result<Vec<crate::dns::DnsRecord>>,
aaaa: Result<Vec<crate::dns::DnsRecord>>,
) -> Option<Vec<String>> {
let (a, aaaa) = (a.ok()?, aaaa.ok()?);
let mut ips: Vec<String> = a
.iter()
.chain(&aaaa)
.filter_map(|r| r.data.address().map(str::to_string))
.collect();
ips.sort();
ips.dedup();
Some(ips)
}
pub(in crate::dns) fn read_reply(
response: DnsQueryResult,
record_type: RecordType,
) -> (
Vec<crate::dns::DnsRecord>,
Option<DnsStatus>,
Option<String>,
) {
let status = Some(response.status);
if response.referral_zone().is_some() {
return (
vec![],
status,
Some("referral only (not a recursive resolver)".into()),
);
}
match response.status {
DnsStatus::NoError | DnsStatus::NxDomain if !response.failed_types.is_empty() => {
let missing: Vec<String> = response
.failed_types
.iter()
.map(|f| format!("{} ({})", f.record_type, f.error))
.collect();
(
response.answers,
status,
Some(format!(
"incomplete answer, no reply for {}",
missing.join(", ")
)),
)
}
DnsStatus::NoError | DnsStatus::NxDomain => {
let records = if record_type == RecordType::ANY {
response.answers
} else {
response
.records()
.filter(|r| r.record_type == record_type)
.cloned()
.collect()
};
(records, status, None)
}
other => (vec![], status, Some(other.to_string())),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dns::test_support::MockReply;
use crate::error::SeerError;
#[tokio::test]
async fn check_stores_normalized_domain() {
let checker = PropagationChecker::empty_for_tests();
let result = checker
.check("HTTPS://WWW.Example.COM/some/path", RecordType::A)
.await
.expect("check with empty server list should succeed");
assert_eq!(
result.domain, "www.example.com",
"stored domain must be the normalized form, not the raw input"
);
assert_eq!(result.servers_checked, 0);
assert_eq!(result.servers_responding, 0);
}
#[tokio::test]
async fn check_accepts_ipv6_ptr_literal() {
let checker = PropagationChecker::empty_for_tests();
let result = checker
.check("2606:4700:4700::1111", RecordType::PTR)
.await
.expect("IPv6 PTR literal must pass normalization");
assert_eq!(result.domain, "2606:4700:4700::1111");
}
#[tokio::test]
async fn check_rejects_srv_against_bare_domain() {
let checker = PropagationChecker::empty_for_tests();
let err = checker
.check("example.com", RecordType::SRV)
.await
.expect_err("bare-domain SRV must be rejected as an input error");
assert!(
matches!(err, SeerError::InvalidInput(_)),
"expected InvalidInput, got: {err:?}"
);
}
#[tokio::test]
async fn check_allows_well_formed_srv_query() {
let checker = PropagationChecker::empty_for_tests();
let result = checker
.check("_sip._tcp.example.com", RecordType::SRV)
.await
.expect("well-formed SRV query must pass the upfront guard");
assert_eq!(result.servers_checked, 0);
}
#[test]
fn per_server_budget_allows_a_retry_plus_margin() {
assert_eq!(
per_server_budget(Duration::from_secs(5)),
Duration::from_secs(11)
);
assert_eq!(
per_server_budget(Duration::from_secs(2)),
Duration::from_secs(5)
);
assert_eq!(
per_server_budget(Duration::from_secs(0)),
Duration::from_secs(1)
);
}
#[tokio::test]
async fn check_returns_partial_results_when_servers_fail() {
let servers = vec![
DnsServer::new("Reserved-A", "10.0.0.1", "Test", "Test"),
DnsServer::new("Reserved-B", "192.168.1.1", "Test", "Test"),
];
let checker = PropagationChecker::new().with_servers(servers);
let result = checker
.check("example.com", RecordType::A)
.await
.expect("failing servers must yield Ok partial results, not a whole-check error");
assert_eq!(result.servers_checked, 2);
assert_eq!(result.servers_responding, 0);
assert_eq!(result.unreachable_servers.len(), 2);
}
#[tokio::test]
async fn server_replies_are_classified_with_their_reason() {
use crate::dns::test_support::{
a_rdata, cname_rdata, mock_dns_resolver, record, spawn_mock_dns_fn, MockReply,
};
use hickory_resolver::proto::op::ResponseCode;
let port = spawn_mock_dns_fn(|name, _| match name {
"answer.seer.test" => MockReply::Answer(vec![a_rdata([192, 0, 2, 1])]),
"www.seer.test" => MockReply::Records(vec![
record("www.seer.test", 300, cname_rdata("answer.seer.test")),
record("answer.seer.test", 300, a_rdata([192, 0, 2, 1])),
]),
"nx.seer.test" => MockReply::NxDomainWithSoa("seer.test"),
"nodata.seer.test" => MockReply::NoDataWithSoa("seer.test"),
"refused.seer.test" => MockReply::Rcode(ResponseCode::Refused),
"servfail.seer.test" => MockReply::Rcode(ResponseCode::ServFail),
"referral.seer.test" => MockReply::Delegation {
zone: "referral.seer.test".into(),
servers: vec![("ns1.elsewhere.test".into(), vec![])],
},
_ => MockReply::NoReply,
})
.await;
let timeout = Duration::from_millis(300);
let checker = PropagationChecker::new()
.with_servers(vec![DnsServer::new("Mock", "127.0.0.1", "Test", "Test")])
.with_resolver(mock_dns_resolver(port).with_timeout(timeout), timeout);
type Case<'a> = (
&'a str,
bool,
Option<DnsStatus>,
Option<&'a str>,
&'a [&'a str],
);
let cases: [Case; 8] = [
(
"answer.seer.test",
true,
Some(DnsStatus::NoError),
None,
&["192.0.2.1"],
),
(
"www.seer.test",
true,
Some(DnsStatus::NoError),
None,
&["192.0.2.1"],
),
("nx.seer.test", true, Some(DnsStatus::NxDomain), None, &[]),
(
"nodata.seer.test",
true,
Some(DnsStatus::NoError),
None,
&[],
),
(
"refused.seer.test",
false,
Some(DnsStatus::Refused),
Some("REFUSED"),
&[],
),
(
"servfail.seer.test",
false,
Some(DnsStatus::ServFail),
Some("SERVFAIL"),
&[],
),
(
"referral.seer.test",
false,
Some(DnsStatus::NoError),
Some("referral only (not a recursive resolver)"),
&[],
),
("silent.seer.test", false, None, Some("timed out"), &[]),
];
for (name, success, status, error, values) in cases {
let result = checker.check(name, RecordType::A).await.expect(name);
let server = &result.results[0];
assert_eq!(server.success, success, "{name}");
assert_eq!(server.status, status, "{name}");
assert_eq!(server.error.as_deref(), error, "{name}");
let got: Vec<String> = server.records.iter().map(|r| r.format_short()).collect();
assert_eq!(got, values, "{name}");
assert_eq!(result.servers_responding, usize::from(success), "{name}");
}
let nx = checker.check("nx.seer.test", RecordType::A).await.unwrap();
assert_eq!(nx.results[0].empty_answer_label(), "NXDOMAIN");
assert_eq!(nx.empty_consensus_label(), Some("NXDOMAIN"));
let nodata = checker
.check("nodata.seer.test", RecordType::A)
.await
.unwrap();
assert_eq!(nodata.empty_consensus_label(), Some("NODATA"));
}
#[test]
fn from_config_applies_dns_timeout() {
let mut config = crate::config::SeerConfig::default();
config.timeouts.dns_secs = 9;
let checker = PropagationChecker::from_config(&config);
assert_eq!(checker.query_timeout, Duration::from_secs(9));
assert_eq!(checker.resolver.timeout(), Duration::from_secs(9));
assert_eq!(checker.servers.len(), default_dns_servers().len());
}
#[test]
fn failed_vantage_lookup_is_no_data() {
use crate::dns::{DnsRecord, RecordData};
let a = |ip: &str| DnsRecord {
name: "ns1.example.com".into(),
record_type: RecordType::A,
ttl: 300,
data: RecordData::A { address: ip.into() },
};
let failed = || Err(SeerError::DnsError("SERVFAIL".into()));
assert_eq!(
vantage_addresses(Ok(vec![a("192.0.2.2"), a("192.0.2.1")]), Ok(vec![])),
Some(vec!["192.0.2.1".to_string(), "192.0.2.2".to_string()])
);
assert_eq!(vantage_addresses(Ok(vec![]), Ok(vec![])), Some(vec![]));
assert_eq!(vantage_addresses(Ok(vec![a("192.0.2.1")]), failed()), None);
assert_eq!(vantage_addresses(failed(), failed()), None);
}
#[tokio::test]
async fn slow_vantage_lookup_keeps_the_answers_that_arrived() {
use crate::dns::test_support::{a_rdata, mock_dns_resolver, spawn_mock_dns_fn, MockReply};
use hickory_resolver::proto::rr::rdata as wire;
use hickory_resolver::proto::rr::{Name, RData};
let ns = |host: &str| RData::NS(wire::NS(Name::from_ascii(host).unwrap()));
let port = spawn_mock_dns_fn(
move |name, qtype| match (name, qtype.to_string().as_str()) {
("seer.test", "NS") => {
MockReply::Answer(vec![ns("ns1.seer.test."), ns("ns2.seer.test.")])
}
("ns1.seer.test", "A") => MockReply::Answer(vec![a_rdata([192, 0, 2, 1])]),
("ns1.seer.test", _) => MockReply::NoData,
_ => MockReply::NoReply,
},
)
.await;
let timeout = Duration::from_millis(200);
let checker = PropagationChecker::new()
.with_servers(vec![DnsServer::new("Mock", "127.0.0.1", "Test", "Test")])
.with_resolver(mock_dns_resolver(port).with_timeout(timeout), timeout);
let result = checker
.check("seer.test", RecordType::NS)
.await
.expect("check");
let details = result.nameserver_details.expect("NS details");
let vantage = &details.per_vantage["127.0.0.1"];
assert_eq!(vantage["ns1.seer.test."], vec!["192.0.2.1".to_string()]);
assert!(!vantage.contains_key("ns2.seer.test."), "{vantage:?}");
assert!(details.inconsistencies.is_empty());
}
#[test]
fn read_reply_keeps_an_incomplete_any_answer_out_of_the_consensus() {
let record = crate::dns::DnsRecord {
name: "example.com".into(),
record_type: RecordType::A,
ttl: 300,
data: RecordData::A {
address: "192.0.2.1".into(),
},
};
let mut response = DnsQueryResult {
name: "example.com".into(),
record_type: RecordType::ANY,
server: Some("8.8.8.8".into()),
answered_locally: false,
status: DnsStatus::NoError,
flags: vec![],
answers: vec![record],
authority: vec![],
failed_types: vec![crate::dns::FailedType {
record_type: RecordType::TXT,
error: "8.8.8.8: timed out".into(),
}],
wildcard: None,
query_time_ms: 5,
};
let (records, status, error) = read_reply(response.clone(), RecordType::ANY);
assert_eq!(records.len(), 1, "the records that arrived still show");
assert_eq!(status, Some(DnsStatus::NoError));
let error = error.expect("an incomplete answer is not a success");
assert!(error.contains("TXT (8.8.8.8: timed out)"), "{error}");
response.failed_types.clear();
assert_eq!(read_reply(response, RecordType::ANY).2, None);
}
fn many_ns(count: usize) -> MockReply {
use hickory_resolver::proto::rr::rdata as wire;
use hickory_resolver::proto::rr::{Name, RData};
MockReply::Answer(
(0..count)
.map(|i| {
RData::NS(wire::NS(
Name::from_ascii(format!("ns{i}.seer.test.")).unwrap(),
))
})
.collect(),
)
}
fn mock_servers(count: usize) -> Vec<DnsServer> {
(0..count)
.map(|i| DnsServer::new(&format!("Mock{i}"), "127.0.0.1", "Test", "Test"))
.collect()
}
#[tokio::test]
async fn ns_enrichment_is_capped() {
use crate::dns::test_support::{a_rdata, mock_dns_resolver, spawn_mock_dns_fn};
use std::sync::Mutex;
let asked: Arc<Mutex<HashSet<String>>> = Arc::default();
let log = asked.clone();
let port = spawn_mock_dns_fn(
move |name, qtype| match (name, qtype.to_string().as_str()) {
("seer.test", "NS") => many_ns(50),
(host, "A") => {
log.lock().unwrap().insert(host.to_string());
MockReply::Answer(vec![a_rdata([192, 0, 2, 1])])
}
_ => MockReply::NoData,
},
)
.await;
let timeout = Duration::from_millis(500);
let checker = PropagationChecker::new()
.with_servers(mock_servers(2))
.with_resolver(mock_dns_resolver(port).with_timeout(timeout), timeout);
let result = checker.check("seer.test", RecordType::NS).await.unwrap();
assert!(result.results.iter().all(|r| r.records.len() == 50));
assert_eq!(asked.lock().unwrap().len(), MAX_ENRICHED_NAMESERVERS);
let details = result.nameserver_details.expect("NS details");
assert_eq!(details.consensus.len(), MAX_ENRICHED_NAMESERVERS);
assert_eq!(
details.per_vantage["127.0.0.1"].len(),
MAX_ENRICHED_NAMESERVERS
);
}
#[tokio::test]
async fn ns_enrichment_is_held_to_one_deadline() {
use crate::dns::test_support::{mock_dns_resolver, spawn_mock_dns_fn};
const SERVERS: usize = 100;
let port = spawn_mock_dns_fn(|name, qtype| match (name, qtype.to_string().as_str()) {
("seer.test", "NS") => many_ns(MAX_ENRICHED_NAMESERVERS),
_ => MockReply::NoReply,
})
.await;
let timeout = Duration::from_millis(200);
let budget = per_server_budget(timeout);
let checker = PropagationChecker::new()
.with_servers(mock_servers(SERVERS))
.with_resolver(mock_dns_resolver(port).with_timeout(timeout), timeout);
let started = Instant::now();
let result = checker.check("seer.test", RecordType::NS).await.unwrap();
let elapsed = started.elapsed();
assert!(elapsed < budget * 2 + timeout, "took {elapsed:?}");
assert_eq!(result.servers_responding, SERVERS);
assert!(result
.results
.iter()
.all(|r| r.records.len() == MAX_ENRICHED_NAMESERVERS));
let details = result.nameserver_details.expect("NS details");
assert!(details.per_vantage.is_empty(), "{:?}", details.per_vantage);
assert!(details.inconsistencies.is_empty());
}
#[test]
fn nameservers_to_enrich_prefers_the_most_reported() {
use crate::dns::{DnsRecord, RecordData};
let server = |names: &[&str]| ServerResult {
server: DnsServer::new("Mock", "127.0.0.1", "Test", "Test"),
records: names
.iter()
.map(|n| DnsRecord {
name: "seer.test".into(),
record_type: RecordType::NS,
ttl: 300,
data: RecordData::NS {
nameserver: (*n).into(),
},
})
.collect(),
response_time_ms: 1,
success: true,
error: None,
status: None,
};
let mut lots: Vec<String> = (0..20).map(|i| format!("a{i:02}.test.")).collect();
lots.push("Z.TEST.".into());
let lots: Vec<&str> = lots.iter().map(String::as_str).collect();
let picked = nameservers_to_enrich(&[server(&lots), server(&["z.test.", "z.test."])]);
assert_eq!(picked.len(), MAX_ENRICHED_NAMESERVERS);
assert_eq!(picked[0], "z.test.");
assert_eq!(picked[1], "a00.test.");
}
}