diff --git a/CHANGELOG.md b/CHANGELOG.md index 65cb692..7cc8e6f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,8 @@ # Unreleased * Restore DTLS buffer reuse and reject unpooled buffer returns #167 + * Process DTLS 1.2 handshake messages by sequence number when records arrive out of order + * Resend a handshake flight at most once per timer period on duplicates of the peer's flight # 0.7.4 diff --git a/src/dtls12/engine.rs b/src/dtls12/engine.rs index b60fd54..0d0097d 100644 --- a/src/dtls12/engine.rs +++ b/src/dtls12/engine.rs @@ -3,6 +3,8 @@ use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::time::{Duration, Instant}; +use arrayvec::ArrayVec; + use super::queue::{QueueRx, QueueTx}; use crate::buffer::{Buf, BufferPool, TmpBuf}; use crate::crypto::{Aad, Iv, Nonce}; @@ -89,6 +91,15 @@ pub struct Engine { /// Timeout for the current flight flight_timeout: Timeout, + /// Whether a duplicate of the peer's previous flight may still trigger a resend + /// of the current flight before the flight timer fires. + /// + /// RFC 6347 §4.2.4: retransmission is driven by the timer, with exponential + /// backoff. A duplicate says the peer is missing our flight, but answering every + /// duplicate datagram lets two waiting peers bounce flights at network speed. One + /// early resend per timer period is enough; the timer does the rest. + flight_dupe_resend_allowed: bool, + /// Global timeout for the entire connect operation. connect_timeout: Timeout, @@ -166,6 +177,7 @@ impl Engine { flight_saved_records: Vec::new(), flight_backoff, flight_timeout: Timeout::Unarmed, + flight_dupe_resend_allowed: true, connect_timeout: Timeout::Unarmed, release_app_data: false, peer_handshake_confirmed: false, @@ -283,7 +295,7 @@ impl Engine { // drive a resend. if let Some(dupe_seq) = maybe_dupe_seq { if dupe_seq < self.peer_handshake_seq_no && !self.peer_handshake_confirmed { - if let Err(error) = self.flight_resend("dupe triggers resend") { + if let Err(error) = self.flight_resend_on_dupe() { self.recycle_incoming(incoming); return Err(error); } @@ -425,6 +437,7 @@ impl Engine { ); let timeout = now + self.flight_backoff.rto(); self.flight_timeout = Timeout::Armed(timeout); + self.flight_dupe_resend_allowed = true; self.flight_resend("flight timeout")?; } else { return Err(Error::Timeout(crate::TimeoutError::Handshake)); @@ -557,6 +570,7 @@ impl Engine { self.flight_backoff.reset(&mut self.rng); self.flight_clear_resends(); self.flight_timeout = Timeout::Unarmed; + self.flight_dupe_resend_allowed = true; } pub fn flight_stop_resend_timers(&mut self) { @@ -581,6 +595,23 @@ impl Engine { } } + /// Resend the current flight because the peer resent its previous one. + fn flight_resend_on_dupe(&mut self) -> Result<(), Error> { + // Once our final flight is sent the timers are stopped, and each retransmission + // of the peer's last flight must be answered (RFC 6347 §4.2.4). The peer's own + // timer paces those. While we still wait for the peer, at most one duplicate per + // timer period triggers a resend. + if self.flight_timeout != Timeout::Disabled { + if !self.flight_dupe_resend_allowed { + trace!("Duplicate ignored: flight already resent in this timer period"); + return Ok(()); + } + self.flight_dupe_resend_allowed = false; + } + + self.flight_resend("dupe triggers resend") + } + fn flight_resend(&mut self, reason: &str) -> Result<(), Error> { debug!("Resending flight due to {}", reason); @@ -616,58 +647,15 @@ impl Engine { } pub fn has_complete_handshake(&mut self, wanted: MessageType) -> bool { - self.has_complete_handshake_with_seq(wanted, self.peer_handshake_seq_no) - } - - fn has_complete_handshake_with_seq(&mut self, wanted: MessageType, expected_seq: u16) -> bool { - let mut skip_handled = self - .queue_rx - .iter() - .flat_map(|i| i.records().iter()) - .skip_while(|r| r.is_handled()) - // Cap to MAX_DEFRAGMENT_PACKETS to avoid misbehaving peers - .take(MAX_DEFRAGMENT_PACKETS) - .flat_map(|r| r.handshakes().iter()) - .skip_while(|h| h.is_handled()) - .peekable(); - - let maybe_first_handshake = skip_handled.peek(); - - let Some(first) = maybe_first_handshake else { - return false; - }; - - if first.header.message_seq != expected_seq { - return false; - } + self.discard_stale_handshakes(); - if first.header.msg_type != wanted { - return false; - } - - let wanted_seq = first.header.message_seq; - let wanted_length = first.header.length; - let mut last_fragment_end = 0; - - for h in skip_handled { - // A different seq means we're looking at a different handshake - if wanted_seq != h.header.message_seq { - continue; - } - - // Check fragment contiguity - if h.header.fragment_offset != last_fragment_end { - return false; - } - last_fragment_end = h.header.fragment_offset + h.header.fragment_length; - - // Found the last fragment to complete the wanted handshake. - if last_fragment_end == wanted_length { - return true; - } - } - - false + complete_handshake_fragments( + &self.queue_rx, + self.peer_encryption_enabled, + wanted, + self.peer_handshake_seq_no, + ) + .is_some() } pub fn next_handshake( @@ -675,23 +663,22 @@ impl Engine { wanted: MessageType, defragment_buffer: &mut Buf, ) -> Result, InternalError> { - if !self.has_complete_handshake(wanted) { + self.discard_stale_handshakes(); + + let Some(fragments) = complete_handshake_fragments( + &self.queue_rx, + self.peer_encryption_enabled, + wanted, + self.peer_handshake_seq_no, + ) else { return Ok(None); - } - - let iter = self - .queue_rx - .iter() - .flat_map(|i| i.records().iter()) - .skip_while(|r| r.is_handled()) - .flat_map(|r| r.handshakes().iter().map(move |h| (h, r.buffer()))) - .skip_while(|(h, _)| h.is_handled()); + }; // This sets the handled flag on the handshake. // Passing Some(&mut self.transcript) to have defragment write to transcript // before creating the handshake, avoiding borrow conflicts. let handshake = Handshake::defragment( - iter, + fragments.into_iter(), defragment_buffer, self.cipher_suite, Some(&mut self.transcript), @@ -700,15 +687,25 @@ impl Engine { // Move the expected seq_no along self.peer_handshake_seq_no = handshake.header.message_seq + 1; + // Other copies of this message (resends) are stale now. + self.discard_stale_handshakes(); + Ok(Some(handshake)) } pub(crate) fn next_record(&mut self, ctype: ContentType) -> Option<&Record> { + self.discard_stale_handshakes(); + + let peer_encryption_enabled = self.peer_encryption_enabled; + let record = self .queue_rx .iter() .flat_map(|i| i.records().iter()) - .find(|r| !r.is_handled())?; + .filter(|r| !r.is_handled()) + // A record of the next epoch can arrive before the ChangeCipherSpec that + // starts the epoch. It waits for the keys and must not block that record. + .find(|r| peer_encryption_enabled || r.record().sequence.epoch == 0)?; if record.record().content_type != ctype { return None; @@ -719,6 +716,28 @@ impl Engine { Some(record) } + /// Mark queued copies of handshake messages that were already processed as handled. + /// + /// A datagram can carry both new messages and copies of processed ones (a resent + /// flight, or a copy whose records were reordered). The stale copies must neither + /// block the next message nor keep their datagram in the receive queue. + fn discard_stale_handshakes(&self) { + let records = self + .queue_rx + .iter() + .flat_map(|i| i.records().iter()) + // Records of an epoch whose keys are not in place are not parsed yet. + .filter(|r| self.peer_encryption_enabled || r.record().sequence.epoch == 0); + + for record in records { + for handshake in record.handshakes() { + if handshake.header.message_seq < self.peer_handshake_seq_no { + handshake.set_handled(); + } + } + } + } + /// Mark any pending ChangeCipherSpec records as handled and purge them. /// We can accumulate multiple ChangeCipherSpec due to resends. Since they /// don't have any Handshake message_seq and each resend gives a new DTLSRecord @@ -1272,6 +1291,81 @@ impl Engine { } } +/// The fragments of handshake message `seq`, in fragment order, if they are all queued +/// and the message is of type `wanted`. +/// +/// RFC 6347 §4.2.2: handshake messages can arrive out of order, and so can the records +/// inside one datagram. The next message is looked up by its message_seq wherever it +/// sits in the queue, and later messages stay buffered until it is their turn. +fn complete_handshake_fragments( + queue_rx: &QueueRx, + peer_encryption_enabled: bool, + wanted: MessageType, + seq: u16, +) -> Option> { + let records = queue_rx + .iter() + .flat_map(|i| i.records().iter()) + .filter(|r| !r.is_handled()) + // Records of an epoch whose keys are not in place are not parsed yet. + .filter(|r| peer_encryption_enabled || r.record().sequence.epoch == 0) + // Cap to MAX_DEFRAGMENT_PACKETS to avoid misbehaving peers + .take(MAX_DEFRAGMENT_PACKETS); + + let mut fragments: ArrayVec<(&Handshake, &[u8]), MAX_DEFRAGMENT_PACKETS> = ArrayVec::new(); + + for record in records { + for h in record.handshakes() { + if h.is_handled() || h.header.message_seq != seq { + continue; + } + + // A resent copy of a fragment we already have. + let offset = h.header.fragment_offset; + if fragments + .iter() + .any(|(f, _)| f.header.fragment_offset == offset) + { + continue; + } + + fragments.try_push((h, record.buffer())).ok()?; + } + } + + fragments.sort_by_key(|(h, _)| h.header.fragment_offset); + + let (first, _) = fragments.first()?; + if first.header.msg_type != wanted { + return None; + } + + let length = first.header.length; + let mut last_fragment_end = 0; + let mut complete_at = None; + + for (index, (h, _)) in fragments.iter().enumerate() { + // Check fragment contiguity + if h.header.msg_type != wanted + || h.header.length != length + || h.header.fragment_offset != last_fragment_end + { + return None; + } + last_fragment_end = h.header.fragment_offset + h.header.fragment_length; + + // Found the last fragment to complete the wanted handshake. + if last_fragment_end == length { + complete_at = Some(index); + break; + } + } + + fragments.truncate(complete_at? + 1); + + Some(fragments) +} + impl RecordHandler for Engine { fn pop_buffer(&mut self) -> Buf { Engine::pop_buffer(self) diff --git a/src/dtls13/engine.rs b/src/dtls13/engine.rs index 36a5956..bd5a100 100644 --- a/src/dtls13/engine.rs +++ b/src/dtls13/engine.rs @@ -137,6 +137,15 @@ pub struct Engine { /// Timeout for the current flight flight_timeout: Timeout, + /// Whether a duplicate of the peer's previous flight may still trigger a resend + /// of the current flight before the flight timer fires. + /// + /// Retransmission is driven by the timer, with exponential backoff (RFC 9147 + /// §5.8.1). A duplicate says the peer is missing our flight, but answering every + /// duplicate datagram lets two waiting peers bounce flights at network speed. One + /// early resend per timer period is enough; the timer does the rest. + flight_dupe_resend_allowed: bool, + /// Global timeout for the entire connect operation. connect_timeout: Timeout, @@ -258,6 +267,7 @@ impl Engine { flight_saved_records: ArrayVec::new(), flight_backoff, flight_timeout: Timeout::Unarmed, + flight_dupe_resend_allowed: true, connect_timeout: Timeout::Unarmed, release_app_data: false, exporter_master_secret: None, @@ -390,7 +400,7 @@ impl Engine { if let Some(dupe_seq) = maybe_dupe_seq { if dupe_seq < self.peer_handshake_seq_no { - if let Err(error) = self.flight_resend("dupe triggers resend") { + if let Err(error) = self.flight_resend_on_dupe() { self.recycle_incoming(incoming); return Err(error); } @@ -542,6 +552,7 @@ impl Engine { ); let timeout = now + self.flight_backoff.rto(); self.flight_timeout = Timeout::Armed(timeout); + self.flight_dupe_resend_allowed = true; self.flight_resend("flight timeout")?; } else { return Err(Error::Timeout(crate::TimeoutError::Handshake)); @@ -688,6 +699,7 @@ impl Engine { self.flight_backoff.reset(&mut self.rng); self.flight_clear_resends(); self.flight_timeout = Timeout::Unarmed; + self.flight_dupe_resend_allowed = true; } pub fn flight_stop_resend_timers(&mut self) { @@ -702,6 +714,23 @@ impl Engine { } } + /// Resend the current flight because the peer resent its previous one. + fn flight_resend_on_dupe(&mut self) -> Result<(), Error> { + // With the timers stopped (our last flight is sent or ACKed), each + // retransmission of the peer's last flight is answered; the peer's own timer + // paces those. While we still wait for the peer, at most one duplicate per + // timer period triggers a resend. + if self.flight_timeout != Timeout::Disabled { + if !self.flight_dupe_resend_allowed { + trace!("Duplicate ignored: flight already resent in this timer period"); + return Ok(()); + } + self.flight_dupe_resend_allowed = false; + } + + self.flight_resend("dupe triggers resend") + } + pub fn flight_resend(&mut self, reason: &str) -> Result<(), Error> { debug!("Resending flight due to {}", reason); let mut records = mem::take(&mut self.flight_saved_records); @@ -1943,6 +1972,7 @@ impl Engine { self.flight_backoff.reset(&mut self.rng); self.flight_clear_resends(); self.flight_timeout = Timeout::Unarmed; + self.flight_dupe_resend_allowed = true; let msg_seq = self.next_handshake_seq_no; self.next_handshake_seq_no += 1; diff --git a/tests/dtls12/reorder.rs b/tests/dtls12/reorder.rs index 5b75a37..06c2d94 100644 --- a/tests/dtls12/reorder.rs +++ b/tests/dtls12/reorder.rs @@ -401,3 +401,165 @@ fn dtls12_handles_delayed_burst_delivery() { "Server should connect despite delayed delivery" ); } + +/// Reverse the order of the records in a DTLS 1.2 datagram (whole records only). +#[cfg(feature = "rcgen")] +fn reverse_records(datagram: &[u8]) -> Vec { + let mut records = Vec::new(); + let mut i = 0usize; + while i + 13 <= datagram.len() { + let len = u16::from_be_bytes([datagram[i + 11], datagram[i + 12]]) as usize; + let end = (i + 13 + len).min(datagram.len()); + records.push(&datagram[i..end]); + i = end; + } + records.into_iter().rev().flatten().copied().collect() +} + +/// The most datagrams one side may send in [`run_reversed_final_flight`]'s 3 s. +/// +/// A side sends at most 3 flights, and each flight goes out once, plus at most one early +/// resend per retransmission timer period when duplicates arrive (two periods fit in 3 s), +/// plus a resend of the final flight for each duplicate of the peer's final flight. At +/// MTU 300 a flight is at most 5 datagrams. A resend storm sends one flight per round trip +/// (here one per 10 ms step, so hundreds). +#[cfg(feature = "rcgen")] +const MAX_DATAGRAMS_PER_SIDE: usize = 30; + +/// Run a handshake where the first client datagram carrying ChangeCipherSpec has its +/// records reversed (Finished before ChangeCipherSpec, and the flight's handshake +/// messages in reverse order). With `duplicate`, every datagram is also delivered twice. +/// +/// Asserts that neither side sends more than [`MAX_DATAGRAMS_PER_SIDE`] datagrams over +/// 3 s, and returns the time until both sides are connected. +#[cfg(feature = "rcgen")] +fn run_reversed_final_flight(config: Arc, duplicate: bool) -> Duration { + use dimpl::certificate::generate_self_signed_certificate; + + let client_cert = generate_self_signed_certificate().expect("gen client cert"); + let server_cert = generate_self_signed_certificate().expect("gen server cert"); + + let mtu = config.mtu(); + let cookie = config.use_server_cookie(); + + let start = Instant::now(); + let mut now = start; + + let mut client = Dtls::new_12(Arc::clone(&config), client_cert, now); + client.set_active(true); + + let mut server = Dtls::new_12(config, server_cert, now); + server.set_active(false); + + let mut reversed = false; + let mut client_connected = false; + let mut server_connected = false; + let mut connected_after = None; + let mut client_sent = 0; + let mut server_sent = 0; + + // 3 s: past the first retransmission timer, so a resend storm cannot hide. + for _ in 0..300 { + client.handle_timeout(now).expect("client timeout"); + server.handle_timeout(now).expect("server timeout"); + + let client_out = drain_outputs(&mut client); + let server_out = drain_outputs(&mut server); + + client_connected |= client_out.connected; + server_connected |= server_out.connected; + if client_connected && server_connected && connected_after.is_none() { + connected_after = Some(now - start); + } + + client_sent += client_out.packets.len(); + server_sent += server_out.packets.len(); + + let copies = if duplicate { 2 } else { 1 }; + + for p in &client_out.packets { + let has_ccs = parse_records(p).iter().any(|r| r.ctype == 20); + let p = if has_ccs && !reversed { + reversed = true; + reverse_records(p) + } else { + p.clone() + }; + for _ in 0..copies { + let _ = server.handle_packet(&p); + } + } + for p in &server_out.packets { + for _ in 0..copies { + let _ = client.handle_packet(p); + } + } + + now += Duration::from_millis(10); + } + + println!( + "mtu {mtu} cookie {cookie} duplicate {duplicate}: client sent {client_sent}, \ + server sent {server_sent} datagrams, connected after {connected_after:?}" + ); + + assert!( + reversed, + "the client's final flight should have been reversed" + ); + assert!( + client_sent <= MAX_DATAGRAMS_PER_SIDE && server_sent <= MAX_DATAGRAMS_PER_SIDE, + "mtu {mtu} cookie {cookie} duplicate {duplicate}: resend storm: client sent \ + {client_sent}, server sent {server_sent} datagrams in 3 s (at most \ + {MAX_DATAGRAMS_PER_SIDE} each)" + ); + + connected_after.unwrap_or_else(|| { + panic!( + "mtu {mtu} cookie {cookie} duplicate {duplicate}: handshake should complete \ + with a reversed final flight (client sent {client_sent}, server sent \ + {server_sent} datagrams)" + ) + }) +} + +#[test] +#[cfg(feature = "rcgen")] +fn dtls12_reversed_final_flight_completes_without_resend_storm() { + //! RFC 6347 §4.2.2: handshake messages are processed in message_seq order whatever + //! order their records arrive in, and a record of the next epoch (Finished) that + //! arrives before the ChangeCipherSpec starting that epoch waits for it. One + //! reordered copy of the final flight must complete the handshake on its own: no + //! retransmission, and (RFC 6347 §4.2.4) no flights bounced between the peers, also + //! when the network duplicates datagrams. + + let _ = env_logger::try_init(); + + let configs = [ + // As WebRTC stacks use it (ICE has already proved return routability). + Arc::new( + dimpl::Config::builder() + .use_server_cookie(false) + .build() + .expect("Failed to build config"), + ), + dtls12_config(), + // Small MTU: the final flight spans several datagrams and fragments. + dtls12_config_with_mtu(300), + ]; + + for config in configs { + for duplicate in [false, true] { + let mtu = config.mtu(); + let connected_after = run_reversed_final_flight(Arc::clone(&config), duplicate); + + // Below the 1 s initial retransmission timer (less its jitter): the reordered + // copy alone completed the handshake. + assert!( + connected_after < Duration::from_millis(500), + "mtu {mtu} duplicate {duplicate}: handshake should complete without a \ + retransmission, took {connected_after:?}" + ); + } + } +} diff --git a/tests/dtls12/retransmit.rs b/tests/dtls12/retransmit.rs index b1c9dde..bb79fda 100644 --- a/tests/dtls12/retransmit.rs +++ b/tests/dtls12/retransmit.rs @@ -1030,6 +1030,111 @@ fn dtls12_retransmit_exponential_backoff() { } } +/// RFC 6347 §4.2.4: retransmission is driven by the timer. Duplicates of the peer's +/// previous flight (network duplication, or a peer that resends on every duplicate it +/// sees) trigger at most one early resend per timer period, never one per datagram. +/// Otherwise two waiting peers bounce their flights back and forth at network speed. +#[test] +#[cfg(feature = "rcgen")] +fn dtls12_duplicate_flights_resend_at_most_once_per_timer_period() { + use dimpl::certificate::generate_self_signed_certificate; + + let _ = env_logger::try_init(); + + let client_cert = generate_self_signed_certificate().expect("gen client cert"); + let server_cert = generate_self_signed_certificate().expect("gen server cert"); + + let config = Arc::new( + Config::builder() + .use_server_cookie(false) + .build() + .expect("Failed to build config"), + ); + + let mut now = Instant::now(); + + let mut client = Dtls::new_12(Arc::clone(&config), client_cert, now); + client.set_active(true); + + let mut server = Dtls::new_12(config, server_cert, now); + server.set_active(false); + + client.handle_timeout(now).expect("client timeout start"); + client.handle_timeout(now).expect("client arm flight 1"); + server.handle_timeout(now).expect("server timeout start"); + let f1 = collect_packets(&mut client); + deliver_packets(&f1, &mut server); + + server.handle_timeout(now).expect("server arm flight 4"); + let f4 = collect_packets(&mut server); + assert!(!f4.is_empty(), "server should emit flight 4 after CH"); + + // The server waits for flight 5. 20 duplicate ClientHellos, each polled for output as + // a socket loop would, resend flight 4 once, not once per duplicate. + let sent = send_duplicates(&f1, &mut server); + assert_eq!( + sent, + f4.len(), + "20 duplicates should resend flight 4 ({} datagrams) once, server sent {sent}", + f4.len() + ); + + // More duplicates in the same timer period resend nothing. + let sent = send_duplicates(&f1, &mut server); + assert_eq!( + sent, 0, + "no second early resend within one timer period, server sent {sent}" + ); + + // The timer fires (a resend of its own) and starts the next period, which again + // allows one early resend. + now += Duration::from_secs(2); + server.handle_timeout(now).expect("server flight timeout"); + assert_eq!( + collect_packets(&mut server).len(), + f4.len(), + "the flight timer should resend flight 4" + ); + let sent = send_duplicates(&f1, &mut server); + assert_eq!( + sent, + f4.len(), + "one early resend in the next timer period, server sent {sent}" + ); + + // The same holds for the client waiting for flight 6: duplicate ServerHelloDones + // resend flight 5 once per timer period. + deliver_packets(&f4, &mut client); + client.handle_timeout(now).expect("client arm flight 5"); + let f5 = collect_packets(&mut client); + assert!(!f5.is_empty(), "client should emit flight 5"); + + let sent = send_duplicates(&f4, &mut client); + assert_eq!( + sent, + f5.len(), + "20 duplicates should resend flight 5 ({} datagrams) once, client sent {sent}", + f5.len() + ); + let sent = send_duplicates(&f4, &mut client); + assert_eq!( + sent, 0, + "no second early resend of flight 5 within one timer period, client sent {sent}" + ); +} + +/// Deliver `flight` 20 times, polling the endpoint after each copy. Returns the number of +/// datagrams it sent in response. +#[cfg(feature = "rcgen")] +fn send_duplicates(flight: &[Vec], endpoint: &mut Dtls) -> usize { + let mut sent = 0; + for _ in 0..20 { + deliver_packets(flight, endpoint); + sent += collect_packets(endpoint).len(); + } + sent +} + #[test] #[cfg(feature = "rcgen")] fn dtls12_handshake_timeout_aborts() {