fix(security): [OBE-10709,OBE-10712,OBE-10718,OBE-11232,OBE-11234,OBE-11235,OBE-11236,OBE-11238,OBE-11555,OBE-11556] OOM/unbounded allocation bounds

lib/codecs/Cargo.toml

@@ -37,7 +37,7 @@ smallvec = { version = "1", default-features = false, features = ["union"] }
 snap = { version = "1.1", default-features = false }
 snafu.workspace = true
 syslog_loose = { version = "0.21", default-features = false, optional = true }
-tokio-util = { version = "0.7", default-features = false, features = ["codec"] }
+tokio-util = { version = "0.7", default-features = false, features = ["codec", "time"] }
 tokio.workspace = true
 tracing = { version = "0.1", default-features = false }
 vrl.workspace = true

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

@@ -10,7 +10,6 @@ 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};
@@ -19,11 +18,24 @@ use vector_config::configurable_component;
 const GELF_MAGIC: &[u8] = &[0x1e, 0x0f];
 const GELF_MAX_TOTAL_CHUNKS: u8 = 128;
 const DEFAULT_TIMEOUT_SECS: f64 = 5.0;
+/// Default cap on concurrent incomplete messages. Prevents HashMap from growing unbounded
+/// when senders open many message IDs without completing them.
+pub const DEFAULT_PENDING_MESSAGES_LIMIT: usize = 1000;
+/// Default cap on the reassembled payload of a single GELF message (5 MiB).
+pub const DEFAULT_MAX_MESSAGE_LENGTH: usize = 5 * 1024 * 1024;
 
 const fn default_timeout_secs() -> f64 {
     DEFAULT_TIMEOUT_SECS
 }
 
+fn default_pending_messages_limit() -> Option<usize> {
+    Some(DEFAULT_PENDING_MESSAGES_LIMIT)
+}
+
+fn default_max_message_length() -> Option<usize> {
+    Some(DEFAULT_MAX_MESSAGE_LENGTH)
+}
+
 /// Config used to build a `ChunkedGelfDecoder`.
 #[configurable_component]
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
@@ -58,21 +70,22 @@ pub struct ChunkedGelfDecoderOptions {
 
     /// The maximum number of pending incomplete messages. If this limit is reached, the decoder starts
     /// dropping chunks of new messages, ensuring the memory usage of the decoder's state is bounded.
-    /// If this option is not set, the decoder does not limit the number of pending messages and the memory usage
-    /// of its messages buffer can grow unbounded. This matches Graylog Server's behavior.
-    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
+    /// Defaults to 1000. Set to a very large value to approximate the previous unbounded behavior.
+    #[serde(default = "default_pending_messages_limit")]
+    #[derivative(Default(value = "Some(DEFAULT_PENDING_MESSAGES_LIMIT)"))]
     pub pending_messages_limit: Option<usize>,
 
     /// The maximum length of a single GELF message, in bytes. Messages longer than this length will
-    /// be dropped. If this option is not set, the decoder does not limit the length of messages and
-    /// the per-message memory is unbounded.
+    /// be dropped. Defaults to 5 MiB. Set to a very large value to approximate the previous
+    /// unbounded behavior.
     ///
     /// Note that a message can be composed of multiple chunks and this limit is applied to the whole
     /// message, not to individual chunks.
     ///
     /// This limit takes only into account the message's payload and the GELF header bytes are excluded from the calculation.
     /// The message's payload is the concatenation of all the chunks' payloads.
-    #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
+    #[serde(default = "default_max_message_length")]
+    #[derivative(Default(value = "Some(DEFAULT_MAX_MESSAGE_LENGTH)"))]
     pub max_length: Option<usize>,
 
     /// Decompression configuration for GELF messages.
@@ -126,17 +139,15 @@ struct MessageState {
     chunks: [Bytes; GELF_MAX_TOTAL_CHUNKS as usize],
     chunks_bitmap: u128,
     current_length: usize,
-    timeout_task: JoinHandle<()>,
 }
 
 impl MessageState {
-    pub const fn new(total_chunks: u8, timeout_task: JoinHandle<()>) -> Self {
+    pub const fn new(total_chunks: u8) -> Self {
         Self {
             total_chunks,
             chunks: [const { Bytes::new() }; GELF_MAX_TOTAL_CHUNKS as usize],
             chunks_bitmap: 0,
             current_length: 0,
-            timeout_task,
         }
     }
 
@@ -162,7 +173,6 @@ impl MessageState {
 
     fn retrieve_message(&self) -> Option<Bytes> {
         if self.is_complete() {
-            self.timeout_task.abort();
             let chunks = &self.chunks[0..self.total_chunks as usize];
             let mut message = BytesMut::new();
             for chunk in chunks {
@@ -306,9 +316,12 @@ pub struct ChunkedGelfDecoder {
     bytes_decoder: BytesDecoder,
     decompression_config: ChunkedGelfDecompressionConfig,
     state: Arc<Mutex<HashMap<u64, MessageState>>>,
-    timeout: Duration,
     pending_messages_limit: Option<usize>,
     max_length: Option<usize>,
+    // Sender to the single background reaper task that uses DelayQueue to evict timed-out
+    // incomplete messages. O(1) tasks instead of O(N) per-message spawns.
+    // UnboundedSender is Clone, so the decoder can be cheaply cloned.
+    reaper_tx: tokio::sync::mpsc::UnboundedSender<u64>,
 }
 
 impl ChunkedGelfDecoder {
@@ -319,13 +332,46 @@ impl ChunkedGelfDecoder {
         max_length: Option<usize>,
         decompression_config: ChunkedGelfDecompressionConfig,
     ) -> Self {
+        let state: Arc<Mutex<HashMap<u64, MessageState>>> = Arc::new(Mutex::new(HashMap::new()));
+        let timeout = Duration::from_secs_f64(timeout_secs);
+
+        let (reaper_tx, mut reaper_rx) = tokio::sync::mpsc::unbounded_channel::<u64>();
+        let reaper_state = Arc::clone(&state);
+        tokio::spawn(async move {
+            use futures::StreamExt;
+            use tokio_util::time::DelayQueue;
+            let mut delay_queue: DelayQueue<u64> = DelayQueue::new();
+            loop {
+                tokio::select! {
+                    msg = reaper_rx.recv() => {
+                        match msg {
+                            Some(message_id) => { delay_queue.insert(message_id, timeout); }
+                            None => break,
+                        }
+                    }
+                    Some(expired) = delay_queue.next() => {
+                        let message_id = expired.into_inner();
+                        let mut state_lock = reaper_state.lock().expect("poisoned lock");
+                        if state_lock.remove(&message_id).is_some() {
+                            warn!(
+                                message_id = message_id,
+                                timeout_secs = timeout.as_secs_f64(),
+                                internal_log_rate_limit = true,
+                                "Message was not fully received within the timeout window. Discarding it."
+                            );
+                        }
+                    }
+                }
+            }
+        });
+
         Self {
             bytes_decoder: BytesDecoder::new(),
             decompression_config,
-            state: Arc::new(Mutex::new(HashMap::new())),
-            timeout: Duration::from_secs_f64(timeout_secs),
+            state,
             pending_messages_limit,
             max_length,
+            reaper_tx,
         }
     }
 
@@ -389,23 +435,8 @@ impl ChunkedGelfDecoder {
         }
 
         let message_state = state_lock.entry(message_id).or_insert_with(|| {
-            // We need to spawn a task that will clear the message state after a certain time
-            // otherwise we will have a memory leak due to messages that never complete
-            let state = Arc::clone(&self.state);
-            let timeout = self.timeout;
-            let timeout_handle = tokio::spawn(async move {
-                tokio::time::sleep(timeout).await;
-                let mut state_lock = state.lock().expect("poisoned lock");
-                if state_lock.remove(&message_id).is_some() {
-                    warn!(
-                        message_id = message_id,
-                        timeout_secs = timeout.as_secs_f64(),
-                        internal_log_rate_limit = true,
-                        "Message was not fully received within the timeout window. Discarding it."
-                    );
-                }
-            });
-            MessageState::new(total_chunks, timeout_handle)
+            let _ = self.reaper_tx.send(message_id);
+            MessageState::new(total_chunks)
         });
 
         ensure!(
@@ -486,8 +517,8 @@ impl Default for ChunkedGelfDecoder {
     fn default() -> Self {
         Self::new(
             DEFAULT_TIMEOUT_SECS,
-            None,
-            None,
+            Some(DEFAULT_PENDING_MESSAGES_LIMIT),
+            Some(DEFAULT_MAX_MESSAGE_LENGTH),
             ChunkedGelfDecompressionConfig::Auto,
         )
     }
@@ -1278,4 +1309,80 @@ mod tests {
 
         assert_eq!(detected_compression, ChunkedGelfDecompression::None);
     }
+
+    #[tokio::test]
+    async fn default_pending_messages_limit_is_finite() {
+        // The default decoder must enforce a pending-messages cap so an attacker
+        // cannot grow the HashMap unbounded by opening many message IDs.
+        let decoder = ChunkedGelfDecoder::default();
+        assert_eq!(decoder.pending_messages_limit, Some(DEFAULT_PENDING_MESSAGES_LIMIT));
+    }
+
+    #[tokio::test]
+    async fn default_max_length_is_finite() {
+        let decoder = ChunkedGelfDecoder::default();
+        assert_eq!(decoder.max_length, Some(DEFAULT_MAX_MESSAGE_LENGTH));
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[traced_test]
+    async fn reaper_evicts_multiple_incomplete_messages() {
+        // Verify the DelayQueue reaper (O(1) tasks) correctly evicts N concurrent
+        // incomplete messages — not just one.
+        let timeout_secs = 1.0_f64;
+        let mut decoder = ChunkedGelfDecoder::new(
+            timeout_secs,
+            None,
+            None,
+            ChunkedGelfDecompressionConfig::Auto,
+        );
+
+        // Open 5 different message IDs, each with 2 chunks, but only send chunk 0.
+        for msg_id in 1u64..=5 {
+            let mut chunk = create_chunk(msg_id, 0, 2, &b"partial");
+            let result = decoder.decode_eof(&mut chunk).unwrap();
+            assert!(result.is_none());
+        }
+        assert_eq!(decoder.state.lock().unwrap().len(), 5);
+
+        // Advance time past the timeout; reaper should clear all five entries.
+        tokio::time::sleep(Duration::from_secs_f64(timeout_secs + 0.5)).await;
+        assert!(
+            decoder.state.lock().unwrap().is_empty(),
+            "reaper must evict all incomplete messages"
+        );
+    }
+
+    #[rstest]
+    #[tokio::test]
+    async fn pending_messages_limit_rejects_excess_when_default(
+        two_chunks_message: ([BytesMut; 2], String),
+    ) {
+        // With pending_messages_limit = 1, a second in-flight message is rejected.
+        let (mut two_chunks, _) = two_chunks_message;
+        let second_msg_id = 99u64;
+        let mut extra_chunk = {
+            let mut c = BytesMut::new();
+            c.put_slice(GELF_MAGIC);
+            c.put_u64(second_msg_id);
+            c.put_u8(0u8);
+            c.put_u8(2u8);
+            c.extend_from_slice(b"x");
+            c
+        };
+        let mut decoder = ChunkedGelfDecoder {
+            pending_messages_limit: Some(1),
+            ..Default::default()
+        };
+
+        let frame = decoder.decode_eof(&mut two_chunks[0]).unwrap();
+        assert!(frame.is_none());
+
+        let err = decoder.decode_eof(&mut extra_chunk).unwrap_err();
+        let downcasted = downcast_framing_error(&err);
+        assert!(matches!(
+            downcasted,
+            ChunkedGelfDecoderError::PendingMessagesLimitReached { .. }
+        ));
+    }
 }

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

@@ -66,14 +66,18 @@ impl NewlineDelimitedDecoderConfig {
     }
 }
 
+/// Default maximum line length (100 KiB) applied when no explicit limit is configured.
+/// Guards against unbounded `BytesMut` growth from malformed or adversarial streams.
+pub const DEFAULT_MAX_LENGTH: usize = 100 * 1024;
+
 /// A codec for handling bytes that are delimited by (a) newline(s).
 #[derive(Debug, Clone)]
 pub struct NewlineDelimitedDecoder(CharacterDelimitedDecoder);
 
 impl NewlineDelimitedDecoder {
-    /// Creates a new `NewlineDelimitedDecoder`.
+    /// Creates a new `NewlineDelimitedDecoder` with the default 100 KiB max-line limit.
     pub const fn new() -> Self {
-        Self(CharacterDelimitedDecoder::new(b'\n'))
+        Self::new_with_max_length(DEFAULT_MAX_LENGTH)
     }
 
     /// Creates a `NewlineDelimitedDecoder` with a maximum frame length limit.
@@ -170,4 +174,17 @@ mod tests {
         assert_eq!(decoder.decode_eof(&mut input).unwrap().unwrap(), "baz");
         assert_eq!(decoder.decode_eof(&mut input).unwrap(), None);
     }
+
+    #[test]
+    fn new_enforces_default_max_length() {
+        // A line exactly at the limit passes; one byte over is discarded.
+        let at_limit = "a".repeat(DEFAULT_MAX_LENGTH);
+        let over_limit = "b".repeat(DEFAULT_MAX_LENGTH + 1);
+        let mut input = BytesMut::from(format!("{at_limit}\n{over_limit}\nok\n").as_str());
+        let mut decoder = NewlineDelimitedDecoder::new();
+
+        assert_eq!(decoder.decode(&mut input).unwrap().unwrap().len(), DEFAULT_MAX_LENGTH);
+        // Oversized line is silently discarded.
+        assert_eq!(decoder.decode(&mut input).unwrap().unwrap(), "ok");
+    }
 }