Unbounded queuing of path validation messages in cloudflare-quiche

@@ -457,6 +457,9 @@ const MAX_SEND_UDP_PAYLOAD_SIZE: usize = 1200;
 // The default length of DATAGRAM queues.
 const DEFAULT_MAX_DGRAM_QUEUE_LEN: usize = 0;
 
+// The default length of PATH_CHALLENGE receive queue.
+const DEFAULT_MAX_PATH_CHALLENGE_RX_QUEUE_LEN: usize = 3;
+
 // The DATAGRAM standard recommends either none or 65536 as maximum DATAGRAM
 // frames size. We enforce the recommendation for forward compatibility.
 const MAX_DGRAM_FRAME_SIZE: u64 = 65536;
@@ -718,6 +721,8 @@ pub struct Config {
     dgram_recv_max_queue_len: usize,
     dgram_send_max_queue_len: usize,
 
+    path_challenge_recv_max_queue_len: usize,
+
     max_send_udp_payload_size: usize,
 
     max_connection_window: u64,
@@ -780,6 +785,9 @@ impl Config {
             dgram_recv_max_queue_len: DEFAULT_MAX_DGRAM_QUEUE_LEN,
             dgram_send_max_queue_len: DEFAULT_MAX_DGRAM_QUEUE_LEN,
 
+            path_challenge_recv_max_queue_len:
+                DEFAULT_MAX_PATH_CHALLENGE_RX_QUEUE_LEN,
+
             max_send_udp_payload_size: MAX_SEND_UDP_PAYLOAD_SIZE,
 
             max_connection_window: MAX_CONNECTION_WINDOW,
@@ -1194,6 +1202,16 @@ impl Config {
         self.dgram_send_max_queue_len = send_queue_len;
     }
 
+    /// Configures the max number of queued received PATH_CHALLENGE frames.
+    ///
+    /// When an endpoint receives a PATH_CHALLENGE frame and the queue is full,
+    /// the frame is discarded.
+    ///
+    /// The default is 3.
+    pub fn set_path_challenge_recv_max_queue_len(&mut self, queue_len: usize) {
+        self.path_challenge_recv_max_queue_len = queue_len;
+    }
+
     /// Sets the maximum size of the connection window.
     ///
     /// The default value is MAX_CONNECTION_WINDOW (24MBytes).
@@ -1268,6 +1286,12 @@ pub struct Connection {
     /// The path manager.
     paths: path::PathMap,
 
+    /// PATH_CHALLENGE receive queue max length.
+    path_challenge_recv_max_queue_len: usize,
+
+    /// Total number of received PATH_CHALLENGE frames.
+    path_challenge_rx_count: u64,
+
     /// List of supported application protocols.
     application_protos: Vec<Vec<u8>>,
 
@@ -1720,7 +1744,13 @@ impl Connection {
 
         let recovery_config = recovery::RecoveryConfig::from_config(config);
 
-        let mut path = path::Path::new(local, peer, &recovery_config, true);
+        let mut path = path::Path::new(
+            local,
+            peer,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            true,
+        );
         // If we did stateless retry assume the peer's address is verified.
         path.verified_peer_address = odcid.is_some();
         // Assume clients validate the server's address implicitly.
@@ -1766,6 +1796,9 @@ impl Connection {
             recovery_config,
 
             paths,
+            path_challenge_recv_max_queue_len: config
+                .path_challenge_recv_max_queue_len,
+            path_challenge_rx_count: 0,
 
             application_protos: config.application_protos.clone(),
 
@@ -6284,6 +6317,7 @@ impl Connection {
             stopped_stream_count_local: self.stopped_stream_local_count,
             reset_stream_count_remote: self.reset_stream_remote_count,
             stopped_stream_count_remote: self.stopped_stream_remote_count,
+            path_challenge_rx_count: self.path_challenge_rx_count,
         }
     }
 
@@ -7009,6 +7043,8 @@ impl Connection {
             },
 
             frame::Frame::PathChallenge { data } => {
+                self.path_challenge_rx_count += 1;
+
                 self.paths
                     .get_mut(recv_path_id)?
                     .on_challenge_received(data);
@@ -7301,8 +7337,13 @@ impl Connection {
         }
 
         // This is a new path using an unassigned CID; create it!
-        let mut path =
-            path::Path::new(info.to, info.from, &self.recovery_config, false);
+        let mut path = path::Path::new(
+            info.to,
+            info.from,
+            &self.recovery_config,
+            self.path_challenge_recv_max_queue_len,
+            false,
+        );
 
         path.max_send_bytes = buf_len * MAX_AMPLIFICATION_FACTOR;
         path.active_scid_seq = Some(in_scid_seq);
@@ -7424,8 +7465,13 @@ impl Connection {
                 .ok_or(Error::OutOfIdentifiers)?
         };
 
-        let mut path =
-            path::Path::new(local_addr, peer_addr, &self.recovery_config, false);
+        let mut path = path::Path::new(
+            local_addr,
+            peer_addr,
+            &self.recovery_config,
+            self.path_challenge_recv_max_queue_len,
+            false,
+        );
         path.active_dcid_seq = Some(dcid_seq);
 
         let pid = self
@@ -7532,6 +7578,9 @@ pub struct Stats {
 
     /// The number of streams stopped by remote.
     pub stopped_stream_count_remote: u64,
+
+    /// The total number of PATH_CHALLENGE frames that were received.
+    pub path_challenge_rx_count: u64,
 }
 
 impl std::fmt::Debug for Stats {
@@ -15300,6 +15349,9 @@ mod tests {
         };
         assert_eq!(pipe.server.recv(&mut buf[..sent], ri), Ok(sent));
 
+        let stats = pipe.server.stats();
+        assert_eq!(stats.path_challenge_rx_count, 1);
+
         // A non-existing 4-tuple raises an InvalidState.
         let client_addr_3 = "127.0.0.1:9012".parse().unwrap();
         let server_addr_2 = "127.0.0.1:9876".parse().unwrap();
@@ -15341,6 +15393,9 @@ mod tests {
         };
         assert_eq!(pipe.server.recv(&mut buf[..sent], ri), Ok(sent));
 
+        let stats = pipe.server.stats();
+        assert_eq!(stats.path_challenge_rx_count, 2);
+
         // STREAM frame on active path.
         let (sent, si) = pipe
             .client
@@ -15355,6 +15410,9 @@ mod tests {
         };
         assert_eq!(pipe.server.recv(&mut buf[..sent], ri), Ok(sent));
 
+        let stats = pipe.server.stats();
+        assert_eq!(stats.path_challenge_rx_count, 2);
+
         // PATH_CHALLENGE
         let (sent, si) = pipe
             .client
@@ -15370,6 +15428,9 @@ mod tests {
         };
         assert_eq!(pipe.server.recv(&mut buf[..sent], ri), Ok(sent));
 
+        let stats = pipe.server.stats();
+        assert_eq!(stats.path_challenge_rx_count, 3);
+
         // STREAM frame on active path.
         let (sent, si) = pipe
             .client
@@ -15419,6 +15480,9 @@ mod tests {
         v2.sort();
 
         assert_eq!(v1, v2);
+
+        let stats = pipe.server.stats();
+        assert_eq!(stats.path_challenge_rx_count, 3);
     }
 
     #[test]

@@ -151,6 +151,9 @@ pub struct Path {
     /// Received challenge data.
     received_challenges: VecDeque<[u8; 8]>,
 
+    /// Max length of received challenges queue.
+    received_challenges_max_len: usize,
+
     /// Number of packets sent on this path.
     pub sent_count: usize,
 
@@ -199,7 +202,8 @@ impl Path {
     /// the fields being set to their default value.
     pub fn new(
         local_addr: SocketAddr, peer_addr: SocketAddr,
-        recovery_config: &recovery::RecoveryConfig, is_initial: bool,
+        recovery_config: &recovery::RecoveryConfig,
+        path_challenge_recv_max_queue_len: usize, is_initial: bool,
     ) -> Self {
         let (state, active_scid_seq, active_dcid_seq) = if is_initial {
             (PathState::Validated, Some(0), Some(0))
@@ -219,7 +223,10 @@ impl Path {
             max_challenge_size: 0,
             probing_lost: 0,
             last_probe_lost_time: None,
-            received_challenges: VecDeque::new(),
+            received_challenges: VecDeque::with_capacity(
+                path_challenge_recv_max_queue_len,
+            ),
+            received_challenges_max_len: path_challenge_recv_max_queue_len,
             sent_count: 0,
             recv_count: 0,
             retrans_count: 0,
@@ -334,6 +341,13 @@ impl Path {
     }
 
     pub fn on_challenge_received(&mut self, data: [u8; 8]) {
+        // Discard challenges that would cause us to queue more than we want.
+        if self.received_challenges.len() ==
+            self.received_challenges_max_len
+        {
+            return;
+        }
+
         self.received_challenges.push_back(data);
         self.peer_verified_local_address = true;
     }
@@ -906,11 +920,22 @@ mod tests {
         let config = Config::new(crate::PROTOCOL_VERSION).unwrap();
         let recovery_config = RecoveryConfig::from_config(&config);
 
-        let path = Path::new(client_addr, server_addr, &recovery_config, true);
+        let path = Path::new(
+            client_addr,
+            server_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            true,
+        );
         let mut path_mgr = PathMap::new(path, 2, false);
 
-        let probed_path =
-            Path::new(client_addr_2, server_addr, &recovery_config, false);
+        let probed_path = Path::new(
+            client_addr_2,
+            server_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            false,
+        );
         path_mgr.insert_path(probed_path, false).unwrap();
 
         let pid = path_mgr
@@ -980,10 +1005,21 @@ mod tests {
         let config = Config::new(crate::PROTOCOL_VERSION).unwrap();
         let recovery_config = RecoveryConfig::from_config(&config);
 
-        let path = Path::new(client_addr, server_addr, &recovery_config, true);
+        let path = Path::new(
+            client_addr,
+            server_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            true,
+        );
         let mut client_path_mgr = PathMap::new(path, 2, false);
-        let mut server_path =
-            Path::new(server_addr, client_addr, &recovery_config, false);
+        let mut server_path = Path::new(
+            server_addr,
+            client_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            false,
+        );
 
         let client_pid = client_path_mgr
             .path_id_from_addrs(&(client_addr, server_addr))
@@ -1049,4 +1085,152 @@ mod tests {
             0
         );
     }
+
+    #[test]
+    fn too_many_probes() {
+        let client_addr = "127.0.0.1:1234".parse().unwrap();
+        let server_addr = "127.0.0.1:4321".parse().unwrap();
+
+        // Default to DEFAULT_MAX_PATH_CHALLENGE_RX_QUEUE_LEN
+        let config = Config::new(crate::PROTOCOL_VERSION).unwrap();
+        let recovery_config = RecoveryConfig::from_config(&config);
+
+        let path = Path::new(
+            client_addr,
+            server_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            true,
+        );
+        let mut client_path_mgr = PathMap::new(path, 2, false);
+        let mut server_path = Path::new(
+            server_addr,
+            client_addr,
+            &recovery_config,
+            config.path_challenge_recv_max_queue_len,
+            false,
+        );
+
+        let client_pid = client_path_mgr
+            .path_id_from_addrs(&(client_addr, server_addr))
+            .unwrap();
+
+        // First probe.
+        let data = rand::rand_u64().to_be_bytes();
+
+        client_path_mgr
+            .get_mut(client_pid)
+            .unwrap()
+            .add_challenge_sent(
+                data,
+                MIN_CLIENT_INITIAL_LEN,
+                time::Instant::now(),
+            );
+
+        // Second probe.
+        let data_2 = rand::rand_u64().to_be_bytes();
+
+        client_path_mgr
+            .get_mut(client_pid)
+            .unwrap()
+            .add_challenge_sent(
+                data_2,
+                MIN_CLIENT_INITIAL_LEN,
+                time::Instant::now(),
+            );
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            2
+        );
+
+        // Third probe.
+        let data_3 = rand::rand_u64().to_be_bytes();
+
+        client_path_mgr
+            .get_mut(client_pid)
+            .unwrap()
+            .add_challenge_sent(
+                data_3,
+                MIN_CLIENT_INITIAL_LEN,
+                time::Instant::now(),
+            );
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            3
+        );
+
+        // Fourth probe.
+        let data_4 = rand::rand_u64().to_be_bytes();
+
+        client_path_mgr
+            .get_mut(client_pid)
+            .unwrap()
+            .add_challenge_sent(
+                data_4,
+                MIN_CLIENT_INITIAL_LEN,
+                time::Instant::now(),
+            );
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            4
+        );
+
+        // If we receive multiple challenges, we can store them up to our queue size
+        server_path.on_challenge_received(data);
+        assert_eq!(server_path.received_challenges.len(), 1);
+        server_path.on_challenge_received(data_2);
+        assert_eq!(server_path.received_challenges.len(), 2);
+        server_path.on_challenge_received(data_3);
+        assert_eq!(server_path.received_challenges.len(), 3);
+        server_path.on_challenge_received(data_4);
+        assert_eq!(server_path.received_challenges.len(), 3);
+
+        // Response for first probe.
+        client_path_mgr.on_response_received(data).unwrap();
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            3
+        );
+
+        // Response for second probe.
+        client_path_mgr.on_response_received(data_2).unwrap();
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            2
+        );
+
+        // Response for third probe.
+        client_path_mgr.on_response_received(data_3).unwrap();
+        assert_eq!(
+            client_path_mgr
+                .get(client_pid)
+                .unwrap()
+                .in_flight_challenges
+                .len(),
+            1
+        );
+
+        // There will never be a response for fourth probe...
+
+    }
 }