feat(server): add h1 idle_timeout

Author: finnbear

src/error.rs

@@ -64,6 +64,9 @@ pub(super) enum Kind {
     /// User took too long to send headers
     #[cfg(all(feature = "http1", feature = "server"))]
     HeaderTimeout,
+    /// User took too long to send another request
+    #[cfg(all(feature = "http1", feature = "server"))]
+    IdleTimeout,
     /// Error while reading a body from connection.
     #[cfg(all(
         any(feature = "client", feature = "server"),
@@ -360,6 +363,11 @@ impl Error {
         Error::new(Kind::HeaderTimeout)
     }
 
+    #[cfg(all(feature = "http1", feature = "server"))]
+    pub(super) fn new_idle_timeout() -> Error {
+        Error::new(Kind::IdleTimeout)
+    }
+
     #[cfg(feature = "http1")]
     #[cfg(feature = "server")]
     pub(super) fn new_user_unsupported_status_code() -> Error {
@@ -458,6 +466,8 @@ impl Error {
             Kind::Canceled => "operation was canceled",
             #[cfg(all(feature = "http1", feature = "server"))]
             Kind::HeaderTimeout => "read header from client timeout",
+            #[cfg(all(feature = "http1", feature = "server"))]
+            Kind::IdleTimeout => "idle client timeout",
             #[cfg(all(
                 any(feature = "client", feature = "server"),
                 any(feature = "http1", feature = "http2")

src/proto/h1/conn.rs

@@ -61,10 +61,14 @@ where
                 #[cfg(feature = "server")]
                 h1_header_read_timeout: None,
                 #[cfg(feature = "server")]
-                h1_header_read_timeout_fut: None,
-                #[cfg(feature = "server")]
                 h1_header_read_timeout_running: false,
                 #[cfg(feature = "server")]
+                h1_idle_timeout: None,
+                #[cfg(feature = "server")]
+                h1_idle_timeout_running: false,
+                #[cfg(feature = "server")]
+                h1_timeout_fut: None,
+                #[cfg(feature = "server")]
                 date_header: true,
                 #[cfg(feature = "server")]
                 timer: Time::Empty,
@@ -147,6 +151,11 @@ where
         self.state.h1_header_read_timeout = Some(val);
     }
 
+    #[cfg(feature = "server")]
+    pub(crate) fn set_http1_idle_timeout(&mut self, val: Duration) {
+        self.state.h1_idle_timeout = Some(val);
+    }
+
     #[cfg(feature = "server")]
     pub(crate) fn set_allow_half_close(&mut self) {
         self.state.allow_half_close = true;
@@ -217,24 +226,10 @@ where
         trace!("Conn::read_head");
 
         #[cfg(feature = "server")]
-        if !self.state.h1_header_read_timeout_running {
-            if let Some(h1_header_read_timeout) = self.state.h1_header_read_timeout {
-                let deadline = Instant::now() + h1_header_read_timeout;
-                self.state.h1_header_read_timeout_running = true;
-                match self.state.h1_header_read_timeout_fut {
-                    Some(ref mut h1_header_read_timeout_fut) => {
-                        trace!("resetting h1 header read timeout timer");
-                        self.state.timer.reset(h1_header_read_timeout_fut, deadline);
-                    }
-                    None => {
-                        trace!("setting h1 header read timeout timer");
-                        self.state.h1_header_read_timeout_fut =
-                            Some(self.state.timer.sleep_until(deadline));
-                    }
-                }
-            }
-        }
+        let first_head = !self.io.is_read_blocked();
 
+        #[cfg_attr(not(feature = "server"), allow(unused))]
+        let mut progress = false;
         let msg = match self.io.parse::<T>(
             cx,
             ParseContext {
@@ -249,20 +244,69 @@ where
                 #[cfg(feature = "ffi")]
                 on_informational: &mut self.state.on_informational,
             },
+            &mut progress,
         ) {
             Poll::Ready(Ok(msg)) => msg,
             Poll::Ready(Err(e)) => return self.on_read_head_error(e),
             Poll::Pending => {
+                // - Use the read timeout on the first head to avoid common DoS.
+                // - If made progress in reading header, must no longer be idle.
                 #[cfg(feature = "server")]
-                if self.state.h1_header_read_timeout_running {
-                    if let Some(ref mut h1_header_read_timeout_fut) =
-                        self.state.h1_header_read_timeout_fut
-                    {
-                        if Pin::new(h1_header_read_timeout_fut).poll(cx).is_ready() {
-                            self.state.h1_header_read_timeout_running = false;
-
-                            warn!("read header from client timeout");
-                            return Poll::Ready(Some(Err(crate::Error::new_header_timeout())));
+                if first_head || progress {
+                    if !self.state.h1_header_read_timeout_running {
+                        if let Some(h1_header_read_timeout) = self.state.h1_header_read_timeout {
+                            let deadline = Instant::now() + h1_header_read_timeout;
+                            self.state.h1_idle_timeout_running = false;
+                            self.state.h1_header_read_timeout_running = true;
+                            match self.state.h1_timeout_fut {
+                                Some(ref mut ht_timeout_fut) => {
+                                    trace!("resetting h1 timeout timer for header read");
+                                    self.state.timer.reset(ht_timeout_fut, deadline);
+                                }
+                                None => {
+                                    trace!("setting h1 timeout timer for header read");
+                                    self.state.h1_timeout_fut =
+                                        Some(self.state.timer.sleep_until(deadline));
+                                }
+                            }
+                        } else if std::mem::take(&mut self.state.h1_idle_timeout_running) {
+                            trace!("unsetting h1 timeout timer for idle");
+                            self.state.h1_timeout_fut = None;
+                        }
+                    }
+                } else if !self.state.h1_header_read_timeout_running
+                    && !self.state.h1_idle_timeout_running
+                {
+                    if let Some(h1_idle_timeout) = self.state.h1_idle_timeout {
+                        let deadline = Instant::now() + h1_idle_timeout;
+                        self.state.h1_idle_timeout_running = true;
+                        match self.state.h1_timeout_fut {
+                            Some(ref mut h1_timeout_fut) => {
+                                trace!("resetting h1 timeout timer for idle");
+                                self.state.timer.reset(h1_timeout_fut, deadline);
+                            }
+                            None => {
+                                trace!("setting h1 timeout timer for idle");
+                                self.state.h1_timeout_fut =
+                                    Some(self.state.timer.sleep_until(deadline));
+                            }
+                        }
+                    }
+                }
+
+                #[cfg(feature = "server")]
+                if self.state.h1_header_read_timeout_running || self.state.h1_idle_timeout_running {
+                    if let Some(ref mut h1_timeout_fut) = self.state.h1_timeout_fut {
+                        if Pin::new(h1_timeout_fut).poll(cx).is_ready() {
+                            return Poll::Ready(Some(Err(
+                                if self.state.h1_header_read_timeout_running {
+                                    warn!("read header from client timeout");
+                                    crate::Error::new_header_timeout()
+                                } else {
+                                    warn!("idle client timeout");
+                                    crate::Error::new_idle_timeout()
+                                },
+                            )));
                         }
                     }
                 }
@@ -274,7 +318,8 @@ where
         #[cfg(feature = "server")]
         {
             self.state.h1_header_read_timeout_running = false;
-            self.state.h1_header_read_timeout_fut = None;
+            self.state.h1_idle_timeout_running = false;
+            self.state.h1_timeout_fut = None;
         }
 
         // Note: don't deconstruct `msg` into local variables, it appears
@@ -919,10 +964,14 @@ struct State {
     #[cfg(feature = "server")]
     h1_header_read_timeout: Option<Duration>,
     #[cfg(feature = "server")]
-    h1_header_read_timeout_fut: Option<Pin<Box<dyn Sleep>>>,
-    #[cfg(feature = "server")]
     h1_header_read_timeout_running: bool,
     #[cfg(feature = "server")]
+    h1_idle_timeout: Option<Duration>,
+    #[cfg(feature = "server")]
+    h1_idle_timeout_running: bool,
+    #[cfg(feature = "server")]
+    h1_timeout_fut: Option<Pin<Box<dyn Sleep>>>,
+    #[cfg(feature = "server")]
     date_header: bool,
     #[cfg(feature = "server")]
     timer: Time,
@@ -1106,6 +1155,12 @@ impl State {
         self.reading = Reading::Init;
         self.writing = Writing::Init;
 
+        #[cfg(feature = "server")]
+        if self.h1_idle_timeout.is_some() {
+            // Next read will start and poll the idle timeout.
+            self.notify_read = true;
+        }
+
         // !T::should_read_first() means Client.
         //
         // If Client connection has just gone idle, the Dispatcher

src/proto/h1/io.rs

@@ -169,6 +169,7 @@ where
         &mut self,
         cx: &mut Context<'_>,
         parse_ctx: ParseContext<'_>,
+        progress: &mut bool,
     ) -> Poll<crate::Result<ParsedMessage<S::Incoming>>>
     where
         S: Http1Transaction,
@@ -205,6 +206,7 @@ where
                 trace!("parse eof");
                 return Poll::Ready(Err(crate::Error::new_incomplete()));
             }
+            *progress = true;
         }
     }
 
@@ -702,7 +704,7 @@ mod tests {
                 on_informational: &mut None,
             };
             assert!(buffered
-                .parse::<ClientTransaction>(cx, parse_ctx)
+                .parse::<ClientTransaction>(cx, parse_ctx, &mut false)
                 .is_pending());
             Poll::Ready(())
         })

src/server/conn/http1.rs

@@ -76,6 +76,7 @@ pub struct Builder {
     h1_preserve_header_case: bool,
     h1_max_headers: Option<usize>,
     h1_header_read_timeout: Dur,
+    h1_idle_timeout: Dur,
     h1_writev: Option<bool>,
     max_buf_size: Option<usize>,
     pipeline_flush: bool,
@@ -238,6 +239,7 @@ impl Builder {
             h1_preserve_header_case: false,
             h1_max_headers: None,
             h1_header_read_timeout: Dur::Default(Some(Duration::from_secs(30))),
+            h1_idle_timeout: Dur::Default(Some(Duration::from_secs(120))),
             h1_writev: None,
             max_buf_size: None,
             pipeline_flush: false,
@@ -322,6 +324,21 @@ impl Builder {
         self
     }
 
+    /// Set a timeout for idle time between requests. If a client does not
+    /// transmit another request within this time after receiving the last
+    /// response, the connection is closed.
+    ///
+    /// Requires a [`Timer`] set by [`Builder::timer`] to take effect. Panics if `idle_timeout` is configured
+    /// without a [`Timer`].
+    ///
+    /// Pass `None` to disable.
+    ///
+    /// Default is 120 seconds.
+    pub fn idle_timeout(&mut self, idle_timeout: impl Into<Option<Duration>>) -> &mut Self {
+        self.h1_idle_timeout = Dur::Configured(idle_timeout.into());
+        self
+    }
+
     /// Set whether HTTP/1 connections should try to use vectored writes,
     /// or always flatten into a single buffer.
     ///
@@ -448,6 +465,9 @@ impl Builder {
         {
             conn.set_http1_header_read_timeout(dur);
         };
+        if let Some(dur) = self.timer.check(self.h1_idle_timeout, "idle_timeout") {
+            conn.set_http1_idle_timeout(dur);
+        };
         if let Some(writev) = self.h1_writev {
             if writev {
                 conn.set_write_strategy_queue();