streams: add a second check in front of reset flood mitigations

src/client.rs

@@ -138,7 +138,7 @@
 use crate::codec::{Codec, SendError, UserError};
 use crate::ext::Protocol;
 use crate::frame::{Headers, Pseudo, Reason, Settings, StreamId};
-use crate::proto::{self, Error};
+use crate::proto::{self, Error, DEFAULT_RESET_FLOOD_PENDING_FRAMES_MIN};
 use crate::{FlowControl, PingPong, RecvStream, SendStream};
 
 use bytes::{Buf, Bytes};
@@ -342,8 +342,11 @@ pub struct Builder {
     /// Maximum number of locally reset streams due to protocol error across
     /// the lifetime of the connection.
     ///
-    /// When this gets exceeded, we issue GOAWAYs.
+    /// When this gets exceeded, we issue GOAWAYs if we have more than
+    /// `reset_flood_pending_frames_min` queued.
     local_max_error_reset_streams: Option<usize>,
+
+    reset_flood_pending_frames_min: usize,
 }
 
 #[derive(Debug)]
@@ -654,6 +657,7 @@ impl Builder {
             settings: Default::default(),
             stream_id: 1.into(),
             local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX),
+            reset_flood_pending_frames_min: DEFAULT_RESET_FLOOD_PENDING_FRAMES_MIN,
         }
     }
 
@@ -993,7 +997,9 @@ impl Builder {
     /// Too many of these often indicate a malicious client, and there are attacks which can abuse this to DOS servers.
     /// This limit protects against these DOS attacks by limiting the amount of resets we can be forced to generate.
     ///
-    /// When the number of local resets exceeds this threshold, the client will close the connection.
+    /// When the number of local resets exceeds this threshold, the client will close the connection, but only if
+    /// we have more than `reset_flood_pending_frames_min` frames currently queued for transmission through this
+    /// connection.
     ///
     /// If you really want to disable this, supply [`Option::None`] here.
     /// Disabling this is not recommended and may expose you to DOS attacks.
@@ -1004,6 +1010,14 @@ impl Builder {
         self
     }
 
+    /// Sets the minimum number of pending frames before we mitigate a reset flood.
+    ///
+    /// See [`Self::max_local_error_reset_streams`] for more information.
+    pub fn min_reset_flood_pending_frames(&mut self, min: usize) -> &mut Self {
+        self.reset_flood_pending_frames_min = min;
+        self
+    }
+
     /// Sets the maximum number of pending-accept remotely-reset streams.
     ///
     /// Streams that have been received by the peer, but not accepted by the
@@ -1325,6 +1339,7 @@ where
                 reset_stream_max: builder.reset_stream_max,
                 remote_reset_stream_max: builder.pending_accept_reset_stream_max,
                 local_error_reset_streams_max: builder.local_max_error_reset_streams,
+                reset_flood_pending_frames_min: builder.reset_flood_pending_frames_min,
                 settings: builder.settings.clone(),
             },
         );

src/proto/connection.rs

@@ -82,6 +82,7 @@ pub(crate) struct Config {
     pub reset_stream_max: usize,
     pub remote_reset_stream_max: usize,
     pub local_error_reset_streams_max: Option<usize>,
+    pub reset_flood_pending_frames_min: usize,
     pub settings: frame::Settings,
 }
 
@@ -127,6 +128,7 @@ where
                     .max_concurrent_streams()
                     .map(|max| max as usize),
                 local_max_error_reset_streams: config.local_error_reset_streams_max,
+                reset_flood_pending_frames_min: config.reset_flood_pending_frames_min,
             }
         }
         let streams = Streams::new(streams_config(&config));

src/proto/mod.rs

@@ -32,7 +32,9 @@ pub type WindowSize = u32;
 // Constants
 pub const MAX_WINDOW_SIZE: WindowSize = (1 << 31) - 1; // i32::MAX as u32
 pub const DEFAULT_REMOTE_RESET_STREAM_MAX: usize = 20;
-pub const DEFAULT_LOCAL_RESET_COUNT_MAX: usize = 1024;
+pub const DEFAULT_LOCAL_RESET_COUNT_MAX: usize = 2048;
+
+pub const DEFAULT_RESET_FLOOD_PENDING_FRAMES_MIN: usize = DEFAULT_LOCAL_RESET_COUNT_MAX / 4;
 pub const DEFAULT_RESET_STREAM_MAX: usize = 10;
 pub const DEFAULT_RESET_STREAM_SECS: u64 = 30;
 pub const DEFAULT_MAX_SEND_BUFFER_SIZE: usize = 1024 * 400;

src/proto/streams/buffer.rs

@@ -29,6 +29,10 @@ impl<T> Buffer<T> {
     pub fn new() -> Self {
         Buffer { slab: Slab::new() }
     }
+
+    pub fn len(&self) -> usize {
+        self.slab.len()
+    }
 }
 
 impl Deque {

src/proto/streams/counts.rs

@@ -41,6 +41,9 @@ pub(super) struct Counts {
     /// Total number of locally reset streams due to protocol error across the
     /// lifetime of the connection.
     num_local_error_reset_streams: usize,
+
+    /// Minimum number of frames after which we will mitigate for Reset Flood
+    min_reset_flood_pending_frames: usize,
 }
 
 impl Counts {
@@ -54,6 +57,7 @@ impl Counts {
             num_recv_streams: 0,
             max_local_reset_streams: config.local_reset_max,
             num_local_reset_streams: 0,
+            min_reset_flood_pending_frames: config.reset_flood_pending_frames_min,
             max_remote_reset_streams: config.remote_reset_max,
             num_remote_reset_streams: 0,
             max_local_error_reset_streams: config.local_max_error_reset_streams,
@@ -98,6 +102,10 @@ impl Counts {
         self.max_local_error_reset_streams
     }
 
+    pub fn min_reset_flood_pending_frames(&self) -> usize {
+        self.min_reset_flood_pending_frames
+    }
+
     /// Returns true if the receive stream concurrency can be incremented
     pub fn can_inc_num_recv_streams(&self) -> bool {
         self.max_recv_streams > self.num_recv_streams

src/proto/streams/mod.rs

@@ -73,6 +73,11 @@ pub struct Config {
     /// Maximum number of locally reset streams due to protocol error across
     /// the lifetime of the connection.
     ///
-    /// When this gets exceeded, we issue GOAWAYs.
+    /// When this gets exceeded, we issue GOAWAYs if the
+    /// reset_flood_pending_frames_min check also fails.
     pub local_max_error_reset_streams: Option<usize>,
+
+    /// Minimum number of pending frames after which we will start mitigating
+    /// reset flood attacks.
+    pub reset_flood_pending_frames_min: usize,
 }

src/proto/streams/streams.rs

@@ -1547,7 +1547,14 @@ impl Actions {
         if let Err(Error::Reset(stream_id, reason, initiator)) = res {
             debug_assert_eq!(stream_id, stream.id);
 
-            if counts.can_inc_num_local_error_resets() {
+            // only mitigate a reset flood attack if the cumulative number of local error resets is past our strikeout
+            // threshold AND we have a lot of frames queued up to transmit
+            //
+            // some clients like to send bad stuff at a low but consistent rate, so we need to also gate on frames
+            // queueing up internally so that we only mitigate if they are actually causing us to buffer excessively
+            if counts.can_inc_num_local_error_resets()
+                || buffer.len() < counts.min_reset_flood_pending_frames()
+            {
                 counts.inc_num_local_error_resets();
 
                 // Reset the stream.

src/server.rs

@@ -258,6 +258,10 @@ pub struct Builder {
     ///
     /// When this gets exceeded, we issue GOAWAYs.
     local_max_error_reset_streams: Option<usize>,
+
+    /// Minimum number of pending frames after which we will start mitigating
+    /// reset flood attacks.
+    reset_flood_pending_frames_min: usize,
 }
 
 /// Send a response back to the client
@@ -652,6 +656,7 @@ impl Builder {
             max_send_buffer_size: proto::DEFAULT_MAX_SEND_BUFFER_SIZE,
 
             local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX),
+            reset_flood_pending_frames_min: proto::DEFAULT_RESET_FLOOD_PENDING_FRAMES_MIN,
         }
     }
 
@@ -896,7 +901,8 @@ impl Builder {
     /// This limit protects against these DOS attacks by limiting the amount of resets we can be forced to generate.
     ///
     /// When the number of local resets exceeds this threshold, the server will issue GOAWAYs with an error code of
-    /// `ENHANCE_YOUR_CALM` to the client.
+    /// `ENHANCE_YOUR_CALM` to the client, but only if there are more than min_reset_flood_pending_frames queued for
+    /// transmission.
     ///
     /// If you really want to disable this, supply [`Option::None`] here.
     /// Disabling this is not recommended and may expose you to DOS attacks.
@@ -907,6 +913,14 @@ impl Builder {
         self
     }
 
+    /// Sets the minimum number of pending frames before we mitigate a reset flood.
+    ///
+    /// See [`Self::max_local_error_reset_streams`] for more information.
+    pub fn min_reset_flood_pending_frames(&mut self, min: usize) -> &mut Self {
+        self.reset_flood_pending_frames_min = min;
+        self
+    }
+
     /// Sets the maximum number of pending-accept remotely-reset streams.
     ///
     /// Streams that have been received by the peer, but not accepted by the
@@ -1384,6 +1398,9 @@ where
                             local_error_reset_streams_max: self
                                 .builder
                                 .local_max_error_reset_streams,
+                            reset_flood_pending_frames_min: self
+                                .builder
+                                .reset_flood_pending_frames_min,
                             settings: self.builder.settings.clone(),
                         },
                     );