Configuration Options for HTTP/1 Max Headers and Buffer Limits

Cargo.toml

@@ -76,3 +76,6 @@ sha1 = "0.10.6"
 tempfile = "3.10.1"
 tokio = { version = "1.36.0", features = ["full"] }
 tower = { version = "0.4.13", features = ["full"] }
+
+[patch.crates-io]
+"hyper" = { git = "https://github.com/IvanGoncharov/hyper.git", branch = "backpor_header_size" }

apollo-router/Cargo.toml

@@ -105,7 +105,7 @@ http-body = "0.4.6"
 heck = "0.5.0"
 humantime = "2.1.0"
 humantime-serde = "1.1.1"
-hyper = { version = "0.14.28", features = ["server", "client", "stream"] }
+hyper = { version = "0.14.31", features = ["server", "client", "stream"] }
 hyper-rustls = { version = "0.24.2", features = ["http1", "http2"] }
 indexmap = { version = "2.2.6", features = ["serde"] }
 itertools = "0.13.0"

apollo-router/src/axum_factory/axum_http_server_factory.rs

@@ -5,6 +5,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use std::time::Duration;
 use std::time::Instant;
 
 use axum::error_handling::HandleErrorLayer;
@@ -24,6 +25,7 @@ use http::header::CONTENT_ENCODING;
 use http::HeaderValue;
 use http::Request;
 use http_body::combinators::UnsyncBoxBody;
+use hyper::server::conn::Http;
 use hyper::Body;
 use itertools::Itertools;
 use multimap::MultiMap;
@@ -298,12 +300,22 @@ impl HttpServerFactory for AxumHttpServerFactory {
             let actual_main_listen_address = main_listener
                 .local_addr()
                 .map_err(ApolloRouterError::ServerCreationError)?;
+            let mut http_config = Http::new();
+            http_config.http1_keep_alive(true);
+            http_config.http1_header_read_timeout(Duration::from_secs(10));
+            if let Some(max_headers) = configuration.supergraph.experimental_http1_max_headers {
+                http_config.http1_max_headers(max_headers);
+            }
+            if let Some(max_buf_size) = configuration.supergraph.experimental_http1_max_buf_size {
+                http_config.max_buf_size(max_buf_size);
+            }
 
             let (main_server, main_shutdown_sender) = serve_router_on_listen_addr(
                 main_listener,
                 actual_main_listen_address.clone(),
                 all_routers.main.1,
                 true,
+                http_config.clone(),
                 all_connections_stopped_sender.clone(),
             );
 
@@ -343,6 +355,7 @@ impl HttpServerFactory for AxumHttpServerFactory {
                             listen_addr.clone(),
                             router,
                             false,
+                            http_config.clone(),
                             all_connections_stopped_sender.clone(),
                         );
                         (

apollo-router/src/axum_factory/listeners.rs

@@ -202,6 +202,7 @@ pub(super) fn serve_router_on_listen_addr(
     address: ListenAddr,
     router: axum::Router,
     main_graphql_port: bool,
+    http_config: Http,
     all_connections_stopped_sender: mpsc::Sender<()>,
 ) -> (impl Future<Output = Listener>, oneshot::Sender<()>) {
     let (shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();
@@ -243,6 +244,7 @@ pub(super) fn serve_router_on_listen_addr(
                             }
 
                             let address = address.clone();
+                            let mut http_config = http_config.clone();
                             tokio::task::spawn(async move {
                                 // this sender must be moved into the session to track that it is still running
                                 let _connection_stop_signal = connection_stop_signal;
@@ -261,11 +263,8 @@ pub(super) fn serve_router_on_listen_addr(
                                             .expect(
                                                 "this should not fail unless the socket is invalid",
                                             );
-                                            let connection = Http::new()
-                                            .http1_keep_alive(true)
-                                            .http1_header_read_timeout(Duration::from_secs(10))
-                                            .serve_connection(stream, app);
 
+                                        let connection = http_config.serve_connection(stream, app);
                                         tokio::pin!(connection);
                                         tokio::select! {
                                             // the connection finished first
@@ -291,9 +290,7 @@ pub(super) fn serve_router_on_listen_addr(
                                     NetworkStream::Unix(stream) => {
                                         let received_first_request = Arc::new(AtomicBool::new(false));
                                         let app = IdleConnectionChecker::new(received_first_request.clone(), app);
-                                        let connection = Http::new()
-                                        .http1_keep_alive(true)
-                                        .serve_connection(stream, app);
+                                        let connection = http_config.serve_connection(stream, app);
 
                                         tokio::pin!(connection);
                                         tokio::select! {
@@ -329,9 +326,7 @@ pub(super) fn serve_router_on_listen_addr(
                                             let protocol = stream.get_ref().1.alpn_protocol();
                                             let http2 = protocol == Some(&b"h2"[..]);
 
-                                            let connection = Http::new()
-                                            .http1_keep_alive(true)
-                                            .http1_header_read_timeout(Duration::from_secs(10))
+                                        let connection = http_config
                                             .http2_only(http2)
                                             .serve_connection(stream, app);
 

apollo-router/src/configuration/mod.rs

@@ -699,6 +699,19 @@ pub(crate) struct Supergraph {
     /// Log a message if the client closes the connection before the response is sent.
     /// Default: false.
     pub(crate) experimental_log_on_broken_pipe: bool,
+
+    /// Set the maximum number of headers for HTTP1 protocol
+    ///
+    /// If server receives more headers than the buffer size, it responds to the client with
+    /// "431 Request Header Fields Too Large".
+    ///
+    /// Default is 100.
+    pub(crate) experimental_http1_max_headers: Option<usize>,
+
+    /// Set the maximum buffer size for the HTTP1 connection.
+    ///
+    /// Default is ~400kb.
+    pub(crate) experimental_http1_max_buf_size: Option<usize>,
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
@@ -737,6 +750,8 @@ impl Supergraph {
         generate_query_fragments: Option<bool>,
         early_cancel: Option<bool>,
         experimental_log_on_broken_pipe: Option<bool>,
+        experimental_http1_max_headers: Option<usize>,
+        experimental_http1_max_buf_size: Option<usize>,
     ) -> Self {
         Self {
             listen: listen.unwrap_or_else(default_graphql_listen),
@@ -756,6 +771,8 @@ impl Supergraph {
             generate_query_fragments: generate_query_fragments.unwrap_or_default(),
             early_cancel: early_cancel.unwrap_or_default(),
             experimental_log_on_broken_pipe: experimental_log_on_broken_pipe.unwrap_or_default(),
+            experimental_http1_max_headers,
+            experimental_http1_max_buf_size,
         }
     }
 }
@@ -774,6 +791,8 @@ impl Supergraph {
         generate_query_fragments: Option<bool>,
         early_cancel: Option<bool>,
         experimental_log_on_broken_pipe: Option<bool>,
+        experimental_http1_max_headers: Option<usize>,
+        experimental_http1_max_buf_size: Option<usize>,
     ) -> Self {
         Self {
             listen: listen.unwrap_or_else(test_listen),
@@ -793,6 +812,8 @@ impl Supergraph {
             generate_query_fragments: generate_query_fragments.unwrap_or_default(),
             early_cancel: early_cancel.unwrap_or_default(),
             experimental_log_on_broken_pipe: experimental_log_on_broken_pipe.unwrap_or_default(),
+            experimental_http1_max_headers,
+            experimental_http1_max_buf_size,
         }
     }
 }

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -6592,6 +6592,14 @@ expression: "&schema"
           "description": "abort request handling when the client drops the connection. Default: false. When set to true, some parts of the request pipeline like telemetry will not work properly, but request handling will stop immediately when the client connection is closed.",
           "type": "boolean"
         },
+        "experimental_http1_max_headers": {
+          "default": null,
+          "description": "Set the maximum number of headers for HTTP1 protocol\n\nIf server receives more headers than the buffer size, it responds to the client with \"431 Request Header Fields Too Large\".\n\nDefault is 100.",
+          "format": "uint",
+          "minimum": 0.0,
+          "nullable": true,
+          "type": "integer"
+        },
         "experimental_log_on_broken_pipe": {
           "default": false,
           "description": "Log a message if the client closes the connection before the response is sent. Default: false.",

apollo-router/tests/integration/mod.rs

@@ -12,6 +12,7 @@ mod operation_limits;
 mod operation_name;
 mod query_planner;
 mod subgraph_response;
+mod supergraph;
 mod traffic_shaping;
 mod typename;
 

apollo-router/tests/integration/supergraph.rs

@@ -0,0 +1,119 @@
+use std::collections::HashMap;
+
+use serde_json::json;
+use tower::BoxError;
+
+use crate::integration::IntegrationTest;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_supergraph_errors_on_http1_max_headers() -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            supergraph:
+              experimental_http1_max_headers: 100
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    for i in 0..100 {
+        headers.insert(format!("test-header-{i}"), format!("value_{i}"));
+    }
+
+    let (_trace_id, response) = router
+        .execute_query_with_headers(&json!({ "query":  "{ __typename }"}), headers)
+        .await;
+    assert_eq!(response.status(), 431);
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_supergraph_allow_to_change_http1_max_headers() -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            supergraph:
+              experimental_http1_max_headers: 200
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    for i in 0..100 {
+        headers.insert(format!("test-header-{i}"), format!("value_{i}"));
+    }
+
+    let (_trace_id, response) = router
+        .execute_query_with_headers(&json!({ "query":  "{ __typename }"}), headers)
+        .await;
+    assert_eq!(response.status(), 200);
+    assert_eq!(
+        response.json::<serde_json::Value>().await?,
+        json!({ "data": { "__typename": "Query" } })
+    );
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_supergraph_allow_to_change_http1_max_buf_size() -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            supergraph:
+              experimental_http1_max_buf_size: 1000000
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    headers.insert(format!("test-header"), "x".repeat(1048576 + 1));
+
+    let (_trace_id, response) = router
+        .execute_query_with_headers(&json!({ "query":  "{ __typename }"}), headers)
+        .await;
+    assert_eq!(response.status(), 431);
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_supergraph_errors_on_http1_header_that_does_not_fit_inside_buffer(
+) -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            supergraph:
+              experimental_http1_max_buf_size: 2000000
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    headers.insert(format!("test-header"), "x".repeat(1048576 + 1));
+
+    let (_trace_id, response) = router
+        .execute_query_with_headers(&json!({ "query":  "{ __typename }"}), headers)
+        .await;
+    assert_eq!(response.status(), 200);
+    assert_eq!(
+        response.json::<serde_json::Value>().await?,
+        json!({ "data": { "__typename": "Query" } })
+    );
+    Ok(())
+}