From e4fbd6ee388af50bed7b0e8044ec392ae6e48c6e Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Thu, 3 Sep 2026 12:57:15 -0400 Subject: [PATCH 1/4] security(codecs): cap pending GELF messages --- ..._gelf_pending_messages_bounded.security.md | 5 ++ .../src/decoding/framing/chunked_gelf.rs | 50 +++++++++++++++---- 2 files changed, 44 insertions(+), 11 deletions(-) create mode 100644 changelog.d/chunked_gelf_pending_messages_bounded.security.md 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..cfb066fddd157 --- /dev/null +++ b/changelog.d/chunked_gelf_pending_messages_bounded.security.md @@ -0,0 +1,5 @@ +The `chunked_gelf` framing decoder now limits incomplete messages to 4096 at a time. 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` can lower this ceiling but can no longer raise it. + +authors: pront diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 60d8f04d8f8e8..761e30c774d9a 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -22,6 +22,9 @@ 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 } @@ -60,8 +63,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. + /// + /// **Note**: The decoder caps this at 4096 internally. Higher values do not raise it. #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] pub pending_messages_limit: Option, @@ -300,7 +306,7 @@ pub struct ChunkedGelfDecoder { decompression_config: ChunkedGelfDecompressionConfig, state: Arc>>>, timeout: Duration, - pending_messages_limit: Option, + pending_messages_limit: usize, max_length: Option, } @@ -317,7 +323,9 @@ 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(MAX_PENDING_MESSAGES) + .min(MAX_PENDING_MESSAGES), max_length, } } @@ -401,15 +409,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 +975,28 @@ mod tests { )); } + #[test] + fn pending_messages_limit_is_clamped_to_the_hard_ceiling() { + let default = ChunkedGelfDecoder::default(); + assert_eq!(default.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); + + 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 +1006,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 +1038,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() }; From a52c8060d2038419d82382607f88549b1b1860f6 Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Tue, 8 Sep 2026 13:45:32 -0400 Subject: [PATCH 2/4] fix(codecs): allow overriding GELF pending message limit --- .../chunked_gelf_pending_messages_bounded.security.md | 2 +- lib/codecs/src/decoding/framing/chunked_gelf.rs | 10 +++------- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/changelog.d/chunked_gelf_pending_messages_bounded.security.md b/changelog.d/chunked_gelf_pending_messages_bounded.security.md index cfb066fddd157..d26f232aa4d0f 100644 --- a/changelog.d/chunked_gelf_pending_messages_bounded.security.md +++ b/changelog.d/chunked_gelf_pending_messages_bounded.security.md @@ -1,5 +1,5 @@ The `chunked_gelf` framing decoder now limits incomplete messages to 4096 at a time. 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` can lower this ceiling but can no longer raise it. +`pending_messages_limit` defaults to 4096 and can be set higher or lower. authors: pront diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 761e30c774d9a..11319eb98c93b 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -66,8 +66,6 @@ pub struct ChunkedGelfDecoderOptions { /// /// Chunks belonging to messages that are already pending are still accepted once the limit is /// reached, so in-flight messages can complete. - /// - /// **Note**: The decoder caps this at 4096 internally. Higher values do not raise it. #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] pub pending_messages_limit: Option, @@ -323,9 +321,7 @@ impl ChunkedGelfDecoder { decompression_config, state: Arc::new(Mutex::new(HashMap::new())), timeout: Duration::from_secs_f64(timeout_secs), - pending_messages_limit: pending_messages_limit - .unwrap_or(MAX_PENDING_MESSAGES) - .min(MAX_PENDING_MESSAGES), + pending_messages_limit: pending_messages_limit.unwrap_or(MAX_PENDING_MESSAGES), max_length, } } @@ -976,7 +972,7 @@ mod tests { } #[test] - fn pending_messages_limit_is_clamped_to_the_hard_ceiling() { + fn pending_messages_limit_defaults_and_allows_overrides() { let default = ChunkedGelfDecoder::default(); assert_eq!(default.pending_messages_limit, MAX_PENDING_MESSAGES); @@ -986,7 +982,7 @@ mod tests { None, ChunkedGelfDecompressionConfig::Auto, ); - assert_eq!(raised.pending_messages_limit, MAX_PENDING_MESSAGES); + assert_eq!(raised.pending_messages_limit, MAX_PENDING_MESSAGES + 1); let lowered = ChunkedGelfDecoder::new( DEFAULT_TIMEOUT_SECS, From 279a767c1649a4c750993954c1a0d577064bff2a Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Tue, 8 Sep 2026 14:29:00 -0400 Subject: [PATCH 3/4] fix(codecs): expose GELF pending message default --- ..._gelf_pending_messages_bounded.security.md | 2 +- .../src/decoding/framing/chunked_gelf.rs | 19 ++++++++++------- .../generated/schema_definitions.cue | 21 +++++++++++-------- 3 files changed, 25 insertions(+), 17 deletions(-) diff --git a/changelog.d/chunked_gelf_pending_messages_bounded.security.md b/changelog.d/chunked_gelf_pending_messages_bounded.security.md index d26f232aa4d0f..68cae430b3670 100644 --- a/changelog.d/chunked_gelf_pending_messages_bounded.security.md +++ b/changelog.d/chunked_gelf_pending_messages_bounded.security.md @@ -1,5 +1,5 @@ The `chunked_gelf` framing decoder now limits incomplete messages to 4096 at a time. 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` defaults to 4096 and can be set higher or lower. +`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 11319eb98c93b..7d8ce442f110a 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -29,6 +29,10 @@ 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)] @@ -66,8 +70,9 @@ pub struct ChunkedGelfDecoderOptions { /// /// Chunks belonging to messages that are already pending are still accepted once the limit is /// reached, so in-flight messages can complete. - #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] - pub pending_messages_limit: Option, + #[serde(default = "default_pending_messages_limit")] + #[derivative(Default(value = "default_pending_messages_limit()"))] + pub pending_messages_limit: usize, /// The maximum length of a single GELF message, in bytes. Messages longer than this length are /// dropped. If this option is not set, the decoder does not limit the length of messages and @@ -312,7 +317,7 @@ impl ChunkedGelfDecoder { /// Creates a new `ChunkedGelfDecoder`. pub fn new( timeout_secs: f64, - pending_messages_limit: Option, + pending_messages_limit: usize, max_length: Option, decompression_config: ChunkedGelfDecompressionConfig, ) -> Self { @@ -321,7 +326,7 @@ impl ChunkedGelfDecoder { decompression_config, state: Arc::new(Mutex::new(HashMap::new())), timeout: Duration::from_secs_f64(timeout_secs), - pending_messages_limit: pending_messages_limit.unwrap_or(MAX_PENDING_MESSAGES), + pending_messages_limit, max_length, } } @@ -544,7 +549,7 @@ impl Default for ChunkedGelfDecoder { fn default() -> Self { Self::new( DEFAULT_TIMEOUT_SECS, - None, + default_pending_messages_limit(), None, ChunkedGelfDecompressionConfig::Auto, ) @@ -978,7 +983,7 @@ mod tests { let raised = ChunkedGelfDecoder::new( DEFAULT_TIMEOUT_SECS, - Some(MAX_PENDING_MESSAGES + 1), + MAX_PENDING_MESSAGES + 1, None, ChunkedGelfDecompressionConfig::Auto, ); @@ -986,7 +991,7 @@ mod tests { let lowered = ChunkedGelfDecoder::new( DEFAULT_TIMEOUT_SECS, - Some(1), + 1, None, ChunkedGelfDecompressionConfig::Auto, ); diff --git a/website/cue/reference/components/generated/schema_definitions.cue b/website/cue/reference/components/generated/schema_definitions.cue index 158815a4a2cb1..624b410908da2 100644 --- a/website/cue/reference/components/generated/schema_definitions.cue +++ b/website/cue/reference/components/generated/schema_definitions.cue @@ -417,11 +417,12 @@ _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. """ required: false - type: uint: {} + type: uint: default: 4096 } timeout_secs: { description: """ @@ -714,11 +715,12 @@ _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. """ required: false - type: uint: {} + type: uint: default: 4096 } timeout_secs: { description: """ @@ -3509,11 +3511,12 @@ _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. """ required: false - type: uint: {} + type: uint: default: 4096 } timeout_secs: { description: """ From 899ece8cc3a8797dc16f176beb2a85dfedb98e3e Mon Sep 17 00:00:00 2001 From: Pavlos Rontidis Date: Tue, 8 Sep 2026 15:37:14 -0400 Subject: [PATCH 4/4] fix(codecs): preserve nullable GELF limit --- ..._gelf_pending_messages_bounded.security.md | 2 +- .../src/decoding/framing/chunked_gelf.rs | 26 +++++++++++++------ .../generated/schema_definitions.cue | 12 ++++++--- 3 files changed, 28 insertions(+), 12 deletions(-) diff --git a/changelog.d/chunked_gelf_pending_messages_bounded.security.md b/changelog.d/chunked_gelf_pending_messages_bounded.security.md index 68cae430b3670..9d20a54a61d82 100644 --- a/changelog.d/chunked_gelf_pending_messages_bounded.security.md +++ b/changelog.d/chunked_gelf_pending_messages_bounded.security.md @@ -1,4 +1,4 @@ -The `chunked_gelf` framing decoder now limits incomplete messages to 4096 at a time. An unauthenticated sender could previously exhaust memory by sending unique message IDs it never completed, most easily on the `socket` source in UDP mode. +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. diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index 7d8ce442f110a..5d535d211a07c 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -70,9 +70,10 @@ pub struct ChunkedGelfDecoderOptions { /// /// Chunks belonging to messages that are already pending are still accepted once the limit is /// reached, so in-flight messages can complete. - #[serde(default = "default_pending_messages_limit")] - #[derivative(Default(value = "default_pending_messages_limit()"))] - pub pending_messages_limit: usize, + /// + /// If unset or `null`, this defaults to 4096. + #[serde(default, skip_serializing_if = "vector_core::serde::is_default")] + pub pending_messages_limit: Option, /// The maximum length of a single GELF message, in bytes. Messages longer than this length are /// dropped. If this option is not set, the decoder does not limit the length of messages and @@ -317,7 +318,7 @@ impl ChunkedGelfDecoder { /// Creates a new `ChunkedGelfDecoder`. pub fn new( timeout_secs: f64, - pending_messages_limit: usize, + pending_messages_limit: Option, max_length: Option, decompression_config: ChunkedGelfDecompressionConfig, ) -> Self { @@ -326,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, } } @@ -549,7 +551,7 @@ impl Default for ChunkedGelfDecoder { fn default() -> Self { Self::new( DEFAULT_TIMEOUT_SECS, - default_pending_messages_limit(), + None, None, ChunkedGelfDecompressionConfig::Auto, ) @@ -981,9 +983,17 @@ mod tests { 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, - MAX_PENDING_MESSAGES + 1, + Some(MAX_PENDING_MESSAGES + 1), None, ChunkedGelfDecompressionConfig::Auto, ); @@ -991,7 +1001,7 @@ mod tests { let lowered = ChunkedGelfDecoder::new( DEFAULT_TIMEOUT_SECS, - 1, + Some(1), None, ChunkedGelfDecompressionConfig::Auto, ); diff --git a/website/cue/reference/components/generated/schema_definitions.cue b/website/cue/reference/components/generated/schema_definitions.cue index 624b410908da2..33fc61f3ba4b0 100644 --- a/website/cue/reference/components/generated/schema_definitions.cue +++ b/website/cue/reference/components/generated/schema_definitions.cue @@ -420,9 +420,11 @@ _schemaDefinitions: { 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: default: 4096 + type: uint: {} } timeout_secs: { description: """ @@ -718,9 +720,11 @@ _schemaDefinitions: { 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: default: 4096 + type: uint: {} } timeout_secs: { description: """ @@ -3514,9 +3518,11 @@ _schemaDefinitions: { 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: default: 4096 + type: uint: {} } timeout_secs: { description: """