feat(h1): add server header read timeout

Cargo.toml

@@ -86,7 +86,7 @@ full = [
 ]
 
 # HTTP versions
-http1 = []
+http1 = ["tokio/time"]
 http2 = ["h2"]
 
 # Client/Server

src/error.rs

@@ -104,6 +104,9 @@ pub(super) enum User {
     #[cfg(any(feature = "http1", feature = "http2"))]
     #[cfg(feature = "server")]
     UnexpectedHeader,
+    /// User took too long to send headers
+    #[cfg(feature = "http1")]
+    HeaderTimeout,
     /// User tried to create a Request with bad version.
     #[cfg(any(feature = "http1", feature = "http2"))]
     #[cfg(feature = "client")]
@@ -310,6 +313,11 @@ impl Error {
         Error::new_user(User::UnexpectedHeader)
     }
 
+    #[cfg(feature = "http1")]
+    pub(super) fn new_header_timeout() -> Error {
+        Error::new_user(User::HeaderTimeout)
+    }
+
     #[cfg(any(feature = "http1", feature = "http2"))]
     #[cfg(feature = "client")]
     pub(super) fn new_user_unsupported_version() -> Error {
@@ -441,6 +449,8 @@ impl Error {
             #[cfg(any(feature = "http1", feature = "http2"))]
             #[cfg(feature = "server")]
             Kind::User(User::UnexpectedHeader) => "user sent unexpected header",
+            #[cfg(feature = "http1")]
+            Kind::User(User::HeaderTimeout) => "read header from client timeout",
             #[cfg(any(feature = "http1", feature = "http2"))]
             #[cfg(feature = "client")]
             Kind::User(User::UnsupportedVersion) => "request has unsupported HTTP version",

src/proto/h1/conn.rs

@@ -1,12 +1,14 @@
 use std::fmt;
 use std::io;
 use std::marker::PhantomData;
+use std::time::Duration;
 
 use bytes::{Buf, Bytes};
 use http::header::{HeaderValue, CONNECTION};
 use http::{HeaderMap, Method, Version};
 use httparse::ParserConfig;
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::time::Sleep;
 use tracing::{debug, error, trace};
 
 use super::io::Buffered;
@@ -47,6 +49,8 @@ where
                 keep_alive: KA::Busy,
                 method: None,
                 h1_parser_config: ParserConfig::default(),
+                h1_header_read_timeout: None,
+                h1_header_read_timeout_fut: None,
                 preserve_header_case: false,
                 title_case_headers: false,
                 h09_responses: false,
@@ -106,6 +110,11 @@ where
         self.state.h09_responses = true;
     }
 
+    #[cfg(feature = "server")]
+    pub(crate) fn set_http1_header_read_timeout(&mut self, val: Duration) {
+        self.state.h1_header_read_timeout = Some(val);
+    }
+
     #[cfg(feature = "server")]
     pub(crate) fn set_allow_half_close(&mut self) {
         self.state.allow_half_close = true;
@@ -178,6 +187,8 @@ where
                 cached_headers: &mut self.state.cached_headers,
                 req_method: &mut self.state.method,
                 h1_parser_config: self.state.h1_parser_config.clone(),
+                h1_header_read_timeout: self.state.h1_header_read_timeout,
+                h1_header_read_timeout_fut: &mut self.state.h1_header_read_timeout_fut,
                 preserve_header_case: self.state.preserve_header_case,
                 h09_responses: self.state.h09_responses,
                 #[cfg(feature = "ffi")]
@@ -798,6 +809,8 @@ struct State {
     /// a body or not.
     method: Option<Method>,
     h1_parser_config: ParserConfig,
+    h1_header_read_timeout: Option<Duration>,
+    h1_header_read_timeout_fut: Option<Pin<Box<Sleep>>>,
     preserve_header_case: bool,
     title_case_headers: bool,
     h09_responses: bool,

src/proto/h1/io.rs

@@ -3,10 +3,11 @@ use std::fmt;
 use std::io::{self, IoSlice};
 use std::marker::Unpin;
 use std::mem::MaybeUninit;
+use std::future::Future;
 
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
-use tracing::{debug, trace};
+use tracing::{debug, warn, trace};
 
 use super::{Http1Transaction, ParseContext, ParsedMessage};
 use crate::common::buf::BufList;
@@ -181,6 +182,8 @@ where
                     cached_headers: parse_ctx.cached_headers,
                     req_method: parse_ctx.req_method,
                     h1_parser_config: parse_ctx.h1_parser_config.clone(),
+                    h1_header_read_timeout: parse_ctx.h1_header_read_timeout,
+                    h1_header_read_timeout_fut: parse_ctx.h1_header_read_timeout_fut,
                     preserve_header_case: parse_ctx.preserve_header_case,
                     h09_responses: parse_ctx.h09_responses,
                     #[cfg(feature = "ffi")]
@@ -191,6 +194,8 @@ where
             )? {
                 Some(msg) => {
                     debug!("parsed {} headers", msg.head.headers.len());
+
+                    *parse_ctx.h1_header_read_timeout_fut = None;
                     return Poll::Ready(Ok(msg));
                 }
                 None => {
@@ -199,6 +204,15 @@ where
                         debug!("max_buf_size ({}) reached, closing", max);
                         return Poll::Ready(Err(crate::Error::new_too_large()));
                     }
+
+                    if let Some(h1_header_read_timeout_fut) = parse_ctx.h1_header_read_timeout_fut {
+                        if Pin::new( h1_header_read_timeout_fut).poll(cx).is_ready() {
+                            *parse_ctx.h1_header_read_timeout_fut = None;
+
+                            warn!("read header from client timeout");
+                            return Poll::Ready(Err(crate::Error::new_header_timeout()))
+                        }
+                    }
                 }
             }
             if ready!(self.poll_read_from_io(cx)).map_err(crate::Error::new_io)? == 0 {
@@ -693,6 +707,8 @@ mod tests {
                 cached_headers: &mut None,
                 req_method: &mut None,
                 h1_parser_config: Default::default(),
+                h1_header_read_timeout: None,
+                h1_header_read_timeout_fut: &mut None,
                 preserve_header_case: false,
                 h09_responses: false,
                 #[cfg(feature = "ffi")]

src/proto/h1/mod.rs

@@ -1,6 +1,10 @@
+use std::pin::Pin;
+use std::time::Duration;
+
 use bytes::BytesMut;
 use http::{HeaderMap, Method};
 use httparse::ParserConfig;
+use tokio::time::Sleep;
 
 use crate::body::DecodedLength;
 use crate::proto::{BodyLength, MessageHead};
@@ -72,6 +76,8 @@ pub(crate) struct ParseContext<'a> {
     cached_headers: &'a mut Option<HeaderMap>,
     req_method: &'a mut Option<Method>,
     h1_parser_config: ParserConfig,
+    h1_header_read_timeout: Option<Duration>,
+    h1_header_read_timeout_fut: &'a mut Option<Pin<Box<Sleep>>>,
     preserve_header_case: bool,
     h09_responses: bool,
     #[cfg(feature = "ffi")]

src/proto/h1/role.rs

@@ -69,6 +69,14 @@ where
 
     let span = trace_span!("parse_headers");
     let _s = span.enter();
+
+    if let Some(h1_header_read_timeout) = ctx.h1_header_read_timeout {
+        if ctx.h1_header_read_timeout_fut.is_none() {
+            debug!("setting h1 header read timeout timer");
+            *ctx.h1_header_read_timeout_fut = Some(Box::pin(tokio::time::sleep(h1_header_read_timeout)));
+        }
+    }
+
     T::parse(bytes, ctx)
 }
 
@@ -1428,6 +1436,8 @@ mod tests {
                 cached_headers: &mut None,
                 req_method: &mut method,
                 h1_parser_config: Default::default(),
+                h1_header_read_timeout: None,
+                h1_header_read_timeout_fut: &mut None,
                 preserve_header_case: false,
                 h09_responses: false,
                 #[cfg(feature = "ffi")]
@@ -1455,6 +1465,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut Some(crate::Method::GET),
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1477,6 +1489,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut None,
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1497,6 +1511,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut Some(crate::Method::GET),
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: true,
             #[cfg(feature = "ffi")]
@@ -1519,6 +1535,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut Some(crate::Method::GET),
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1545,6 +1563,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut Some(crate::Method::GET),
             h1_parser_config,
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1568,6 +1588,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut Some(crate::Method::GET),
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: false,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1586,6 +1608,8 @@ mod tests {
             cached_headers: &mut None,
             req_method: &mut None,
             h1_parser_config: Default::default(),
+            h1_header_read_timeout: None,
+            h1_header_read_timeout_fut: &mut None,
             preserve_header_case: true,
             h09_responses: false,
             #[cfg(feature = "ffi")]
@@ -1625,6 +1649,8 @@ mod tests {
                     cached_headers: &mut None,
                     req_method: &mut None,
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -1645,6 +1671,8 @@ mod tests {
                     cached_headers: &mut None,
                     req_method: &mut None,
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -1874,6 +1902,8 @@ mod tests {
                     cached_headers: &mut None,
                     req_method: &mut Some(Method::GET),
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -1894,6 +1924,8 @@ mod tests {
                     cached_headers: &mut None,
                     req_method: &mut Some(m),
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -1914,6 +1946,8 @@ mod tests {
                     cached_headers: &mut None,
                     req_method: &mut Some(Method::GET),
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -2411,6 +2445,8 @@ mod tests {
                 cached_headers: &mut None,
                 req_method: &mut Some(Method::GET),
                 h1_parser_config: Default::default(),
+                h1_header_read_timeout: None,
+                h1_header_read_timeout_fut: &mut None,
                 preserve_header_case: false,
                 h09_responses: false,
                 #[cfg(feature = "ffi")]
@@ -2495,6 +2531,8 @@ mod tests {
                     cached_headers: &mut headers,
                     req_method: &mut None,
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]
@@ -2535,6 +2573,8 @@ mod tests {
                     cached_headers: &mut headers,
                     req_method: &mut None,
                     h1_parser_config: Default::default(),
+                    h1_header_read_timeout: None,
+                    h1_header_read_timeout_fut: &mut None,
                     preserve_header_case: false,
                     h09_responses: false,
                     #[cfg(feature = "ffi")]

src/server/conn.rs

@@ -50,7 +50,6 @@
 use std::marker::PhantomData;
 #[cfg(feature = "tcp")]
 use std::net::SocketAddr;
-#[cfg(all(feature = "runtime", feature = "http2"))]
 use std::time::Duration;
 
 #[cfg(feature = "http2")]
@@ -103,6 +102,8 @@ pub struct Http<E = Exec> {
     h1_keep_alive: bool,
     h1_title_case_headers: bool,
     h1_preserve_header_case: bool,
+    #[cfg(feature = "http1")]
+    h1_header_read_timeout: Option<Duration>,
     h1_writev: Option<bool>,
     #[cfg(feature = "http2")]
     h2_builder: proto::h2::server::Config,
@@ -285,6 +286,8 @@ impl Http {
             h1_keep_alive: true,
             h1_title_case_headers: false,
             h1_preserve_header_case: false,
+            #[cfg(feature = "http1")]
+            h1_header_read_timeout: None,
             h1_writev: None,
             #[cfg(feature = "http2")]
             h2_builder: Default::default(),
@@ -365,6 +368,17 @@ impl<E> Http<E> {
         self
     }
 
+    /// Set a timeout for reading client request headers. If a client does not 
+    /// transmit the entire header withing this time, the connection is closed.
+    ///
+    /// Default is None.
+    #[cfg(feature = "http1")]
+    #[cfg_attr(docsrs, doc(cfg(feature = "http1")))]
+    pub fn http1_header_read_timeout(&mut self, read_timeout: Duration) -> &mut Self {
+        self.h1_header_read_timeout = Some(read_timeout);
+        self
+    }
+
     /// Set whether HTTP/1 connections should try to use vectored writes,
     /// or always flatten into a single buffer.
     ///
@@ -560,6 +574,8 @@ impl<E> Http<E> {
             h1_keep_alive: self.h1_keep_alive,
             h1_title_case_headers: self.h1_title_case_headers,
             h1_preserve_header_case: self.h1_preserve_header_case,
+            #[cfg(feature = "http1")]
+            h1_header_read_timeout: self.h1_header_read_timeout,
             h1_writev: self.h1_writev,
             #[cfg(feature = "http2")]
             h2_builder: self.h2_builder,
@@ -622,6 +638,9 @@ impl<E> Http<E> {
                 if self.h1_preserve_header_case {
                     conn.set_preserve_header_case();
                 }
+                if let Some(header_read_timeout) = self.h1_header_read_timeout {
+                    conn.set_http1_header_read_timeout(header_read_timeout);
+                }
                 if let Some(writev) = self.h1_writev {
                     if writev {
                         conn.set_write_strategy_queue();