streams: limit error resets for misbehaving connections

This change causes GOAWAYs to be issued to misbehaving connections which for one reason or another cause us to emit lots of error resets.

Error resets are not generally expected from valid implementations anyways.

The threshold after which we issue GOAWAYs is tunable, and will default to 1024.

src/client.rs

@@ -336,6 +336,12 @@ pub struct Builder {
     /// The stream ID of the first (lowest) stream. Subsequent streams will use
     /// monotonically increasing stream IDs.
     stream_id: StreamId,
+
+    /// Maximum number of locally reset streams due to protocol error across
+    /// the lifetime of the connection.
+    ///
+    /// When this gets exceeded, we issue GOAWAYs.
+    local_max_error_reset_streams: Option<usize>,
 }
 
 #[derive(Debug)]
@@ -645,6 +651,7 @@ impl Builder {
             initial_max_send_streams: usize::MAX,
             settings: Default::default(),
             stream_id: 1.into(),
+            local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX),
         }
     }
 
@@ -973,6 +980,23 @@ impl Builder {
         self
     }
 
+    /// Sets the maximum number of local resets due to protocol errors made by the remote end.
+    ///
+    /// Invalid frames and many other protocol errors will lead to resets being generated for those streams.
+    /// 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.
+    ///
+    /// If you really want to disable this, supply [`Option::None`] here.
+    /// Disabling this is not recommended and may expose you to DOS attacks.
+    ///
+    /// The default value is currently 1024, but could change.
+    pub fn max_local_error_reset_streams(&mut self, max: Option<usize>) -> &mut Self {
+        self.local_max_error_reset_streams = max;
+        self
+    }
+
     /// Sets the maximum number of pending-accept remotely-reset streams.
     ///
     /// Streams that have been received by the peer, but not accepted by the
@@ -1293,6 +1317,7 @@ where
                 reset_stream_duration: builder.reset_stream_duration,
                 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,
                 settings: builder.settings.clone(),
             },
         );

src/proto/connection.rs

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

src/proto/mod.rs

@@ -32,6 +32,7 @@ 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_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/counts.rs

@@ -31,6 +31,16 @@ pub(super) struct Counts {
 
     /// Current number of "pending accept" streams that were remotely reset
     num_remote_reset_streams: usize,
+
+    /// Maximum number of locally reset streams due to protocol error across
+    /// the lifetime of the connection.
+    ///
+    /// When this gets exceeded, we issue GOAWAYs.
+    max_local_error_reset_streams: Option<usize>,
+
+    /// Total number of locally reset streams due to protocol error across the
+    /// lifetime of the connection.
+    num_local_error_reset_streams: usize,
 }
 
 impl Counts {
@@ -46,6 +56,8 @@ impl Counts {
             num_local_reset_streams: 0,
             max_remote_reset_streams: config.remote_reset_max,
             num_remote_reset_streams: 0,
+            max_local_error_reset_streams: config.local_max_error_reset_streams,
+            num_local_error_reset_streams: 0,
         }
     }
 
@@ -66,6 +78,26 @@ impl Counts {
         self.num_send_streams != 0 || self.num_recv_streams != 0
     }
 
+    /// Returns true if we can issue another local reset due to protocol error.
+    pub fn can_inc_num_local_error_resets(&self) -> bool {
+        if let Some(max) = self.max_local_error_reset_streams {
+            max > self.num_local_error_reset_streams
+        } else {
+            true
+        }
+    }
+
+    pub fn inc_num_local_error_resets(&mut self) {
+        assert!(self.can_inc_num_local_error_resets());
+
+        // Increment the number of remote initiated streams
+        self.num_local_error_reset_streams += 1;
+    }
+
+    pub(crate) fn max_local_error_resets(&self) -> Option<usize> {
+        self.max_local_error_reset_streams
+    }
+
     /// 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

@@ -69,4 +69,10 @@ pub struct Config {
 
     /// Maximum number of remote initiated streams
     pub remote_max_initiated: Option<usize>,
+
+    /// Maximum number of locally reset streams due to protocol error across
+    /// the lifetime of the connection.
+    ///
+    /// When this gets exceeded, we issue GOAWAYs.
+    pub local_max_error_reset_streams: Option<usize>,
 }

src/proto/streams/streams.rs

@@ -1542,10 +1542,24 @@ impl Actions {
     ) -> Result<(), Error> {
         if let Err(Error::Reset(stream_id, reason, initiator)) = res {
             debug_assert_eq!(stream_id, stream.id);
-            // Reset the stream.
-            self.send
-                .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
-            Ok(())
+
+            if counts.can_inc_num_local_error_resets() {
+                counts.inc_num_local_error_resets();
+
+                // Reset the stream.
+                self.send
+                    .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
+                Ok(())
+            } else {
+                tracing::warn!(
+                    "reset_on_recv_stream_err; locally-reset streams reached limit ({:?})",
+                    counts.max_local_error_resets().unwrap(),
+                );
+                Err(Error::library_go_away_data(
+                    Reason::ENHANCE_YOUR_CALM,
+                    "too_many_internal_resets",
+                ))
+            }
         } else {
             res
         }

src/server.rs

@@ -252,6 +252,12 @@ pub struct Builder {
 
     /// Maximum amount of bytes to "buffer" for writing per stream.
     max_send_buffer_size: usize,
+
+    /// Maximum number of locally reset streams due to protocol error across
+    /// the lifetime of the connection.
+    ///
+    /// When this gets exceeded, we issue GOAWAYs.
+    local_max_error_reset_streams: Option<usize>,
 }
 
 /// Send a response back to the client
@@ -650,6 +656,8 @@ impl Builder {
             settings: Settings::default(),
             initial_target_connection_window_size: None,
             max_send_buffer_size: proto::DEFAULT_MAX_SEND_BUFFER_SIZE,
+
+            local_max_error_reset_streams: Some(proto::DEFAULT_LOCAL_RESET_COUNT_MAX),
         }
     }
 
@@ -887,6 +895,24 @@ impl Builder {
         self
     }
 
+    /// Sets the maximum number of local resets due to protocol errors made by the remote end.
+    ///
+    /// Invalid frames and many other protocol errors will lead to resets being generated for those streams.
+    /// 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 server will issue GOAWAYs with an error code of
+    /// `ENHANCE_YOUR_CALM` to the client.
+    ///
+    /// If you really want to disable this, supply [`Option::None`] here.
+    /// Disabling this is not recommended and may expose you to DOS attacks.
+    ///
+    /// The default value is currently 1024, but could change.
+    pub fn max_local_error_reset_streams(&mut self, max: Option<usize>) -> &mut Self {
+        self.local_max_error_reset_streams = max;
+        self
+    }
+
     /// Sets the maximum number of pending-accept remotely-reset streams.
     ///
     /// Streams that have been received by the peer, but not accepted by the
@@ -1361,6 +1387,9 @@ where
                             reset_stream_duration: self.builder.reset_stream_duration,
                             reset_stream_max: self.builder.reset_stream_max,
                             remote_reset_stream_max: self.builder.pending_accept_reset_stream_max,
+                            local_error_reset_streams_max: self
+                                .builder
+                                .local_max_error_reset_streams,
                             settings: self.builder.settings.clone(),
                         },
                     );