security(codecs): cap pending GELF messages

changelog.d/chunked_gelf_pending_messages_bounded.security.md

@@ -0,0 +1,5 @@
+The `chunked_gelf` framing decoder now applies a configurable limit to the number of incomplete messages held in memory. An unauthenticated sender could previously exhaust memory by sending unique message IDs it never completed, most easily on the `socket` source in UDP mode.
+
+`pending_messages_limit` now defaults to 4096. It was previously unset and therefore unbounded.
+
+authors: pront

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

@@ -22,10 +22,17 @@ const GELF_MAGIC: &[u8] = &[0x1e, 0x0f];
 const GELF_MAX_TOTAL_CHUNKS: u8 = 128;
 const DEFAULT_TIMEOUT_SECS: f64 = 5.0;
 
+/// Most messages that may await completion at once.
+const MAX_PENDING_MESSAGES: usize = 4096;
+
 const fn default_timeout_secs() -> f64 {
     DEFAULT_TIMEOUT_SECS
 }
 
+const fn default_pending_messages_limit() -> usize {
+    MAX_PENDING_MESSAGES
+}
+
 /// Config used to build a `ChunkedGelfDecoder`.
 #[configurable_component]
 #[derive(Debug, Clone, Default)]
@@ -60,8 +67,11 @@ 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.
+    ///
+    /// Chunks belonging to messages that are already pending are still accepted once the limit is
+    /// reached, so in-flight messages can complete.
+    ///
+    /// If unset or `null`, this defaults to 4096.
     #[serde(default, skip_serializing_if = "vector_core::serde::is_default")]
     pub pending_messages_limit: Option<usize>,
 
@@ -300,7 +310,7 @@ pub struct ChunkedGelfDecoder {
     decompression_config: ChunkedGelfDecompressionConfig,
     state: Arc<Mutex<HashMap<u64, Box<MessageState>>>>,
     timeout: Duration,
-    pending_messages_limit: Option<usize>,
+    pending_messages_limit: usize,
     max_length: Option<usize>,
 }
 
@@ -317,7 +327,8 @@ impl ChunkedGelfDecoder {
             decompression_config,
             state: Arc::new(Mutex::new(HashMap::new())),
             timeout: Duration::from_secs_f64(timeout_secs),
-            pending_messages_limit,
+            pending_messages_limit: pending_messages_limit
+                .unwrap_or_else(default_pending_messages_limit),
             max_length,
         }
     }
@@ -401,15 +412,13 @@ impl ChunkedGelfDecoder {
         // Only a new message grows the table, so the limit applies on insert. Checking it
         // before the lookup rejected chunks of messages already pending, which could then
         // never complete and expired instead.
-        if !state_lock.contains_key(&message_id)
-            && let Some(pending_messages_limit) = self.pending_messages_limit
-        {
+        if !state_lock.contains_key(&message_id) {
             ensure!(
-                state_lock.len() < pending_messages_limit,
+                state_lock.len() < self.pending_messages_limit,
                 PendingMessagesLimitReachedSnafu {
                     message_id,
                     sequence_number,
-                    pending_messages_limit
+                    pending_messages_limit: self.pending_messages_limit
                 }
             );
         }
@@ -969,6 +978,36 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn pending_messages_limit_defaults_and_allows_overrides() {
+        let default = ChunkedGelfDecoder::default();
+        assert_eq!(default.pending_messages_limit, MAX_PENDING_MESSAGES);
+
+        let explicit_null: ChunkedGelfDecoderOptions =
+            serde_json::from_value(serde_json::json!({ "pending_messages_limit": null })).unwrap();
+        let decoder = ChunkedGelfDecoderConfig {
+            chunked_gelf: explicit_null,
+        }
+        .build();
+        assert_eq!(decoder.pending_messages_limit, MAX_PENDING_MESSAGES);
+
+        let raised = ChunkedGelfDecoder::new(
+            DEFAULT_TIMEOUT_SECS,
+            Some(MAX_PENDING_MESSAGES + 1),
+            None,
+            ChunkedGelfDecompressionConfig::Auto,
+        );
+        assert_eq!(raised.pending_messages_limit, MAX_PENDING_MESSAGES + 1);
+
+        let lowered = ChunkedGelfDecoder::new(
+            DEFAULT_TIMEOUT_SECS,
+            Some(1),
+            None,
+            ChunkedGelfDecompressionConfig::Auto,
+        );
+        assert_eq!(lowered.pending_messages_limit, 1);
+    }
+
     #[rstest]
     #[tokio::test]
     async fn decode_reached_pending_messages_limit(
@@ -978,7 +1017,7 @@ mod tests {
         let (mut two_chunks, _) = two_chunks_message;
         let (mut three_chunks, _) = three_chunks_message;
         let mut decoder = ChunkedGelfDecoder {
-            pending_messages_limit: Some(1),
+            pending_messages_limit: 1,
             ..Default::default()
         };
 
@@ -1010,7 +1049,7 @@ mod tests {
         let (mut two_chunks, two_chunks_expected) = two_chunks_message;
         let (mut three_chunks, _) = three_chunks_message;
         let mut decoder = ChunkedGelfDecoder {
-            pending_messages_limit: Some(1),
+            pending_messages_limit: 1,
             ..Default::default()
         };
 

website/cue/reference/components/generated/schema_definitions.cue

@@ -417,8 +417,11 @@ _schemaDefinitions: {
 					description: """
 						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.
+
+						Chunks belonging to messages that are already pending are still accepted once the limit is
+						reached, so in-flight messages can complete.
+
+						If unset or `null`, this defaults to 4096.
 						"""
 					required: false
 					type: uint: {}
@@ -714,8 +717,11 @@ _schemaDefinitions: {
 			description: """
 				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.
+
+				Chunks belonging to messages that are already pending are still accepted once the limit is
+				reached, so in-flight messages can complete.
+
+				If unset or `null`, this defaults to 4096.
 				"""
 			required: false
 			type: uint: {}
@@ -3509,8 +3515,11 @@ _schemaDefinitions: {
 							description: """
 															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.
+
+															Chunks belonging to messages that are already pending are still accepted once the limit is
+															reached, so in-flight messages can complete.
+
+															If unset or `null`, this defaults to 4096.
 															"""
 							required: false
 							type: uint: {}