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/apollographql/hyper.git", tag = "header-customizations-20241108" }

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.limits.experimental_http1_max_headers {
+                http_config.http1_max_headers(max_headers);
+            }
+            if let Some(max_buf_size) = configuration.limits.experimental_http1_max_buf_size {
+                http_config.max_buf_size(max_buf_size.as_u64() as usize);
+            }
 
             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/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -1,6 +1,7 @@
 ---
 source: apollo-router/src/configuration/tests.rs
 expression: "&schema"
+snapshot_kind: text
 ---
 {
   "$schema": "http://json-schema.org/draft-07/schema#",
@@ -1317,6 +1318,20 @@ expression: "&schema"
       "additionalProperties": false,
       "description": "Configuration for operation limits, parser limits, HTTP limits, etc.",
       "properties": {
+        "experimental_http1_max_buf_size": {
+          "default": null,
+          "description": "Set the maximum buffer size for the HTTP1 connection.\n\nDefault is ~400kb.",
+          "nullable": true,
+          "type": "string"
+        },
+        "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"
+        },
         "http_max_request_bytes": {
           "default": 2000000,
           "description": "Limit the size of incoming HTTP requests read from the network, to protect against running out of memory. Default: 2000000 (2 MB)",

apollo-router/src/plugins/limits/mod.rs

@@ -4,6 +4,7 @@ mod limited;
 use std::error::Error;
 
 use async_trait::async_trait;
+use bytesize::ByteSize;
 use http::StatusCode;
 use schemars::JsonSchema;
 use serde::Deserialize;
@@ -101,6 +102,20 @@ pub(crate) struct Config {
     /// Limit the size of incoming HTTP requests read from the network,
     /// to protect against running out of memory. Default: 2000000 (2 MB)
     pub(crate) http_max_request_bytes: usize,
+
+    /// 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.
+    #[schemars(with = "Option<String>", default)]
+    pub(crate) experimental_http1_max_buf_size: Option<ByteSize>,
 }
 
 impl Default for Config {
@@ -113,6 +128,8 @@ impl Default for Config {
             max_aliases: None,
             warn_only: false,
             http_max_request_bytes: 2_000_000,
+            experimental_http1_max_headers: None,
+            experimental_http1_max_buf_size: None,
             parser_max_tokens: 15_000,
 
             // This is `apollo-parser`’s default, which protects against stack overflow

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#"
+            limits:
+              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#"
+            limits:
+              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_errors_on_http1_header_that_does_not_fit_inside_buffer(
+) -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            limits:
+              experimental_http1_max_buf_size: 100kb
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    headers.insert("test-header".to_string(), "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_allow_to_change_http1_max_buf_size() -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(
+            r#"
+            limits:
+              experimental_http1_max_buf_size: 2mb
+            "#,
+        )
+        .build()
+        .await;
+
+    router.start().await;
+    router.assert_started().await;
+
+    let mut headers = HashMap::new();
+    headers.insert("test-header".to_string(), "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(())
+}