Fix DTLS 1.2 stall and resend storm when a flight's records arrive out of order

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
 

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,82 +647,38 @@ 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(
         &mut self,
         wanted: MessageType,
         defragment_buffer: &mut Buf,
     ) -> Result<Option<Handshake>, 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<ArrayVec<(&Handshake, &[u8]), MAX_DEFRAGMENT_PACKETS>> {
+    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)