apollo_http_server: accept compressed HTTP requests (gzip, zstd, brotli) (#13366)

Author: avi-starkware

Cargo.lock

@@ -42,6 +42,7 @@ serde_json.workspace = true
 starknet_api.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt"] }
+tower-http = { workspace = true, features = ["decompression-full", "limit"] }
 tracing.workspace = true
 
 
@@ -62,3 +63,4 @@ rstest.workspace = true
 serde_json.workspace = true
 starknet-types-core.workspace = true
 tracing-test.workspace = true
+zstd.workspace = true

crates/apollo_http_server/Cargo.toml

@@ -35,6 +35,8 @@ use starknet_api::transaction::fields::ValidResourceBounds;
 use tokio::net::TcpListener;
 use tokio::sync::watch::{channel, Receiver, Sender};
 use tokio::time;
+use tower_http::decompression::RequestDecompressionLayer;
+use tower_http::limit::RequestBodyLimitLayer;
 use tracing::{debug, info, instrument, warn};
 
 use crate::deprecated_gateway_transaction::DeprecatedGatewayTransactionV3;
@@ -142,6 +144,13 @@ impl HttpServer {
                 get(|| futures::future::ready("Gateway is ready".to_owned()))
             )
             .layer(Extension(self.app_state.clone()))
+            // Hard streaming limit on decompressed bytes — wraps the body in
+            // http_body_util::Limited which errors during poll_frame() once the
+            // limit is exceeded, preventing zip bombs from expanding in memory.
+            .layer(RequestBodyLimitLayer::new(self.config.static_config.max_request_body_size))
+            .layer(RequestDecompressionLayer::new())
+            // Cap compressed wire bytes to bound network I/O.
+            .layer(RequestBodyLimitLayer::new(self.config.static_config.max_request_body_size))
     }
 
     fn post_method_router<H, T, S>(&self, handler: H) -> MethodRouter<S>

crates/apollo_http_server/src/http_server.rs

@@ -1,3 +1,5 @@
+use std::io::Write;
+
 use apollo_gateway_types::communication::{GatewayClientError, MockGatewayClient};
 use apollo_gateway_types::deprecated_gateway_error::{
     KnownStarknetErrorCode,
@@ -385,3 +387,67 @@ async fn request_body_size_limit_enforced(
     let response = http_client.add_tx(tx).await;
     assert_eq!(response.status(), expected_status, "Unexpected status: {}", response.status());
 }
+
+#[tokio::test]
+async fn zstd_compressed_request_decompression() {
+    let mut mock_gateway_client = MockGatewayClient::new();
+    mock_gateway_client.expect_add_tx().times(1).return_const(Ok(default_gateway_output()));
+
+    let http_client = HttpClientServerSetupBuilder::new(unique_u16!())
+        .with_mock_gateway_client(mock_gateway_client)
+        .build()
+        .await;
+
+    let tx_json = serde_json::to_string(&rpc_invoke_tx()).unwrap();
+    let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), 0).unwrap();
+    encoder.write_all(tx_json.as_bytes()).unwrap();
+    let compressed_body = encoder.finish().unwrap();
+
+    let response = http_client.add_rpc_tx_with_zstd(compressed_body).await;
+
+    assert_eq!(response.status(), StatusCode::OK, "Request should be decompressed and handled");
+    let response_body = response.text().await.unwrap();
+    let gateway_output: GatewayOutput =
+        serde_json::from_str(&response_body).expect("Response should be valid GatewayOutput");
+    assert_eq!(gateway_output.transaction_hash(), EXPECTED_TX_HASH);
+}
+
+#[tokio::test]
+async fn zstd_compressed_request_too_large() {
+    let tx_json = serde_json::to_string(&rpc_invoke_tx()).unwrap();
+    let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), 0).unwrap();
+    encoder.write_all(tx_json.as_bytes()).unwrap();
+    let compressed_body = encoder.finish().unwrap();
+
+    let http_client = HttpClientServerSetupBuilder::new(unique_u16!())
+        .with_max_request_body_size(compressed_body.len() - 1)
+        .build()
+        .await;
+
+    let response = http_client.add_rpc_tx_with_zstd(compressed_body).await;
+
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+}
+
+#[tokio::test]
+async fn zstd_decompressed_request_too_large() {
+    // 10 KB of repeated bytes — compresses to ~50 bytes with zstd.
+    let large_body = vec![b'a'; 10 * 1024];
+    let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), 0).unwrap();
+    encoder.write_all(&large_body).unwrap();
+    let compressed_body = encoder.finish().unwrap();
+
+    // Limit between compressed size and decompressed size.
+    // compressed_body is ~50 bytes; decompressed is 10240 bytes.
+    let max_request_body_size = large_body.len() - 1;
+    assert!(compressed_body.len() < max_request_body_size);
+
+    let http_client = HttpClientServerSetupBuilder::new(unique_u16!())
+        .with_max_request_body_size(max_request_body_size)
+        .build()
+        .await;
+
+    let response = http_client.add_rpc_tx_with_zstd(compressed_body).await;
+
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+}

crates/apollo_http_server/src/http_server_test.rs

@@ -94,6 +94,17 @@ impl HttpTestClient {
             .await
             .unwrap()
     }
+
+    pub async fn add_rpc_tx_with_zstd(&self, compressed_body: Vec<u8>) -> Response {
+        self.client
+            .post(format!("http://{}/gateway/add_rpc_transaction", self.socket))
+            .header("content-type", "application/json")
+            .header("content-encoding", "zstd")
+            .body(Body::from(compressed_body))
+            .send()
+            .await
+            .unwrap()
+    }
 }
 
 pub fn create_http_server_config(socket: SocketAddr) -> HttpServerConfig {