fix(sources): cap decompressed body to prevent OOM

changelog.d/http_source_decompression_bomb.security.md

@@ -0,0 +1,3 @@
+HTTP-based sources (`http_server`, `prometheus_pushgateway`, `prometheus_remote_write`, `heroku_logs`, `opentelemetry`) now cap decompressed request bodies at 100 MiB. Previously, a single unauthenticated request carrying a compressed payload (e.g. a gzip bomb) could allocate unbounded memory and OOM-kill the Vector process. Decompressed payloads exceeding the cap are rejected with HTTP 413, as are requests whose declared `Content-Length` exceeds the same limit.
+
+authors: pront

src/sources/http_server.rs

@@ -1328,6 +1328,41 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn http_rejects_gzip_bomb_with_413() {
+        // A modestly-sized gzipped blob of zeros that would expand past the default
+        // 100 MiB cap if decompression were unbounded.
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        let chunk = [0u8; 8 * 1024];
+        for _ in 0..(200 * 1024 * 1024 / chunk.len()) {
+            encoder.write_all(&chunk).unwrap();
+        }
+        let body = encoder.finish().unwrap();
+
+        let mut headers = HeaderMap::new();
+        headers.insert("Content-Encoding", "gzip".parse().unwrap());
+
+        components::init_test();
+        let (_rx, addr) = source(
+            vec![],
+            vec![],
+            "http_path",
+            "remote_ip",
+            "/",
+            "POST",
+            StatusCode::OK,
+            None,
+            true,
+            EventStatus::Delivered,
+            true,
+            None,
+            None,
+        )
+        .await;
+
+        assert_eq!(413, send_bytes(addr, body, headers).await);
+    }
+
     #[tokio::test]
     async fn http_path() {
         let mut events = assert_source_compliance(&HTTP_PUSH_SOURCE_TAGS, async {

src/sources/opentelemetry/http.rs

@@ -39,7 +39,10 @@ use crate::{
     sources::{
         http_server::HttpConfigParamKind,
         opentelemetry::config::{LOGS, METRICS, OpentelemetryConfig, TRACES},
-        util::{add_headers, decompress_body},
+        util::{
+            add_headers, decompress_body,
+            http::{DEFAULT_MAX_DECOMPRESSED_BODY_SIZE, limited_body},
+        },
     },
     tls::MaybeTlsSettings,
 };
@@ -191,6 +194,8 @@ where
         + 'static
         + Fn(Option<String>, HeaderMap, Bytes) -> Result<Vec<Event>, ErrorMessage>,
 {
+    let body_filter = limited_body(DEFAULT_MAX_DECOMPRESSED_BODY_SIZE);
+
     warp::post()
         .and(warp::path("v1"))
         .and(warp::path(telemetry_type))
@@ -201,7 +206,7 @@ where
         ))
         .and(warp::header::optional::<String>("content-encoding"))
         .and(warp::header::headers_cloned())
-        .and(warp::body::bytes())
+        .and(body_filter)
         .and_then(
             move |encoding_header: Option<String>, headers: HeaderMap, body: Bytes| {
                 let events = make_events(encoding_header, headers, body);