1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
use alloc::sync::Arc;
use bytes::Bytes;
use octseq::Octets;
use tokio::sync::Semaphore;
use tokio::sync::mpsc::{Receiver, UnboundedSender};
use tracing::{debug, error};
use crate::base::iana::OptRcode;
use crate::base::rdata::RecordData;
use crate::base::wire::Composer;
use crate::base::{Message, Name, Rtype};
use crate::net::server::batcher::ResourceRecordBatcher;
use crate::net::server::middleware::xfr::util::add_to_stream;
use crate::net::server::service::ServiceResult;
use crate::net::server::util::mk_builder_for_target;
use crate::zonetree::{Answer, SharedRrset};
use super::batcher::{BatchReadyError, XfrRrBatcher};
//------------ BatchingRrResponder ---------------------------------------------
pub struct BatchingRrResponder<RequestOctets, Target> {
msg: Arc<Message<RequestOctets>>,
zone_soa_answer: Answer,
batcher_rx: Receiver<(Name<Bytes>, SharedRrset)>,
response_tx: UnboundedSender<ServiceResult<Target>>,
compatibility_mode: bool,
soft_byte_limit: usize,
must_fit_in_single_message: bool,
batcher_semaphore: Arc<Semaphore>,
}
impl<RequestOctets, Target> BatchingRrResponder<RequestOctets, Target>
where
RequestOctets: Octets + Send + Sync + 'static + Unpin,
Target: Composer + Default + Send + Sync + 'static,
{
#[allow(clippy::too_many_arguments)]
pub fn new(
msg: Arc<Message<RequestOctets>>,
zone_soa_answer: Answer,
batcher_rx: Receiver<(Name<Bytes>, SharedRrset)>,
response_tx: UnboundedSender<ServiceResult<Target>>,
compatibility_mode: bool,
soft_byte_limit: usize,
must_fit_in_single_message: bool,
batcher_semaphore: Arc<Semaphore>,
) -> Self {
Self {
msg,
zone_soa_answer,
batcher_rx,
response_tx,
compatibility_mode,
soft_byte_limit,
must_fit_in_single_message,
batcher_semaphore,
}
}
pub async fn run(mut self) -> Result<(), OptRcode> {
// Limit the number of concurrently running XFR batching
// operations.
if self.batcher_semaphore.acquire().await.is_err() {
error!("Internal error: Failed to acquire XFR batcher semaphore");
return Err(OptRcode::SERVFAIL);
}
// SAFETY: msg.sole_question() was already checked in
// get_relevant_question().
let qclass = self.msg.sole_question().unwrap().qclass();
// Note: NSD apparently uses name compresson on AXFR responses
// because AXFR responses they typically contain lots of
// alphabetically ordered duplicate names which compress well. NSD
// limits AXFR responses to 16,383 bytes because DNS name
// compression uses a 14-bit offset (2^14-1=16383) from the start
// of the message to the first occurence of a name instead of
// repeating the name, and name compression is less effective
// over 16383 bytes. (Credit: Wouter Wijngaards)
//
// TODO: Once we start supporting name compression in responses decide
// if we want to behave the same way.
let hard_rr_limit = match self.compatibility_mode {
true => Some(1),
false => None,
};
let mut batcher = XfrRrBatcher::build(
self.msg.clone(),
self.response_tx.clone(),
Some(self.soft_byte_limit),
hard_rr_limit,
self.must_fit_in_single_message,
);
let mut last_rr_rtype = None;
while let Some((owner, rrset)) = self.batcher_rx.recv().await {
for rr in rrset.data() {
last_rr_rtype = Some(rr.rtype());
if let Err(err) =
batcher.push((owner.clone(), qclass, rrset.ttl(), rr))
{
match err {
BatchReadyError::MustFitInSingleMessage => {
// https://datatracker.ietf.org/doc/html/rfc1995#section-2
// 2. Brief Description of the Protocol
// ..
// "If the UDP reply does not fit, the
// query is responded to with a single SOA
// record of the server's current version
// to inform the client that a TCP query
// should be initiated."
debug_assert!(self.must_fit_in_single_message);
debug!(
"Responding to IXFR with single SOA because response does not fit in a single UDP reply"
);
let builder = mk_builder_for_target();
let resp = self
.zone_soa_answer
.to_message(&self.msg, builder);
add_to_stream(resp, &self.response_tx);
return Ok(());
}
BatchReadyError::PushError(err) => {
error!(
"Internal error: Failed to send RR to batcher: {err}"
);
return Err(OptRcode::SERVFAIL);
}
BatchReadyError::SendError => {
debug!(
"Batcher was unable to send completed batch. Was the receiver dropped?"
);
return Err(OptRcode::SERVFAIL);
}
}
}
}
}
if let Err(err) = batcher.finish() {
debug!("Batcher was unable to finish: {err}");
return Err(OptRcode::SERVFAIL);
}
if last_rr_rtype != Some(Rtype::SOA) {
error!(
"Internal error: Last RR was {}, expected SOA",
last_rr_rtype.unwrap()
);
return Err(OptRcode::SERVFAIL);
}
Ok(())
}
}