fix(sources): cap compressed and decompressed payload size to prevent OOM

Cargo.toml

@@ -494,7 +494,7 @@ vrl.workspace = true
 rustls = "0.23"
 tokio-rustls = "0.26"
 wiremock = "0.6.2"
-zstd = { version = "0.13.0", default-features = false }
+zstd.workspace = true
 
 [patch.crates-io]
 # The upgrade for `tokio-util` >= 0.6.9 is blocked on https://github.com/vectordotdev/vector/issues/11257.
@@ -655,10 +655,10 @@ sources-metrics = [
 sources-amqp = ["lapin"]
 sources-apache_metrics = ["sources-utils-http-client"]
 sources-aws_ecs_metrics = ["sources-utils-http-client"]
-sources-aws_kinesis_firehose = ["dep:base64"]
+sources-aws_kinesis_firehose = ["dep:base64", "sources-utils-http-encoding"]
 sources-aws_s3 = ["aws-core", "dep:aws-sdk-sqs", "dep:aws-sdk-s3", "dep:semver", "dep:async-compression", "sources-aws_sqs", "tokio-util/io"]
 sources-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
-sources-datadog_agent = ["sources-utils-http-error", "protobuf-build", "dep:prost"]
+sources-datadog_agent = ["sources-utils-http-encoding", "sources-utils-http-error", "protobuf-build", "dep:prost"]
 sources-demo_logs = ["dep:fakedata"]
 sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "dep:dnstap-parser", "protobuf-build", "dep:prost"]
 sources-docker_logs = ["docker"]
@@ -694,7 +694,7 @@ sources-prometheus-pushgateway = ["sinks-prometheus", "sources-utils-http", "vec
 sources-pulsar = ["dep:apache-avro", "dep:pulsar"]
 sources-redis = ["dep:redis"]
 sources-socket = ["sources-utils-net", "tokio-util/net"]
-sources-splunk_hec = ["dep:roaring"]
+sources-splunk_hec = ["dep:roaring", "sources-utils-http-encoding"]
 sources-statsd = ["sources-utils-net", "tokio-util/net"]
 sources-stdin = ["tokio-util/io"]
 sources-syslog = ["codecs-syslog", "sources-utils-net", "tokio-util/net"]

clippy.toml

@@ -4,11 +4,20 @@ cognitive-complexity-threshold = 75
 # https://rust-lang.github.io/rust-clippy/master/index.html#disallowed_method
 disallowed-methods = [
   { path = "std::io::Write::write", reason = "This doesn't handle short writes, use `write_all` instead." },
+  { path = "zstd::stream::copy_decode", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "zstd::stream::decode_all", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "zstd::bulk::decompress", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "warp::body::bytes", reason = "Reads the whole request body into memory unbounded. Use `crate::sources::util::http::capped_body()` to cap the compressed body size at the global decompressed-size limit." },
 ]
 
 disallowed-types = [
   { path = "once_cell::sync::OnceCell", reason = "Use `std::sync::OnceLock` instead." },
   { path = "once_cell::unsync::OnceCell", reason = "Use `std::cell::OnceCell` instead." },
   { path = "once_cell::sync::Lazy", reason = "Use `std::sync::LazyLock` instead." },
   { path = "once_cell::unsync::Lazy", reason = "Use `std::sync::LazyCell` instead." },
+  { path = "flate2::read::GzDecoder", reason = "Use `vector_common::decompression::CappedDecoder::gzip` to enforce the decompression size cap." },
+  { path = "flate2::read::MultiGzDecoder", reason = "Use `vector_common::decompression::CappedDecoder::gzip` to enforce the decompression size cap." },
+  { path = "flate2::read::ZlibDecoder", reason = "Use `vector_common::decompression::CappedDecoder::zlib` to enforce the decompression size cap." },
+  { path = "flate2::read::DeflateDecoder", reason = "Use `vector_common::decompression::CappedDecoder` for decompression with the size cap." },
+  { path = "zstd::stream::read::Decoder", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
 ]

lib/codecs/src/decoding/framing/chunked_gelf.rs

@@ -2,18 +2,17 @@ use super::{BoxedFramingError, FramingError};
 use crate::{BytesDecoder, StreamDecodingError};
 use bytes::{Buf, Bytes, BytesMut};
 use derivative::Derivative;
-use flate2::read::{MultiGzDecoder, ZlibDecoder};
 use snafu::{ensure, ResultExt, Snafu};
 use std::any::Any;
 use std::collections::HashMap;
-use std::io::Read;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio;
 use tokio::task::JoinHandle;
 use tokio_util::codec::Decoder;
 use tracing::{debug, trace, warn};
 use vector_common::constants::{GZIP_MAGIC, ZLIB_MAGIC};
+use vector_common::decompression::CappedDecoder;
 use vector_config::configurable_component;
 
 const GELF_MAGIC: &[u8] = &[0x1e, 0x0f];
@@ -210,22 +209,14 @@ impl ChunkedGelfDecompression {
 
     pub fn decompress(&self, data: Bytes) -> Result<Bytes, ChunkedGelfDecompressionError> {
         let decompressed = match self {
-            Self::Gzip => {
-                let mut decoder = MultiGzDecoder::new(data.reader());
-                let mut decompressed = Vec::new();
-                decoder
-                    .read_to_end(&mut decompressed)
-                    .context(GzipDecompressionSnafu)?;
-                Bytes::from(decompressed)
-            }
-            Self::Zlib => {
-                let mut decoder = ZlibDecoder::new(data.reader());
-                let mut decompressed = Vec::new();
-                decoder
-                    .read_to_end(&mut decompressed)
-                    .context(ZlibDecompressionSnafu)?;
-                Bytes::from(decompressed)
-            }
+            Self::Gzip => CappedDecoder::gzip(data.reader())
+                .decompress()
+                .map(Bytes::from)
+                .context(GzipDecompressionSnafu)?,
+            Self::Zlib => CappedDecoder::zlib(data.reader())
+                .decompress()
+                .map(Bytes::from)
+                .context(ZlibDecompressionSnafu)?,
             Self::None => data,
         };
         Ok(decompressed)
@@ -1278,4 +1269,85 @@ mod tests {
 
         assert_eq!(detected_compression, ChunkedGelfDecompression::None);
     }
+
+    /// OBE-10706: a GELF payload used to be inflated with an unbounded `read_to_end`, so a small
+    /// datagram could drive an arbitrarily large allocation.
+    ///
+    /// `MultiGzDecoder` walks every concatenated member, so one cheap member repeated past the cap
+    /// is enough to exceed it — no single oversized member required.
+    #[test]
+    fn gzip_decompression_is_capped() {
+        use vector_common::decompression::DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES;
+
+        let member = Compression::Gzip.compress(&vec![0u8; 1024 * 1024]);
+        let members = DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES / (1024 * 1024) + 1;
+        let mut bomb = BytesMut::new();
+        for _ in 0..members {
+            bomb.put_slice(&member);
+        }
+        let bomb = bomb.freeze();
+
+        assert!(
+            bomb.len() < 1024 * 1024,
+            "the bomb must stay small on the wire to be a meaningful test, got {} bytes",
+            bomb.len()
+        );
+
+        let error = ChunkedGelfDecompression::Gzip
+            .decompress(bomb)
+            .expect_err("a payload inflating past the cap must be rejected");
+
+        assert!(matches!(
+            error,
+            ChunkedGelfDecompressionError::GzipDecompression { .. }
+        ));
+    }
+
+    /// The zlib arm needs its own bomb: unlike gzip, zlib has no concatenated-stream form, so the
+    /// payload must be a single oversized stream. Fed to the encoder in chunks to keep the test's
+    /// own memory bounded.
+    #[test]
+    fn zlib_decompression_is_capped() {
+        use std::io::Write as IoWrite;
+
+        use vector_common::decompression::DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES;
+
+        let mut encoder = ZlibEncoder::new(Vec::new(), flate2::Compression::best());
+        let chunk = vec![0u8; 1024 * 1024];
+        for _ in 0..(DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES / (1024 * 1024) + 1) {
+            encoder.write_all(&chunk).unwrap();
+        }
+        let bomb = Bytes::from(encoder.finish().unwrap());
+
+        assert!(
+            bomb.len() < 1024 * 1024,
+            "the bomb must stay small on the wire to be a meaningful test, got {} bytes",
+            bomb.len()
+        );
+
+        let error = ChunkedGelfDecompression::Zlib
+            .decompress(bomb)
+            .expect_err("a payload inflating past the cap must be rejected");
+
+        assert!(matches!(
+            error,
+            ChunkedGelfDecompressionError::ZlibDecompression { .. }
+        ));
+    }
+
+    /// The cap must not disturb ordinary traffic. Zlib shares the same `CappedDecoder` wrapper as
+    /// gzip, so exercising both here covers the wiring of each arm.
+    #[rstest]
+    #[case(Compression::Gzip)]
+    #[case(Compression::Zlib)]
+    fn decompression_under_the_cap_is_unaffected(#[case] compression: Compression) {
+        let payload = "the quick brown fox".repeat(1024);
+        let compressed = compression.compress(&payload);
+
+        let decompressed = ChunkedGelfDecompression::from_magic(&compressed)
+            .decompress(compressed)
+            .expect("a payload within the cap must decompress");
+
+        assert_eq!(decompressed, Bytes::from(payload));
+    }
 }

lib/vector-common/Cargo.toml

@@ -41,6 +41,7 @@ bytes = { version = "1.9.0", default-features = false }
 chrono.workspace = true
 crossbeam-utils = { version = "0.8.20", default-features = false }
 derivative = { version = "2.2.0", default-features = false }
+flate2.workspace = true
 futures.workspace = true
 indexmap.workspace = true
 metrics.workspace = true
@@ -60,6 +61,7 @@ snafu.workspace = true
 regex.workspace = true
 tokio-util.workspace = true
 serde_with.workspace = true
+zstd.workspace = true
 
 [dev-dependencies]
 futures = { version = "0.3.31", default-features = false, features = ["async-await"] }