dtls13: preserve overlapping KeyUpdate flights

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Unreleased
 
+  * Preserve overlapping DTLS 1.3 KeyUpdate flights #123
+  * Preserve DTLS 1.3 app data following KeyUpdate in one datagram #122
+  * Fix malformed datagrams consuming DTLS replay-window state #121
   * Replace pending DTLS 1.2 handshake output on resend #116
   * Discard bad protected DTLS 1.2 records after handshake #115
   * Reject oversized DTLS certificate lists #113

src/dtls12/incoming.rs

@@ -8,6 +8,7 @@ use crate::Error;
 use crate::buffer::{Buf, TmpBuf};
 use crate::crypto::{Aad, Nonce};
 use crate::dtls12::message::{ContentType, DTLSRecord, Dtls12CipherSuite, Handshake, Sequence};
+use crate::window::ReplayWindow;
 
 /// Holds both the UDP packet and the parsed result of that packet.
 pub struct Incoming {
@@ -67,12 +68,14 @@ pub struct Records {
 }
 
 impl Records {
-    pub fn parse(
+    fn parse(
         mut packet: &[u8],
         decrypt: &mut dyn RecordHandler,
         cs: Option<Dtls12CipherSuite>,
     ) -> Result<Records, Error> {
         let mut parsed_records: ArrayVec<Record, 8> = ArrayVec::new();
+        let mut replay_updates: ArrayVec<Sequence, 8> = ArrayVec::new();
+        let mut pending_replay = ReplayWindow::new();
 
         // Find record boundaries and copy each record ONCE from the packet
         while !packet.is_empty() {
@@ -91,21 +94,43 @@ impl Records {
             // This is the ONLY copy: packet -> record buffer
             let record_slice = &packet[..record_end];
             match Record::parse(record_slice, decrypt, cs) {
-                Ok(record) => {
-                    if let Some(record) = record {
+                Ok(parsed) => {
+                    if let Some(sequence) = parsed.replay_sequence {
+                        if !pending_replay.check(sequence.sequence_number) {
+                            trace!("Discarding duplicate rec in same datagram");
+                            packet = &packet[record_end..];
+                            continue;
+                        }
+                    }
+
+                    if let Some(record) = parsed.record {
                         if parsed_records.try_push(record).is_err() {
                             return Err(Error::TooManyRecords);
                         }
-                    } else {
+                    } else if parsed.replay_sequence.is_none() {
                         trace!("Discarding replayed rec");
                     }
+
+                    if let Some(sequence) = parsed.replay_sequence {
+                        pending_replay.update(sequence.sequence_number);
+                        if replay_updates.try_push(sequence).is_err() {
+                            return Err(Error::TooManyRecords);
+                        }
+                    }
                 }
                 Err(e) => return Err(e),
             }
 
             packet = &packet[record_end..];
         }
 
+        // Commit replay state only after the whole UDP datagram has parsed
+        // successfully. A malformed trailing record must not consume
+        // replay state for an earlier authenticated record in the same datagram.
+        for sequence in replay_updates {
+            decrypt.replay_update(sequence);
+        }
+
         let mut records = ArrayVec::new();
         for record in parsed_records {
             if let Some(record) = decrypt.classify_record(record)? {
@@ -134,14 +159,19 @@ pub struct Record {
     parsed: Box<ParsedRecord>,
 }
 
+struct RecordParse {
+    record: Option<Record>,
+    replay_sequence: Option<Sequence>,
+}
+
 impl Record {
     /// The first parse pass only parses the DTLSRecord header which is unencrypted.
     /// Copies record data from UDP packet ONCE into a pooled buffer.
-    pub fn parse(
+    fn parse(
         record_slice: &[u8],
         decrypt: &mut dyn RecordHandler,
         cs: Option<Dtls12CipherSuite>,
-    ) -> Result<Option<Record>, Error> {
+    ) -> Result<RecordParse, Error> {
         // ONLY COPY: UDP packet slice -> pooled buffer
         let mut buffer = Buf::new();
         buffer.extend_from_slice(record_slice);
@@ -151,7 +181,10 @@ impl Record {
                 // RFC 6347 §4.1.2.7: Invalid records SHOULD be silently discarded.
                 // This includes epoch 0 records with invalid ContentType.
                 trace!("Discarding record: parse failed: {}", e);
-                return Ok(None);
+                return Ok(RecordParse {
+                    record: None,
+                    replay_sequence: None,
+                });
             }
         };
         let parsed = Box::new(parsed);
@@ -162,7 +195,10 @@ impl Record {
         // packet loss, we can end up seeing epoch 1 records before we can decrypt them.
         let is_epoch_0 = record.record().sequence.epoch == 0;
         if is_epoch_0 || !decrypt.is_peer_encryption_enabled() {
-            return Ok(Some(record));
+            return Ok(RecordParse {
+                record: Some(record),
+                replay_sequence: None,
+            });
         }
 
         // We need to decrypt the record and redo the parsing.
@@ -171,12 +207,18 @@ impl Record {
 
         // Anti-replay check (read-only, does not update window)
         if !decrypt.replay_check(sequence) {
-            return Ok(None);
+            return Ok(RecordParse {
+                record: None,
+                replay_sequence: None,
+            });
         }
 
         let explicit_nonce_len = decrypt.explicit_nonce_len();
         if (dtls.length as usize) < decrypt.min_protected_fragment_len() {
-            return Ok(None);
+            return Ok(RecordParse {
+                record: None,
+                replay_sequence: None,
+            });
         }
 
         // Get a reference to the buffer
@@ -203,25 +245,35 @@ impl Record {
                 }
 
                 trace!("Discarding record: decrypt failed: {}", e);
-                return Ok(None);
+                return Ok(RecordParse {
+                    record: None,
+                    replay_sequence: None,
+                });
             }
 
             buffer.len()
         };
 
-        // Decryption succeeded — now commit the replay window update.
-        // RFC 6347 §4.1.2.6: "The receive window is updated only if the
-        // MAC verification succeeds."
-        decrypt.replay_update(sequence);
-
         // Update the length of the record.
         buffer[11] = (new_len >> 8) as u8;
         buffer[12] = new_len as u8;
 
-        let parsed = ParsedRecord::parse(&buffer, cs, explicit_nonce_len)?;
+        let parsed = match ParsedRecord::parse(&buffer, cs, explicit_nonce_len) {
+            Ok(parsed) => parsed,
+            Err(e) => {
+                trace!("Discarding authenticated record: parse failed: {}", e);
+                return Ok(RecordParse {
+                    record: None,
+                    replay_sequence: Some(sequence),
+                });
+            }
+        };
         let parsed = Box::new(parsed);
 
-        Ok(Some(Record { buffer, parsed }))
+        Ok(RecordParse {
+            record: Some(Record { buffer, parsed }),
+            replay_sequence: Some(sequence),
+        })
     }
 
     pub fn record(&self) -> &DTLSRecord {

src/dtls13/client.rs

@@ -220,6 +220,9 @@ impl Client {
     pub fn handle_packet(&mut self, packet: &[u8]) -> Result<(), Error> {
         self.engine.parse_packet(packet)?;
         self.make_progress()?;
+        while self.engine.parse_next_deferred_packet()? {
+            self.make_progress()?;
+        }
         Ok(())
     }
 
@@ -246,6 +249,56 @@ impl Client {
             .create_key_update(KeyUpdateRequest::UpdateRequested)
     }
 
+    fn send_pending_key_update_response(&mut self) -> Result<(), Error> {
+        if self.pending_key_update_response && !self.engine.is_key_update_in_flight() {
+            self.engine.send_ack()?;
+            self.engine
+                .create_key_update(KeyUpdateRequest::UpdateNotRequested)?;
+            self.pending_key_update_response = false;
+        }
+        Ok(())
+    }
+
+    fn handle_incoming_key_update(&mut self) -> Result<(), Error> {
+        if self.engine.has_complete_handshake(MessageType::KeyUpdate) {
+            let maybe = self.engine.next_handshake_no_transcript(
+                MessageType::KeyUpdate,
+                &mut self.defragment_buffer,
+            )?;
+
+            if let Some(handshake) = maybe {
+                let Body::KeyUpdate(request) = handshake.body else {
+                    unreachable!()
+                };
+
+                // Install new recv keys
+                self.engine.update_recv_keys()?;
+
+                // If peer requested us to update, schedule our own KeyUpdate
+                let local_key_update_in_flight = self.engine.is_key_update_in_flight();
+                if request == KeyUpdateRequest::UpdateRequested {
+                    self.pending_key_update_response = true;
+                    if local_key_update_in_flight {
+                        self.engine.send_ack_with_previous_app_epoch()?;
+                    } else {
+                        self.engine.send_ack()?;
+                    }
+                } else {
+                    self.engine.send_ack()?;
+                }
+
+                self.engine.advance_peer_handshake_seq();
+                debug!("Received KeyUpdate (request={:?})", request);
+
+                // Drain a fresh peer-requested response in the same progress
+                // pass when no local KeyUpdate is in flight.
+                self.send_pending_key_update_response()?;
+            }
+        }
+
+        Ok(())
+    }
+
     /// Send application data when the client is connected.
     pub fn send_application_data(&mut self, data: &[u8]) -> Result<(), Error> {
         if self.state == State::Closed || self.state == State::HalfClosedLocal {
@@ -1048,8 +1101,13 @@ impl State {
     }
 
     fn await_application_data(self, client: &mut Client) -> Result<Self, Error> {
+        // Incoming peer requests require an update_not_requested response. They
+        // take priority over local AEAD-limit updates and queued app data.
+        client.handle_incoming_key_update()?;
+        client.send_pending_key_update_response()?;
+
         // Auto-trigger KeyUpdate when AEAD encryption limit is reached
-        if client.engine.needs_key_update() && !client.engine.is_key_update_in_flight() {
+        if !client.engine.is_key_update_in_flight() && client.engine.needs_key_update() {
             client.initiate_key_update()?;
         }
 
@@ -1072,42 +1130,6 @@ impl State {
             }
         }
 
-        // Send pending KeyUpdate response before processing new KeyUpdates
-        if client.pending_key_update_response {
-            client
-                .engine
-                .create_key_update(KeyUpdateRequest::UpdateNotRequested)?;
-            client.pending_key_update_response = false;
-        }
-
-        // Check for incoming KeyUpdate
-        if client.engine.has_complete_handshake(MessageType::KeyUpdate) {
-            let maybe = client.engine.next_handshake_no_transcript(
-                MessageType::KeyUpdate,
-                &mut client.defragment_buffer,
-            )?;
-
-            if let Some(handshake) = maybe {
-                let Body::KeyUpdate(request) = handshake.body else {
-                    unreachable!()
-                };
-
-                // Install new recv keys
-                client.engine.update_recv_keys()?;
-
-                // ACK the KeyUpdate record
-                client.engine.send_ack()?;
-
-                // If peer requested us to update, schedule our own KeyUpdate
-                if request == KeyUpdateRequest::UpdateRequested {
-                    client.pending_key_update_response = true;
-                }
-
-                client.engine.advance_peer_handshake_seq();
-                debug!("Received KeyUpdate (request={:?})", request);
-            }
-        }
-
         Ok(self)
     }