Add request headers read timeout

Cargo.toml

@@ -43,7 +43,7 @@ log = "0.4"
 mimalloc = { version = "0.1.43", default-features = false, features = ["local_dynamic_tls"], optional = true }
 pem = "=3.0"
 percent-encoding = "=2.3"
-pin-project = "1.1"
+pin-project-lite = "=0.2"
 pkcs8 = { version = "=0.10", features = ["encryption", "pkcs5"] }
 pyo3 = { version = "=0.23", features = ["anyhow", "extension-module", "generate-import-lib"] }
 pyo3-log = "=0.12"

README.md

@@ -141,10 +141,15 @@ Options:
                                   GRANIAN_BACKPRESSURE; default:
                                   (backlog/workers); x>=1]
   --http1-buffer-size INTEGER RANGE
-                                  Set the maximum buffer size for HTTP/1
+                                  Sets the maximum buffer size for HTTP/1
                                   connections  [env var:
                                   GRANIAN_HTTP1_BUFFER_SIZE; default: 417792;
                                   x>=8192]
+  --http1-header-read-timeout INTEGER RANGE
+                                  Sets a timeout (in milliseconds) to read
+                                  headers  [env var:
+                                  GRANIAN_HTTP1_HEADER_READ_TIMEOUT; default:
+                                  30000; 1<=x<=60000]
   --http1-keep-alive / --no-http1-keep-alive
                                   Enables or disables HTTP/1 keep-alive  [env
                                   var: GRANIAN_HTTP1_KEEP_ALIVE; default:
@@ -159,42 +164,46 @@ Options:
                                   for HTTP2  [env var:
                                   GRANIAN_HTTP2_ADAPTIVE_WINDOW; default:
                                   (disabled)]
-  --http2-initial-connection-window-size INTEGER
+  --http2-initial-connection-window-size INTEGER RANGE
                                   Sets the max connection-level flow control
                                   for HTTP2  [env var: GRANIAN_HTTP2_INITIAL_C
-                                  ONNECTION_WINDOW_SIZE; default: 1048576]
-  --http2-initial-stream-window-size INTEGER
+                                  ONNECTION_WINDOW_SIZE; default: 1048576;
+                                  x>=1024]
+  --http2-initial-stream-window-size INTEGER RANGE
                                   Sets the `SETTINGS_INITIAL_WINDOW_SIZE`
                                   option for HTTP2 stream-level flow control
                                   [env var:
                                   GRANIAN_HTTP2_INITIAL_STREAM_WINDOW_SIZE;
-                                  default: 1048576]
-  --http2-keep-alive-interval INTEGER
-                                  Sets an interval for HTTP2 Ping frames
-                                  should be sent to keep a connection alive
-                                  [env var: GRANIAN_HTTP2_KEEP_ALIVE_INTERVAL]
-  --http2-keep-alive-timeout INTEGER
-                                  Sets a timeout for receiving an
+                                  default: 1048576; x>=1024]
+  --http2-keep-alive-interval INTEGER RANGE
+                                  Sets an interval (in milliseconds) for HTTP2
+                                  Ping frames should be sent to keep a
+                                  connection alive  [env var:
+                                  GRANIAN_HTTP2_KEEP_ALIVE_INTERVAL;
+                                  1<=x<=60000]
+  --http2-keep-alive-timeout INTEGER RANGE
+                                  Sets a timeout (in seconds) for receiving an
                                   acknowledgement of the HTTP2 keep-alive ping
                                   [env var: GRANIAN_HTTP2_KEEP_ALIVE_TIMEOUT;
-                                  default: 20]
-  --http2-max-concurrent-streams INTEGER
+                                  default: 20; x>=1]
+  --http2-max-concurrent-streams INTEGER RANGE
                                   Sets the SETTINGS_MAX_CONCURRENT_STREAMS
                                   option for HTTP2 connections  [env var:
                                   GRANIAN_HTTP2_MAX_CONCURRENT_STREAMS;
-                                  default: 200]
-  --http2-max-frame-size INTEGER  Sets the maximum frame size to use for HTTP2
+                                  default: 200; x>=10]
+  --http2-max-frame-size INTEGER RANGE
+                                  Sets the maximum frame size to use for HTTP2
                                   [env var: GRANIAN_HTTP2_MAX_FRAME_SIZE;
-                                  default: 16384]
-  --http2-max-headers-size INTEGER
+                                  default: 16384; x>=1024]
+  --http2-max-headers-size INTEGER RANGE
                                   Sets the max size of received header frames
                                   [env var: GRANIAN_HTTP2_MAX_HEADERS_SIZE;
-                                  default: 16777216]
-  --http2-max-send-buffer-size INTEGER
+                                  default: 16777216; x>=1]
+  --http2-max-send-buffer-size INTEGER RANGE
                                   Set the maximum write buffer size for each
                                   HTTP/2 stream  [env var:
                                   GRANIAN_HTTP2_MAX_SEND_BUFFER_SIZE; default:
-                                  409600]
+                                  409600; x>=1024]
   --log / --no-log                Enable logging  [env var:
                                   GRANIAN_LOG_ENABLED; default: (enabled)]
   --log-level [critical|error|warning|warn|info|debug|notset]

granian/cli.py

@@ -110,7 +110,13 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
     '--http1-buffer-size',
     type=click.IntRange(8192),
     default=HTTP1Settings.max_buffer_size,
-    help='Set the maximum buffer size for HTTP/1 connections',
+    help='Sets the maximum buffer size for HTTP/1 connections',
+)
+@option(
+    '--http1-header-read-timeout',
+    type=click.IntRange(1, 60_000),
+    default=HTTP1Settings.header_read_timeout,
+    help='Sets a timeout (in milliseconds) to read headers',
 )
 @option(
     '--http1-keep-alive/--no-http1-keep-alive',
@@ -129,49 +135,49 @@ def option(*param_decls: str, cls: Optional[Type[click.Option]] = None, **attrs:
 )
 @option(
     '--http2-initial-connection-window-size',
-    type=int,
+    type=click.IntRange(1024),
     default=HTTP2Settings.initial_connection_window_size,
     help='Sets the max connection-level flow control for HTTP2',
 )
 @option(
     '--http2-initial-stream-window-size',
-    type=int,
+    type=click.IntRange(1024),
     default=HTTP2Settings.initial_stream_window_size,
     help='Sets the `SETTINGS_INITIAL_WINDOW_SIZE` option for HTTP2 stream-level flow control',
 )
 @option(
     '--http2-keep-alive-interval',
-    type=int,
+    type=click.IntRange(1, 60_000),
     default=HTTP2Settings.keep_alive_interval,
-    help='Sets an interval for HTTP2 Ping frames should be sent to keep a connection alive',
+    help='Sets an interval (in milliseconds) for HTTP2 Ping frames should be sent to keep a connection alive',
 )
 @option(
     '--http2-keep-alive-timeout',
-    type=int,
+    type=click.IntRange(1),
     default=HTTP2Settings.keep_alive_timeout,
-    help='Sets a timeout for receiving an acknowledgement of the HTTP2 keep-alive ping',
+    help='Sets a timeout (in seconds) for receiving an acknowledgement of the HTTP2 keep-alive ping',
 )
 @option(
     '--http2-max-concurrent-streams',
-    type=int,
+    type=click.IntRange(10),
     default=HTTP2Settings.max_concurrent_streams,
     help='Sets the SETTINGS_MAX_CONCURRENT_STREAMS option for HTTP2 connections',
 )
 @option(
     '--http2-max-frame-size',
-    type=int,
+    type=click.IntRange(1024),
     default=HTTP2Settings.max_frame_size,
     help='Sets the maximum frame size to use for HTTP2',
 )
 @option(
     '--http2-max-headers-size',
-    type=int,
+    type=click.IntRange(1),
     default=HTTP2Settings.max_headers_size,
     help='Sets the max size of received header frames',
 )
 @option(
     '--http2-max-send-buffer-size',
-    type=int,
+    type=click.IntRange(1024),
     default=HTTP2Settings.max_send_buffer_size,
     help='Set the maximum write buffer size for each HTTP/2 stream',
 )
@@ -284,6 +290,7 @@ def cli(
     backlog: int,
     backpressure: Optional[int],
     http1_buffer_size: int,
+    http1_header_read_timeout: int,
     http1_keep_alive: bool,
     http1_pipeline_flush: bool,
     http2_adaptive_window: bool,
@@ -344,7 +351,10 @@ def cli(
         backlog=backlog,
         backpressure=backpressure,
         http1_settings=HTTP1Settings(
-            keep_alive=http1_keep_alive, max_buffer_size=http1_buffer_size, pipeline_flush=http1_pipeline_flush
+            header_read_timeout=http1_header_read_timeout,
+            keep_alive=http1_keep_alive,
+            max_buffer_size=http1_buffer_size,
+            pipeline_flush=http1_pipeline_flush,
         ),
         http2_settings=HTTP2Settings(
             adaptive_window=http2_adaptive_window,

granian/http.py

@@ -4,6 +4,7 @@
 
 @dataclass
 class HTTP1Settings:
+    header_read_timeout: int = 30_000
     keep_alive: bool = True
     max_buffer_size: int = 8192 + 4096 * 100
     pipeline_flush: bool = False

src/conversion.rs

@@ -56,11 +56,16 @@ impl<'p> IntoPyObject<'p> for FutureResultToPy {
 pub(crate) fn worker_http1_config_from_py(py: Python, cfg: Option<PyObject>) -> PyResult<HTTP1Config> {
     let ret = match cfg {
         Some(cfg) => HTTP1Config {
+            header_read_timeout: cfg
+                .getattr(py, "header_read_timeout")?
+                .extract(py)
+                .map(core::time::Duration::from_millis)?,
             keep_alive: cfg.getattr(py, "keep_alive")?.extract(py)?,
             max_buffer_size: cfg.getattr(py, "max_buffer_size")?.extract(py)?,
             pipeline_flush: cfg.getattr(py, "pipeline_flush")?.extract(py)?,
         },
         None => HTTP1Config {
+            header_read_timeout: core::time::Duration::from_secs(30),
             keep_alive: true,
             max_buffer_size: 8192 + 4096 * 100,
             pipeline_flush: false,
@@ -75,10 +80,11 @@ pub(crate) fn worker_http2_config_from_py(py: Python, cfg: Option<PyObject>) ->
             adaptive_window: cfg.getattr(py, "adaptive_window")?.extract(py)?,
             initial_connection_window_size: cfg.getattr(py, "initial_connection_window_size")?.extract(py)?,
             initial_stream_window_size: cfg.getattr(py, "initial_stream_window_size")?.extract(py)?,
-            keep_alive_interval: match cfg.getattr(py, "keep_alive_interval")?.extract(py) {
-                Ok(v) => Some(core::time::Duration::from_secs(v)),
-                _ => None,
-            },
+            keep_alive_interval: cfg
+                .getattr(py, "keep_alive_interval")?
+                .extract(py)
+                .ok()
+                .map(core::time::Duration::from_millis),
             keep_alive_timeout: core::time::Duration::from_secs(cfg.getattr(py, "keep_alive_timeout")?.extract(py)?),
             max_concurrent_streams: cfg.getattr(py, "max_concurrent_streams")?.extract(py)?,
             max_frame_size: cfg.getattr(py, "max_frame_size")?.extract(py)?,

src/io.rs

@@ -1,10 +1,14 @@
 use std::{
+    future::Future,
     marker::{PhantomData, Unpin},
     pin::Pin,
     task::{Context, Poll},
+    time::{Duration, Instant},
 };
 
+use hyper::rt::{Sleep, Timer};
 use hyper_util::rt::TokioIo;
+use pin_project_lite::pin_project;
 
 pub(crate) struct IOTypeNotSend<T> {
     _marker: PhantomData<*const ()>,
@@ -49,3 +53,57 @@ where
         Pin::new(&mut self.stream).poll_read(cx, buf)
     }
 }
+
+#[derive(Clone, Debug)]
+pub struct TokioTimer;
+
+impl Timer for TokioTimer {
+    fn sleep(&self, duration: Duration) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep(duration),
+        })
+    }
+
+    fn sleep_until(&self, deadline: Instant) -> Pin<Box<dyn Sleep>> {
+        Box::pin(TokioSleep {
+            inner: tokio::time::sleep_until(deadline.into()),
+        })
+    }
+
+    fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
+        if let Some(sleep) = sleep.as_mut().downcast_mut_pin::<TokioSleep>() {
+            sleep.reset(new_deadline);
+        }
+    }
+}
+
+impl TokioTimer {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+// Use TokioSleep to get tokio::time::Sleep to implement Unpin.
+// see https://docs.rs/tokio/latest/tokio/time/struct.Sleep.html
+pin_project! {
+    pub(crate) struct TokioSleep {
+        #[pin]
+        pub(crate) inner: tokio::time::Sleep,
+    }
+}
+
+impl Future for TokioSleep {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        self.project().inner.poll(cx)
+    }
+}
+
+impl Sleep for TokioSleep {}
+
+impl TokioSleep {
+    pub fn reset(self: Pin<&mut Self>, deadline: Instant) {
+        self.project().inner.as_mut().reset(deadline.into());
+    }
+}