CVE-2024-28101: Insufficient Information (NVD-CWE-noinfo)
value: The Apollo Router is a graph router written in Rust to run a federated supergraph that uses Apollo Federation. Versions 0.9.5 until 1.40.2 are subject to a Denial-of-Service (DoS) type vulnerability. When receiving compressed HTTP payloads, affected versions of the Router evaluate the `limits.http_max_request_bytes` configuration option after the entirety of the compressed payload is decompressed. If affected versions of the Router receive highly compressed payloads, this could result in significant memory consumption while the compressed payload is expanded. Router version 1.40.2 has a fix for the vulnerability. Those who are unable to upgrade may be able to implement mitigations at proxies or load balancers positioned in front of their Router fleet (e.g. Nginx, HAProxy, or cloud-native WAF services) by creating limits on HTTP body upload size. 

CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
Published: 2024-03-21T02:52Z
Repository: https://github.com/apollographql/router
Language: Rust
Files changed: apollo-router/src/axum_factory/axum_http_server_factory.rs, apollo-router/src/axum_factory/utils.rs, apollo-router/src/services/http/service.rs, apollo-router/src/services/layers/content_negotiation.rs, apollo-router/tests/common.rs, apollo-router/tests/telemetry/metrics.rs
Commit date: 2024-03-06 10:26:44 +0000
Commit message: streaming body decompression

Client request body was fully loaded in memory before decompressing, we
are now decompressing it as it goes

Diff: diff --git a/apollo-router/src/axum_factory/axum_http_server_factory.rs b/apollo-router/src/axum_factory/axum_http_server_factory.rs
index d0632fe96..c59842076 100644
--- a/apollo-router/src/axum_factory/axum_http_server_factory.rs
+++ b/apollo-router/src/axum_factory/axum_http_server_factory.rs
@@ -4,10 +4,11 @@ use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Instant;
 
+use axum::error_handling::HandleErrorLayer;
 use axum::extract::Extension;
 use axum::extract::State;
 use axum::http::StatusCode;
 use axum::middleware;
 use axum::middleware::Next;
@@ -30,18 +31,19 @@ use serde::Serialize;
 use tokio::net::UnixListener;
 use tokio::sync::mpsc;
 use tokio_rustls::TlsAcceptor;
 use tower::service_fn;
 use tower::BoxError;
+use tower::ServiceBuilder;
 use tower::ServiceExt;
+use tower_http::decompression::DecompressionBody;
 use tower_http::trace::TraceLayer;
 
 use super::listeners::ensure_endpoints_consistency;
 use super::listeners::ensure_listenaddrs_consistency;
 use super::listeners::extra_endpoints;
 use super::listeners::ListenersAndRouters;
-use super::utils::decompress_request_body;
 use super::utils::PropagatingMakeSpan;
 use super::ListenAddrAndRouter;
 use super::ENDPOINT_CALLBACK;
 use crate::axum_factory::compression::Compressor;
 use crate::axum_factory::listeners::get_extra_listeners;
@@ -55,10 +57,11 @@ use crate::plugins::telemetry::SpanMode;
 use crate::plugins::traffic_shaping::Elapsed;
 use crate::plugins::traffic_shaping::RateLimited;
 use crate::router::ApolloRouterError;
 use crate::router_factory::Endpoint;
 use crate::router_factory::RouterFactory;
+use crate::services::http::service::BodyStream;
 use crate::services::router;
 use crate::uplink::license_enforcement::LicenseState;
 use crate::uplink::license_enforcement::APOLLO_ROUTER_LICENSE_EXPIRED;
 use crate::uplink::license_enforcement::LICENSE_EXPIRED_SHORT_MESSAGE;
 
@@ -171,15 +174,13 @@ where
                         }
                     };
                     tracing::trace!(?health, request = ?req.router_request, "health check");
                     async move {
                         Ok(router::Response {
-                            response: http::Response::builder()
-                                .status(status_code)
-                                .body::<hyper::Body>(
-                                    serde_json::to_vec(&health).map_err(BoxError::from)?.into(),
-                                )?,
+                            response: http::Response::builder().status(status_code).body::<Body>(
+                                serde_json::to_vec(&health).map_err(BoxError::from)?.into(),
+                            )?,
                             context: req.context,
                         })
                     }
                 })
                 .boxed(),
@@ -420,10 +421,14 @@ pub(crate) fn span_mode(configuration: &Configuration) -> SpanMode {
                 .and_then(|v| serde_json::from_value(v.clone()).ok())
         })
         .unwrap_or_default()
 }
 
+async fn decompression_error(_error: BoxError) -> axum::response::Response {
+    (StatusCode::BAD_REQUEST, "cannot decompress request body").into_response()
+}
+
 fn main_endpoint<RF>(
     service_factory: RF,
     configuration: &Configuration,
     endpoints_on_main_listener: Vec<Endpoint>,
     license: LicenseState,
@@ -434,12 +439,20 @@ where
     let cors = configuration.cors.clone().into_layer().map_err(|e| {
         ApolloRouterError::ServiceCreationError(format!("CORS configuration error: {e}").into())
     })?;
     let span_mode = span_mode(configuration);
 
+    let decompression = ServiceBuilder::new()
+        .layer(HandleErrorLayer::<_, ()>::new(decompression_error))
+        .layer(
+            tower_http::decompression::RequestDecompressionLayer::new()
+                .br(true)
+                .gzip(true)
+                .deflate(true),
+        );
     let mut main_route = main_router::<RF>(configuration)
-        .layer(middleware::from_fn(decompress_request_body))
+        .layer(decompression)
         .layer(middleware::from_fn_with_state(
             (license, Instant::now(), Arc::new(AtomicU64::new(0))),
             license_handler,
         ))
         .layer(Extension(service_factory))
@@ -528,38 +541,42 @@ async fn license_handler<B>(
     } else {
         next.run(request).await
     }
 }
 
-pub(super) fn main_router<RF>(configuration: &Configuration) -> axum::Router
+pub(super) fn main_router<RF>(
+    configuration: &Configuration,
+) -> axum::Router<(), DecompressionBody<Body>>
 where
     RF: RouterFactory,
 {
     let mut router = Router::new().route(
         &configuration.supergraph.sanitized_path(),
         get({
-            move |Extension(service): Extension<RF>, request: Request<Body>| {
+            move |Extension(service): Extension<RF>, request: Request<DecompressionBody<Body>>| {
                 handle_graphql(service.create().boxed(), request)
             }
         })
         .post({
-            move |Extension(service): Extension<RF>, request: Request<Body>| {
+            move |Extension(service): Extension<RF>, request: Request<DecompressionBody<Body>>| {
                 handle_graphql(service.create().boxed(), request)
             }
         }),
     );
 
     if configuration.supergraph.path == "/*" {
         router = router.route(
             "/",
             get({
-                move |Extension(service): Extension<RF>, request: Request<Body>| {
+                move |Extension(service): Extension<RF>,
+                      request: Request<DecompressionBody<Body>>| {
                     handle_graphql(service.create().boxed(), request)
                 }
             })
             .post({
-                move |Extension(service): Extension<RF>, request: Request<Body>| {
+                move |Extension(service): Extension<RF>,
+                      request: Request<DecompressionBody<Body>>| {
                     handle_graphql(service.create().boxed(), request)
                 }
             }),
         );
     }
@@ -567,14 +584,18 @@ where
     router
 }
 
 async fn handle_graphql(
     service: router::BoxService,
-    http_request: Request<Body>,
+    http_request: Request<DecompressionBody<Body>>,
 ) -> impl IntoResponse {
     let _guard = SessionCountGuard::start();
 
+    let (parts, body) = http_request.into_parts();
+
+    let http_request = http::Request::from_parts(parts, Body::wrap_stream(BodyStream::new(body)));
+
     let request: router::Request = http_request.into();
     let context = request.context.clone();
     let accept_encoding = request
         .router_request
         .headers()
diff --git a/apollo-router/src/axum_factory/utils.rs b/apollo-router/src/axum_factory/utils.rs
index 96c42bb6c..d1e4f51a5 100644
--- a/apollo-router/src/axum_factory/utils.rs
+++ b/apollo-router/src/axum_factory/utils.rs
@@ -1,22 +1,11 @@
 //! Utilities used for [`super::AxumHttpServerFactory`]
 
 use std::net::SocketAddr;
 
-use async_compression::tokio::write::BrotliDecoder;
-use async_compression::tokio::write::GzipDecoder;
-use async_compression::tokio::write::ZlibDecoder;
-use axum::http::StatusCode;
-use axum::middleware::Next;
-use axum::response::*;
-use futures::prelude::*;
-use http::header::CONTENT_ENCODING;
-use http::Request;
-use hyper::Body;
 use opentelemetry::global;
 use opentelemetry::trace::TraceContextExt;
-use tokio::io::AsyncWriteExt;
 use tower_http::trace::MakeSpan;
 use tower_service::Service;
 use tracing::Span;
 
 use crate::plugins::telemetry::SpanMode;
@@ -24,87 +13,10 @@ use crate::plugins::telemetry::OTEL_STATUS_CODE;
 use crate::uplink::license_enforcement::LicenseState;
 use crate::uplink::license_enforcement::LICENSE_EXPIRED_SHORT_MESSAGE;
 
 pub(crate) const REQUEST_SPAN_NAME: &str = "request";
 
-pub(super) async fn decompress_request_body(
-    req: Request<Body>,
-    next: Next<Body>,
-) -> Result<Response, Response> {
-    let (parts, body) = req.into_parts();
-    let content_encoding = parts.headers.get(&CONTENT_ENCODING);
-    macro_rules! decode_body {
-        ($decoder: ident, $error_message: expr) => {{
-            let body_bytes = hyper::body::to_bytes(body)
-                .map_err(|err| {
-                    (
-                        StatusCode::BAD_REQUEST,
-                        format!("cannot read request body: {err}"),
-                    )
-                        .into_response()
-                })
-                .await?;
-            let mut decoder = $decoder::new(Vec::new());
-            decoder.write_all(&body_bytes).await.map_err(|err| {
-                (
-                    StatusCode::BAD_REQUEST,
-                    format!("{}: {err}", $error_message),
-                )
-                    .into_response()
-            })?;
-            decoder.shutdown().await.map_err(|err| {
-                (
-                    StatusCode::BAD_REQUEST,
-                    format!("{}: {err}", $error_message),
-                )
-                    .into_response()
-            })?;
-
-            Ok(next
-                .run(Request::from_parts(parts, Body::from(decoder.into_inner())))
-                .await)
-        }};
-    }
-
-    match content_encoding {
-        Some(content_encoding) => match content_encoding.to_str() {
-            Ok(content_encoding_str) => match content_encoding_str {
-                "br" => decode_body!(BrotliDecoder, "cannot decompress (brotli) request body"),
-                "gzip" => decode_body!(GzipDecoder, "cannot decompress (gzip) request body"),
-                "deflate" => decode_body!(ZlibDecoder, "cannot decompress (deflate) request body"),
-                "identity" => Ok(next.run(Request::from_parts(parts, body)).await),
-                unknown => {
-                    let message = format!("unknown content-encoding header value {unknown:?}");
-                    tracing::error!(message);
-                    u64_counter!(
-                        "apollo_router_http_requests_total",
-                        "Total number of HTTP requests made.",
-                        1,
-                        status = StatusCode::BAD_REQUEST.as_u16() as i64,
-                        error = message.clone()
-                    );
-
-                    Err((StatusCode::BAD_REQUEST, message).into_response())
-                }
-            },
-
-            Err(err) => {
-                let message = format!("cannot read content-encoding header: {err}");
-                u64_counter!(
-                    "apollo_router_http_requests_total",
-                    "Total number of HTTP requests made.",
-                    1,
-                    status = 400,
-                    error = message.clone()
-                );
-                Err((StatusCode::BAD_REQUEST, message).into_response())
-            }
-        },
-        None => Ok(next.run(Request::from_parts(parts, body)).await),
-    }
-}
-
 #[derive(Clone, Default)]
 pub(crate) struct PropagatingMakeSpan {
     pub(crate) license: LicenseState,
     pub(crate) span_mode: SpanMode,
 }
diff --git a/apollo-router/src/services/http/service.rs b/apollo-router/src/services/http/service.rs
index d9fc6f722..256320dba 100644
--- a/apollo-router/src/services/http/service.rs
+++ b/apollo-router/src/services/http/service.rs
@@ -389,10 +389,17 @@ pin_project! {
         #[pin]
         inner: DecompressionBody<B>
     }
 }
 
+impl<B: hyper::body::HttpBody> BodyStream<B> {
+    /// Create a new `BodyStream`.
+    pub(crate) fn new(body: DecompressionBody<B>) -> Self {
+        Self { inner: body }
+    }
+}
+
 impl<B> Stream for BodyStream<B>
 where
     B: hyper::body::HttpBody,
     B::Error: Into<tower_http::BoxError>,
 {
diff --git a/apollo-router/src/services/layers/content_negotiation.rs b/apollo-router/src/services/layers/content_negotiation.rs
index fe33d3adb..59ccc5498 100644
--- a/apollo-router/src/services/layers/content_negotiation.rs
+++ b/apollo-router/src/services/layers/content_negotiation.rs
@@ -69,10 +69,21 @@ where
                                 ]
                             })
                             .to_string(),
                         ))
                         .expect("cannot fail");
+                    u64_counter!(
+                        "apollo_router_http_requests_total",
+                        "Total number of HTTP requests made.",
+                        1,
+                        status = StatusCode::UNSUPPORTED_MEDIA_TYPE.as_u16() as i64,
+                        error = format!(
+                            r#"'content-type' header must be one of: {:?} or {:?}"#,
+                            APPLICATION_JSON.essence_str(),
+                            GRAPHQL_JSON_RESPONSE_HEADER_VALUE,
+                        )
+                    );
 
                     return Ok(ControlFlow::Break(response.into()));
                 }
 
                 let accepts = parse_accept(req.router_request.headers());
diff --git a/apollo-router/tests/common.rs b/apollo-router/tests/common.rs
index e1d09bc09..798cf7558 100644
--- a/apollo-router/tests/common.rs
+++ b/apollo-router/tests/common.rs
@@ -8,11 +8,10 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::time::SystemTime;
 
 use buildstructor::buildstructor;
 use http::header::ACCEPT;
-use http::header::CONTENT_ENCODING;
 use http::header::CONTENT_TYPE;
 use http::HeaderValue;
 use jsonpath_lib::Selector;
 use mime::APPLICATION_JSON;
 use once_cell::sync::OnceCell;
@@ -376,20 +375,20 @@ impl IntegrationTest {
     ) -> impl std::future::Future<Output = (String, reqwest::Response)> {
         self.execute_query_internal(&json!({"query":"query {topProducts{name, name, name, name, name, name, name, name, name, name}}","variables":{}}), None)
     }
 
     #[allow(dead_code)]
-    pub fn execute_bad_content_encoding(
+    pub fn execute_bad_content_type(
         &self,
     ) -> impl std::future::Future<Output = (String, reqwest::Response)> {
         self.execute_query_internal(&json!({"garbage":{}}), Some("garbage"))
     }
 
     fn execute_query_internal(
         &self,
         query: &Value,
-        content_encoding: Option<&'static str>,
+        content_type: Option<&'static str>,
     ) -> impl std::future::Future<Output = (String, reqwest::Response)> {
         assert!(
             self.router.is_some(),
             "router was not started, call `router.start().await; router.assert_started().await`"
         );
@@ -402,12 +401,14 @@ impl IntegrationTest {
             async move {
                 let client = reqwest::Client::new();
 
                 let mut request = client
                     .post("http://localhost:4000")
-                    .header(CONTENT_TYPE, APPLICATION_JSON.essence_str())
-                    .header(CONTENT_ENCODING, content_encoding.unwrap_or("identity"))
+                    .header(
+                        CONTENT_TYPE,
+                        content_type.unwrap_or(APPLICATION_JSON.essence_str()),
+                    )
                     .header("apollographql-client-name", "custom_name")
                     .header("apollographql-client-version", "1.0")
                     .header("x-my-header", "test")
                     .json(&query)
                     .build()
diff --git a/apollo-router/tests/telemetry/metrics.rs b/apollo-router/tests/telemetry/metrics.rs
index 5d859a1e4..a195566b1 100644
--- a/apollo-router/tests/telemetry/metrics.rs
+++ b/apollo-router/tests/telemetry/metrics.rs
@@ -133,14 +133,15 @@ async fn test_bad_queries() {
         .assert_metrics_contains(
             r#"apollo_router_http_requests_total{status="200",otel_scope_name="apollo/router"}"#,
             None,
         )
         .await;
-    router.execute_bad_content_encoding().await;
+    router.execute_bad_content_type().await;
+
     router
             .assert_metrics_contains(
-                r#"apollo_router_http_requests_total{error="unknown content-encoding header value \"garbage\"",status="400",otel_scope_name="apollo/router"}"#,
+                r#"apollo_router_http_requests_total{error="'content-type' header must be one of: \"application/json\" or \"application/graphql-response+json\"",status="415",otel_scope_name="apollo/router"}"#,
                 None,
             )
             .await;
 
     router.execute_bad_query().await;

Vulnerable code: use super::utils::decompress_request_body;
                            response: http::Response::builder()
                                .status(status_code)
                                .body::<hyper::Body>(
                                    serde_json::to_vec(&health).map_err(BoxError::from)?.into(),
                                )?,
        .layer(middleware::from_fn(decompress_request_body))
pub(super) fn main_router<RF>(configuration: &Configuration) -> axum::Router
            move |Extension(service): Extension<RF>, request: Request<Body>| {
            move |Extension(service): Extension<RF>, request: Request<Body>| {
                move |Extension(service): Extension<RF>, request: Request<Body>| {
                move |Extension(service): Extension<RF>, request: Request<Body>| {
    http_request: Request<Body>,
use async_compression::tokio::write::BrotliDecoder;
use async_compression::tokio::write::GzipDecoder;
use async_compression::tokio::write::ZlibDecoder;
use axum::http::StatusCode;
use axum::middleware::Next;
use axum::response::*;
use futures::prelude::*;
use http::header::CONTENT_ENCODING;
use http::Request;
use hyper::Body;
use tokio::io::AsyncWriteExt;
pub(super) async fn decompress_request_body(
    req: Request<Body>,
    next: Next<Body>,
) -> Result<Response, Response> {
    let (parts, body) = req.into_parts();
    let content_encoding = parts.headers.get(&CONTENT_ENCODING);
    macro_rules! decode_body {
        ($decoder: ident, $error_message: expr) => {{
            let body_bytes = hyper::body::to_bytes(body)
                .map_err(|err| {
                    (
                        StatusCode::BAD_REQUEST,
                        format!("cannot read request body: {err}"),
                    )
                        .into_response()
                })
                .await?;
            let mut decoder = $decoder::new(Vec::new());
            decoder.write_all(&body_bytes).await.map_err(|err| {
                (
                    StatusCode::BAD_REQUEST,
                    format!("{}: {err}", $error_message),
                )
                    .into_response()
            })?;
            decoder.shutdown().await.map_err(|err| {
                (
                    StatusCode::BAD_REQUEST,
                    format!("{}: {err}", $error_message),
                )
                    .into_response()
            })?;

            Ok(next
                .run(Request::from_parts(parts, Body::from(decoder.into_inner())))
                .await)
        }};
    }

    match content_encoding {
        Some(content_encoding) => match content_encoding.to_str() {
            Ok(content_encoding_str) => match content_encoding_str {
                "br" => decode_body!(BrotliDecoder, "cannot decompress (brotli) request body"),
                "gzip" => decode_body!(GzipDecoder, "cannot decompress (gzip) request body"),
                "deflate" => decode_body!(ZlibDecoder, "cannot decompress (deflate) request body"),
                "identity" => Ok(next.run(Request::from_parts(parts, body)).await),
                unknown => {
                    let message = format!("unknown content-encoding header value {unknown:?}");
                    tracing::error!(message);
                    u64_counter!(
                        "apollo_router_http_requests_total",
                        "Total number of HTTP requests made.",
                        1,
                        status = StatusCode::BAD_REQUEST.as_u16() as i64,
                        error = message.clone()
                    );

                    Err((StatusCode::BAD_REQUEST, message).into_response())
                }
            },

            Err(err) => {
                let message = format!("cannot read content-encoding header: {err}");
                u64_counter!(
                    "apollo_router_http_requests_total",
                    "Total number of HTTP requests made.",
                    1,
                    status = 400,
                    error = message.clone()
                );
                Err((StatusCode::BAD_REQUEST, message).into_response())
            }
        },
        None => Ok(next.run(Request::from_parts(parts, body)).await),
    }
}

use http::header::CONTENT_ENCODING;
    pub fn execute_bad_content_encoding(
        content_encoding: Option<&'static str>,
                    .header(CONTENT_TYPE, APPLICATION_JSON.essence_str())
                    .header(CONTENT_ENCODING, content_encoding.unwrap_or("identity"))
    router.execute_bad_content_encoding().await;
                r#"apollo_router_http_requests_total{error="unknown content-encoding header value \"garbage\"",status="400",otel_scope_name="apollo/router"}"#,
Fixed code: use axum::error_handling::HandleErrorLayer;
use tower::ServiceBuilder;
use tower_http::decompression::DecompressionBody;
use crate::services::http::service::BodyStream;
                            response: http::Response::builder().status(status_code).body::<Body>(
                                serde_json::to_vec(&health).map_err(BoxError::from)?.into(),
                            )?,
async fn decompression_error(_error: BoxError) -> axum::response::Response {
    (StatusCode::BAD_REQUEST, "cannot decompress request body").into_response()
}

    let decompression = ServiceBuilder::new()
        .layer(HandleErrorLayer::<_, ()>::new(decompression_error))
        .layer(
            tower_http::decompression::RequestDecompressionLayer::new()
                .br(true)
                .gzip(true)
                .deflate(true),
        );
        .layer(decompression)
pub(super) fn main_router<RF>(
    configuration: &Configuration,
) -> axum::Router<(), DecompressionBody<Body>>
            move |Extension(service): Extension<RF>, request: Request<DecompressionBody<Body>>| {
            move |Extension(service): Extension<RF>, request: Request<DecompressionBody<Body>>| {
                move |Extension(service): Extension<RF>,
                      request: Request<DecompressionBody<Body>>| {
                move |Extension(service): Extension<RF>,
                      request: Request<DecompressionBody<Body>>| {
    http_request: Request<DecompressionBody<Body>>,
    let (parts, body) = http_request.into_parts();

    let http_request = http::Request::from_parts(parts, Body::wrap_stream(BodyStream::new(body)));

impl<B: hyper::body::HttpBody> BodyStream<B> {
    /// Create a new `BodyStream`.
    pub(crate) fn new(body: DecompressionBody<B>) -> Self {
        Self { inner: body }
    }
}

                    u64_counter!(
                        "apollo_router_http_requests_total",
                        "Total number of HTTP requests made.",
                        1,
                        status = StatusCode::UNSUPPORTED_MEDIA_TYPE.as_u16() as i64,
                        error = format!(
                            r#"'content-type' header must be one of: {:?} or {:?}"#,
                            APPLICATION_JSON.essence_str(),
                            GRAPHQL_JSON_RESPONSE_HEADER_VALUE,
                        )
                    );
    pub fn execute_bad_content_type(
        content_type: Option<&'static str>,
                    .header(
                        CONTENT_TYPE,
                        content_type.unwrap_or(APPLICATION_JSON.essence_str()),
                    )
    router.execute_bad_content_type().await;

                r#"apollo_router_http_requests_total{error="'content-type' header must be one of: \"application/json\" or \"application/graphql-response+json\"",status="415",otel_scope_name="apollo/router"}"#,