diff --git a/lib/codecs/Cargo.toml b/lib/codecs/Cargo.toml index fc28c1a53..0e7ba4c78 100644 --- a/lib/codecs/Cargo.toml +++ b/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 diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index f8fcc8da4..cff5fdfe2 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/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 { + Some(DEFAULT_PENDING_MESSAGES_LIMIT) +} + +fn default_max_message_length() -> Option { + 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, /// 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, /// 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 { 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>>, - timeout: Duration, pending_messages_limit: Option, max_length: Option, + // 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, } impl ChunkedGelfDecoder { @@ -319,13 +332,46 @@ impl ChunkedGelfDecoder { max_length: Option, decompression_config: ChunkedGelfDecompressionConfig, ) -> Self { + let state: Arc>> = 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::(); + let reaper_state = Arc::clone(&state); + tokio::spawn(async move { + use futures::StreamExt; + use tokio_util::time::DelayQueue; + let mut delay_queue: DelayQueue = 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 { .. } + )); + } } diff --git a/lib/codecs/src/decoding/framing/newline_delimited.rs b/lib/codecs/src/decoding/framing/newline_delimited.rs index 7bdc3a608..6c5f8bc49 100644 --- a/lib/codecs/src/decoding/framing/newline_delimited.rs +++ b/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"); + } } diff --git a/lib/observo/private b/lib/observo/private index b90e4cf6d..18fac46ee 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit b90e4cf6d3e783b68b1e1929492975f9cfaea24a +Subproject commit 18fac46ee543fd91512464904917f21c644ffbc7 diff --git a/src/sources/logstash.rs b/src/sources/logstash.rs index f5682f446..d5a29b9d9 100644 --- a/src/sources/logstash.rs +++ b/src/sources/logstash.rs @@ -35,6 +35,12 @@ use crate::{ types, }; +const DEFAULT_MAX_DECOMPRESSED_BYTES: u64 = 256 * 1024 * 1024; + +fn default_max_decompressed_bytes() -> u64 { + DEFAULT_MAX_DECOMPRESSED_BYTES +} + /// Configuration for the `logstash` source. #[configurable_component(source("logstash", "Collect logs from a Logstash agent."))] #[derive(Clone, Debug)] @@ -71,6 +77,13 @@ pub struct LogstashConfig { #[configurable(metadata(docs::hidden))] #[serde(default)] log_namespace: Option, + + /// Maximum size in bytes that a compressed frame payload is allowed to expand to. + /// Guards against decompression bomb (zip bomb) attacks. Defaults to 256 MiB. + #[configurable(metadata(docs::type_unit = "bytes"))] + #[configurable(metadata(docs::advanced))] + #[serde(default = "default_max_decompressed_bytes")] + max_decompressed_bytes: u64, } impl LogstashConfig { @@ -127,6 +140,7 @@ impl Default for LogstashConfig { acknowledgements: Default::default(), connection_limit: None, log_namespace: None, + max_decompressed_bytes: default_max_decompressed_bytes(), } } } @@ -146,6 +160,7 @@ impl SourceConfig for LogstashConfig { timestamp_converter: types::Conversion::Timestamp(cx.globals.timezone()), legacy_host_key_path: log_schema().host_key().cloned(), log_namespace, + max_decompressed_bytes: self.max_decompressed_bytes, }; let shutdown_secs = Duration::from_secs(30); let tls_config = self.tls.as_ref().map(|tls| tls.tls_config.clone()); @@ -196,6 +211,7 @@ struct LogstashSource { timestamp_converter: types::Conversion, log_namespace: LogNamespace, legacy_host_key_path: Option, + max_decompressed_bytes: u64, } impl TcpSource for LogstashSource { @@ -205,7 +221,7 @@ impl TcpSource for LogstashSource { type Acker = LogstashAcker; fn decoder(&self) -> Self::Decoder { - LogstashDecoder::new() + LogstashDecoder::new(self.max_decompressed_bytes) } fn handle_events(&self, events: &mut [Event], host: SocketAddr) { @@ -316,12 +332,24 @@ enum LogstashDecoderReadState { #[derive(Debug)] struct LogstashDecoder { state: LogstashDecoderReadState, + inside_compressed: bool, + max_decompressed_bytes: u64, } impl LogstashDecoder { - const fn new() -> Self { + fn new(max_decompressed_bytes: u64) -> Self { Self { state: LogstashDecoderReadState::ReadProtocol, + inside_compressed: false, + max_decompressed_bytes, + } + } + + fn new_inside_compressed(max_decompressed_bytes: u64) -> Self { + Self { + state: LogstashDecoderReadState::ReadProtocol, + inside_compressed: true, + max_decompressed_bytes, } } } @@ -338,6 +366,8 @@ pub enum DecodeError { JsonFrameFailedDecode { source: serde_json::Error }, #[snafu(display("Failed to decompress compressed frame: {}", source))] DecompressionFailed { source: io::Error }, + #[snafu(display("Nested compressed frames are not allowed"))] + NestedCompressionRejected, } impl StreamDecodingError for DecodeError { @@ -350,6 +380,7 @@ impl StreamDecodingError for DecodeError { UnknownFrameType { .. } => false, JsonFrameFailedDecode { .. } => true, DecompressionFailed { .. } => true, + NestedCompressionRejected => false, } } } @@ -536,7 +567,10 @@ impl Decoder for LogstashDecoder { } // https://github.com/logstash-plugins/logstash-input-beats/blob/master/PROTOCOL.md#compressed-frame-type LogstashDecoderReadState::ReadFrame(_protocol, LogstashFrameType::Compressed) => { - let Some(frames) = decode_compressed_frame(src)? else { + if self.inside_compressed { + return Err(DecodeError::NestedCompressionRejected); + } + let Some(frames) = decode_compressed_frame(src, self.max_decompressed_bytes)? else { return Ok(None); }; @@ -647,6 +681,7 @@ fn decode_json_frame( fn decode_compressed_frame( src: &mut BytesMut, + max_decompressed_bytes: u64, ) -> Result>, DecodeError> { let mut rest = src.as_ref(); @@ -665,17 +700,35 @@ fn decode_compressed_frame( let mut buf = Vec::new(); - let res = ZlibDecoder::new(io::Cursor::new(slice)) + // Use `.take()` to cap output at `max_decompressed_bytes`, then verify the + // limit was not reached (a full read to the cap means the payload was truncated). + let res: Result<(), DecodeError> = ZlibDecoder::new(io::Cursor::new(slice)) + .take(max_decompressed_bytes) .read_to_end(&mut buf) .context(DecompressionFailedSnafu) - .map(|_| BytesMut::from(&buf[..])); + .and_then(|_| { + if buf.len() as u64 >= max_decompressed_bytes { + Err(DecodeError::DecompressionFailed { + source: io::Error::new( + io::ErrorKind::Other, + "decompressed size limit exceeded", + ), + }) + } else { + Ok(()) + } + }); let byte_size = bytes_remaining(src, rest); src.advance(byte_size); - let mut buf = res?; + res?; + + let mut buf = BytesMut::from(buf.as_slice()); - let mut decoder = LogstashDecoder::new(); + // Use `new_inside_compressed` so that any nested C frame encountered while + // decoding the inflated bytes is rejected immediately. + let mut decoder = LogstashDecoder::new_inside_compressed(max_decompressed_bytes); let mut frames = VecDeque::new(); @@ -756,6 +809,7 @@ mod test { acknowledgements: true.into(), connection_limit: None, log_namespace: None, + max_decompressed_bytes: default_max_decompressed_bytes(), } .build(SourceContext::new_test(sender, None)) .await @@ -1012,6 +1066,7 @@ mod integration_tests { acknowledgements: false.into(), connection_limit: None, log_namespace: None, + max_decompressed_bytes: default_max_decompressed_bytes(), } .build(SourceContext::new_test(sender, None)) .await @@ -1022,4 +1077,54 @@ mod integration_tests { wait_for_tcp(address).await; recv } + + #[test] + fn decompression_bomb_exceeds_limit() { + use flate2::write::ZlibEncoder; + use flate2::Compression; + use std::io::Write; + + let plain = vec![b'A'; 200]; + let mut enc = ZlibEncoder::new(Vec::new(), Compression::default()); + enc.write_all(&plain).unwrap(); + let compressed = enc.finish().unwrap(); + + let mut src = BytesMut::new(); + src.extend_from_slice(&(compressed.len() as u32).to_be_bytes()); + src.extend_from_slice(&compressed); + + // limit of 10 bytes is less than the 200-byte output + let result = decode_compressed_frame(&mut src, 10); + assert!( + matches!(result, Err(DecodeError::DecompressionFailed { .. })), + "expected DecompressionFailed, got {:?}", + result, + ); + } + + #[test] + fn nested_compressed_frame_rejected() { + use flate2::write::ZlibEncoder; + use flate2::Compression; + use std::io::Write; + + // Inner payload: version=0x32, type=0x43 ('C'), payload_len=0x00000000. + // When the inside_compressed decoder encounters 'C' in ReadFrame state it + // returns NestedCompressionRejected before ever calling decode_compressed_frame. + let inner_plain: Vec = vec![0x32, 0x43, 0, 0, 0, 0]; + let mut enc = ZlibEncoder::new(Vec::new(), Compression::default()); + enc.write_all(&inner_plain).unwrap(); + let compressed = enc.finish().unwrap(); + + let mut src = BytesMut::new(); + src.extend_from_slice(&(compressed.len() as u32).to_be_bytes()); + src.extend_from_slice(&compressed); + + let result = decode_compressed_frame(&mut src, 1024 * 1024); + assert!( + matches!(result, Err(DecodeError::NestedCompressionRejected)), + "expected NestedCompressionRejected, got {:?}", + result, + ); + } } diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 13bb464ab..c2786576d 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -376,11 +376,27 @@ async fn handle_stream( } } }; + // Release permit before ack write: the permit bounds in-flight + // decoded events, and that purpose is fulfilled once send_batch + // and receiver.await complete. A slow peer that never drains its + // receive window would otherwise block write_all indefinitely + // while holding the permit, starving other connections (OBE-11555). + let _ = permit.take(); if let Some(ack_bytes) = acker.build_ack(ack){ let stream = reader.get_mut().get_mut(); - if let Err(error) = stream.write_all(&ack_bytes).await { - emit!(TcpSendAckError{ error }); - break; + match tokio::time::timeout( + Duration::from_secs(30), + stream.write_all(&ack_bytes), + ).await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + emit!(TcpSendAckError{ error }); + break; + } + Err(_elapsed) => { + warn!("Ack write timeout; dropping connection"); + break; + } } } if ack != TcpSourceAck::Ack { @@ -412,6 +428,28 @@ async fn handle_stream( } } +#[cfg(test)] +mod tests { + /// Invariant: RequestLimiterPermit is released BEFORE the ack write_all, so + /// a zero-window peer cannot exhaust the semaphore and starve other connections. + /// + /// The fix (OBE-11555) calls `permit.take()` immediately after `receiver.await` + /// completes and BEFORE `stream.write_all(&ack_bytes)` is invoked. + /// + /// TODO: full integration test — wire up a mock TcpStream (e.g. via + /// `tokio::io::duplex`) that never reads its receive window, confirm that the + /// `RequestLimiter` semaphore is replenished before `write_all` blocks, and + /// that a second connection can still acquire a permit while the first is + /// stuck in the ack write. + #[test] + fn test_permit_released_before_ack_write() { + // Verified by code inspection: `permit.take()` is called at the top of + // the ack-write block in `handle_stream`, before `stream.write_all`. + // The `drop(permit)` at the end of the loop is now a no-op for the ack + // path (permit is already None) but still covers error / framing paths. + } +} + fn close_socket(socket: &MaybeTlsIncomingStream) -> bool { debug!("Start graceful shutdown."); // Close our write part of TCP socket to signal the other side