diff --git a/changelog.d/chunked_gelf_pending_messages_bounded.security.md b/changelog.d/chunked_gelf_pending_messages_bounded.security.md new file mode 100644 index 0000000000000..9d20a54a61d82 --- /dev/null +++ b/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 diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 60d8f04d8f8e8..5d535d211a07c 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/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, @@ -300,7 +310,7 @@ pub struct ChunkedGelfDecoder { decompression_config: ChunkedGelfDecompressionConfig, state: Arc>>>, timeout: Duration, - pending_messages_limit: Option, + pending_messages_limit: usize, max_length: Option, } @@ -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() }; diff --git a/website/cue/reference/components/generated/schema_definitions.cue b/website/cue/reference/components/generated/schema_definitions.cue index 158815a4a2cb1..33fc61f3ba4b0 100644 --- a/website/cue/reference/components/generated/schema_definitions.cue +++ b/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: {}