diff --git a/changelog.d/security_oom_allocation_bounds.enhancement.md b/changelog.d/security_oom_allocation_bounds.enhancement.md new file mode 100644 index 000000000..500353980 --- /dev/null +++ b/changelog.d/security_oom_allocation_bounds.enhancement.md @@ -0,0 +1,16 @@ +Added default upper bounds to previously-unbounded allocation paths in several sources, so a +malicious or malformed peer can no longer exhaust the heap. Every default is set above documented +producer maxima, so legitimate traffic is unaffected; each is overridable. + +- `logstash`: new `max_decompressed_bytes` (256 MiB) caps compressed-frame inflation; nested + compressed frames are rejected. +- `gcp_gcs`: new `max_decompressed_bytes` (4 GiB); truncation is logged and counted by + `gcs_object_truncated_total`. +- `stcp`: new `max_frame_bytes` (4x `max_event_size`, 64 MiB) and `max_lines_per_event` (1e6). +- `wef`: `max_content_length` is now enforced on the inbound HTTP body, defaulting to 4x the + advertised `max_envelope_size` and never dropping below it. +- GELF chunked framing: `pending_messages_limit` 10000, `max_length` 8 MiB — both above the + protocol's own ceiling of 128 chunks per message. + +The `tcp` source now releases its `RequestLimiter` permit before writing the acknowledgement and +bounds that write with a 30-second timeout, so a peer that stops reading cannot starve others. diff --git a/lib/codecs/src/decoding/framing/chunked_gelf.rs b/lib/codecs/src/decoding/framing/chunked_gelf.rs index f8fcc8da4..516617619 100644 --- a/lib/codecs/src/decoding/framing/chunked_gelf.rs +++ b/lib/codecs/src/decoding/framing/chunked_gelf.rs @@ -19,11 +19,28 @@ 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; +/// Cap on concurrent incomplete messages, bounding the reassembly map. +/// Graylog Server itself has no such cap, so this is sized well above what a +/// legitimate sender holds in flight within the 5s reassembly window. +pub const DEFAULT_PENDING_MESSAGES_LIMIT: usize = 10_000; +/// Cap on one reassembled message. The protocol ceiling is 128 chunks +/// (`GELF_MAX_TOTAL_CHUNKS`) times the 65507-byte max UDP payload, so 8 MiB is +/// above anything the wire format can produce. Matches Graylog's own +/// `decompress_size_limit` default. +pub const DEFAULT_MAX_MESSAGE_LENGTH: usize = 8 * 1024 * 1024; const fn default_timeout_secs() -> f64 { DEFAULT_TIMEOUT_SECS } +const fn default_pending_messages_limit() -> Option { + Some(DEFAULT_PENDING_MESSAGES_LIMIT) +} + +const 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 +75,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 10000. Set explicitly to raise or lower it. + #[serde(default = "default_pending_messages_limit")] + #[derivative(Default(value = "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 8 MiB, which is above the protocol's own ceiling of 128 chunks per + /// message. /// /// 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 = "default_max_message_length()"))] pub max_length: Option, /// Decompression configuration for GELF messages. @@ -486,8 +504,8 @@ impl Default for ChunkedGelfDecoder { fn default() -> Self { Self::new( DEFAULT_TIMEOUT_SECS, - None, - None, + default_pending_messages_limit(), + default_max_message_length(), ChunkedGelfDecompressionConfig::Auto, ) } @@ -1278,4 +1296,38 @@ mod tests { assert_eq!(detected_compression, ChunkedGelfDecompression::None); } + + #[tokio::test] + async fn defaults_are_finite_and_above_the_protocol_ceiling() { + let options = ChunkedGelfDecoderOptions::default(); + assert_eq!( + options.pending_messages_limit, + Some(DEFAULT_PENDING_MESSAGES_LIMIT) + ); + assert_eq!(options.max_length, Some(DEFAULT_MAX_MESSAGE_LENGTH)); + + // 128 chunks x the 65507-byte max UDP payload is the most the wire format can carry, + // so the default can never reject a well-formed message. + let protocol_ceiling = GELF_MAX_TOTAL_CHUNKS as usize * 65_507; + assert!(DEFAULT_MAX_MESSAGE_LENGTH >= protocol_ceiling); + } + + #[tokio::test] + async fn limits_are_per_message_and_do_not_kill_the_stream() { + // Both are per-message conditions; tearing down the connection would let one bad sender + // drop every other message multiplexed over it. + assert!(ChunkedGelfDecoderError::MaxLengthExceed { + message_id: 1, + sequence_number: 0, + length: 10, + max_length: 5, + } + .can_continue()); + assert!(ChunkedGelfDecoderError::PendingMessagesLimitReached { + message_id: 1, + sequence_number: 0, + pending_messages_limit: 1, + } + .can_continue()); + } } diff --git a/lib/observo/private b/lib/observo/private index b90e4cf6d..c377dffb5 160000 --- a/lib/observo/private +++ b/lib/observo/private @@ -1 +1 @@ -Subproject commit b90e4cf6d3e783b68b1e1929492975f9cfaea24a +Subproject commit c377dffb5586308c3387e5337474b2ca01091591 diff --git a/src/sources/logstash.rs b/src/sources/logstash.rs index f5682f446..e81b5498e 100644 --- a/src/sources/logstash.rs +++ b/src/sources/logstash.rs @@ -35,6 +35,21 @@ use crate::{ types, }; +/// Cap on the inflated size of a single compressed frame. +/// +/// One Beats `C` frame carries an entire window, so its inflated size scales with the sender's +/// batch size (`bulk_max_size` defaults to 2048 events; go-lumber's `maxWindowSize` allows 10000) +/// times the per-event size. 256 MiB sits above any such batch, so the bound only ever trips on a +/// decompression bomb. +/// +/// The bound is per frame, so peak memory is this value times the concurrent connection count. +/// Set a finite `connection_limit` if that product matters for your deployment. +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 +86,16 @@ 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. + /// + /// This bound applies per frame, so peak memory scales with the number of concurrent + /// connections. Raise it only alongside a finite `connection_limit`. + #[configurable(metadata(docs::type_unit = "bytes"))] + #[configurable(metadata(docs::advanced))] + #[serde(default = "default_max_decompressed_bytes")] + max_decompressed_bytes: u64, } impl LogstashConfig { @@ -127,6 +152,7 @@ impl Default for LogstashConfig { acknowledgements: Default::default(), connection_limit: None, log_namespace: None, + max_decompressed_bytes: default_max_decompressed_bytes(), } } } @@ -146,6 +172,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 +223,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 +233,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 +344,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 +378,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 +392,7 @@ impl StreamDecodingError for DecodeError { UnknownFrameType { .. } => false, JsonFrameFailedDecode { .. } => true, DecompressionFailed { .. } => true, + NestedCompressionRejected => false, } } } @@ -536,7 +579,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 +693,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 +712,39 @@ fn decode_compressed_frame( let mut buf = Vec::new(); - let res = ZlibDecoder::new(io::Cursor::new(slice)) + // Cap output with `.take()` so a decompression bomb can never allocate without bound. Reading + // up to `max + 1` bytes lets us distinguish "exactly at the limit" (legal) from "truncated at + // the limit" (rejected) — capping at `max` alone makes those two cases indistinguishable and + // would reject a payload that is exactly `max_decompressed_bytes` long. + let res: Result<(), DecodeError> = ZlibDecoder::new(io::Cursor::new(slice)) + .take(max_decompressed_bytes.saturating_add(1)) .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, + format!( + "decompressed size limit of {max_decompressed_bytes} bytes 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(); @@ -732,6 +801,145 @@ mod test { crate::test_util::test_generate_config::(); } + /// Wraps `payload` in the length-prefixed envelope `decode_compressed_frame` expects. + fn zlib_frame(payload: &[u8]) -> BytesMut { + use flate2::write::ZlibEncoder; + use flate2::Compression; + use std::io::Write; + + let mut enc = ZlibEncoder::new(Vec::new(), Compression::default()); + enc.write_all(payload).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); + src + } + + #[test] + fn decompression_bomb_exceeds_limit() { + let mut src = zlib_frame(&vec![b'A'; 200]); + + // A limit of 10 bytes is well below the 200-byte inflated output. + let result = decode_compressed_frame(&mut src, 10); + assert!( + matches!(result, Err(DecodeError::DecompressionFailed { .. })), + "expected DecompressionFailed, got {result:?}", + ); + } + + /// Boundary: a payload that inflates to exactly the limit is legal. Capping the reader at + /// `max` (rather than `max + 1`) made this case indistinguishable from a truncated bomb and + /// rejected it. + #[test] + fn decompression_at_exactly_the_limit_is_accepted() { + let plain = vec![b'A'; 200]; + let mut src = zlib_frame(&plain); + + let result = decode_compressed_frame(&mut src, plain.len() as u64); + assert!( + !matches!(result, Err(DecodeError::DecompressionFailed { .. })), + "a payload exactly at the limit must not be rejected as a bomb, got {result:?}", + ); + } + + #[test] + fn decompression_one_byte_over_the_limit_is_rejected() { + let plain = vec![b'A'; 200]; + let mut src = zlib_frame(&plain); + + let result = decode_compressed_frame(&mut src, plain.len() as u64 - 1); + assert!( + matches!(result, Err(DecodeError::DecompressionFailed { .. })), + "one byte over the limit must be rejected, got {result:?}", + ); + } + + /// The source bytes must be consumed even when the frame is rejected, otherwise the same bomb + /// is re-decoded forever. + #[test] + fn rejected_bomb_still_advances_the_source_buffer() { + let mut src = zlib_frame(&vec![b'A'; 200]); + let original_len = src.len(); + + let _ = decode_compressed_frame(&mut src, 10); + assert!( + src.len() < original_len, + "the rejected frame's bytes must be drained from the buffer" + ); + } + + #[test] + fn nested_compressed_frame_rejected() { + // 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 again. + let mut src = zlib_frame(&[0x32, 0x43, 0, 0, 0, 0]); + + let result = decode_compressed_frame(&mut src, 1024 * 1024); + assert!( + matches!(result, Err(DecodeError::NestedCompressionRejected)), + "expected NestedCompressionRejected, got {result:?}", + ); + } + + /// A nested compressed frame is unrecoverable: continuing would let the sender keep feeding + /// nested bombs down the same connection. + #[test] + fn nested_compression_error_terminates_the_stream() { + assert!(!DecodeError::NestedCompressionRejected.can_continue()); + } + + /// A single oversized frame is a per-frame condition, so the connection survives it. + #[test] + fn decompression_failure_does_not_terminate_the_stream() { + assert!(DecodeError::DecompressionFailed { + source: io::Error::new(io::ErrorKind::Other, "boom"), + } + .can_continue()); + } + + #[test] + fn top_level_decoder_is_not_marked_inside_compressed() { + // Only frames reached *through* a compressed frame may reject nesting; a plain 'C' frame + // at the top level is legal and must still decode. + assert!(!LogstashDecoder::new(DEFAULT_MAX_DECOMPRESSED_BYTES).inside_compressed); + assert!( + LogstashDecoder::new_inside_compressed(DEFAULT_MAX_DECOMPRESSED_BYTES) + .inside_compressed + ); + } + + #[test] + fn default_max_decompressed_bytes_is_256_mib() { + // Pinned deliberately: this bound is per-frame, so raising it multiplies peak memory by + // the concurrent connection count. + assert_eq!(DEFAULT_MAX_DECOMPRESSED_BYTES, 256 * 1024 * 1024); + assert_eq!( + LogstashConfig::default().max_decompressed_bytes, + DEFAULT_MAX_DECOMPRESSED_BYTES + ); + } + + #[test] + fn max_decompressed_bytes_round_trips_through_config() { + let config: LogstashConfig = + serde_json::from_str(r#"{"address":"0.0.0.0:5044","max_decompressed_bytes":1234}"#) + .unwrap(); + assert_eq!(config.max_decompressed_bytes, 1234); + } + + #[test] + fn max_decompressed_bytes_defaults_when_absent_from_config() { + let config: LogstashConfig = + serde_json::from_str(r#"{"address":"0.0.0.0:5044"}"#).unwrap(); + assert_eq!( + config.max_decompressed_bytes, + DEFAULT_MAX_DECOMPRESSED_BYTES + ); + } + #[tokio::test] async fn test_delivered() { test_protocol(EventStatus::Delivered, true).await; @@ -756,6 +964,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 +1221,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 +1232,5 @@ mod integration_tests { wait_for_tcp(address).await; recv } + } diff --git a/src/sources/util/net/tcp/mod.rs b/src/sources/util/net/tcp/mod.rs index 13bb464ab..b16b0a8b0 100644 --- a/src/sources/util/net/tcp/mod.rs +++ b/src/sources/util/net/tcp/mod.rs @@ -9,7 +9,7 @@ use listenfd::ListenFd; use smallvec::SmallVec; use socket2::SockRef; use tokio::{ - io::AsyncWriteExt, + io::{AsyncWrite, AsyncWriteExt}, net::{TcpListener, TcpStream}, time::sleep, }; @@ -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 write_ack(stream, &ack_bytes, ACK_WRITE_TIMEOUT).await { + AckWriteOutcome::Written => {} + AckWriteOutcome::Failed(error) => { + emit!(TcpSendAckError{ error }); + break; + } + AckWriteOutcome::TimedOut => { + warn!( + timeout_secs = ACK_WRITE_TIMEOUT.as_secs(), + "Ack write timed out; dropping connection." + ); + break; + } } } if ack != TcpSourceAck::Ack { @@ -412,6 +428,145 @@ async fn handle_stream( } } +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncReadExt; + + #[tokio::test] + async fn write_ack_succeeds_when_peer_reads() { + let (mut client, mut server) = tokio::io::duplex(64); + + let write = tokio::spawn(async move { + write_ack(&mut server, b"ack", ACK_WRITE_TIMEOUT).await + }); + + let mut buf = [0u8; 3]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ack"); + assert!(matches!(write.await.unwrap(), AckWriteOutcome::Written)); + } + + /// A peer that never drains its receive window must not block the ack write forever. Before + /// the timeout existed, this write parked indefinitely. + #[tokio::test(start_paused = true)] + async fn write_ack_times_out_against_a_peer_that_never_reads() { + // A 1-byte duplex fills immediately and `client` is never read from, so the write stalls. + let (_client, mut server) = tokio::io::duplex(1); + let payload = vec![0u8; 1024]; + + let outcome = write_ack(&mut server, &payload, ACK_WRITE_TIMEOUT).await; + assert!( + matches!(outcome, AckWriteOutcome::TimedOut), + "expected TimedOut, got {outcome:?}" + ); + } + + /// The timeout must not fire early for a peer that is merely slow rather than stuck. + #[tokio::test(start_paused = true)] + async fn write_ack_tolerates_a_slow_but_progressing_peer() { + let (mut client, mut server) = tokio::io::duplex(4); + let payload = vec![7u8; 32]; + let expected = payload.clone(); + + let write = + tokio::spawn(async move { write_ack(&mut server, &payload, ACK_WRITE_TIMEOUT).await }); + + let mut received = Vec::new(); + while received.len() < expected.len() { + // Drain in small sips, pausing well inside the timeout each round. + tokio::time::sleep(Duration::from_secs(1)).await; + let mut chunk = [0u8; 4]; + let n = client.read(&mut chunk).await.unwrap(); + received.extend_from_slice(&chunk[..n]); + } + + assert_eq!(received, expected); + assert!(matches!(write.await.unwrap(), AckWriteOutcome::Written)); + } + + #[tokio::test] + async fn write_ack_reports_failure_when_peer_hangs_up() { + let (client, mut server) = tokio::io::duplex(64); + drop(client); + + let outcome = write_ack(&mut server, &vec![0u8; 4096], ACK_WRITE_TIMEOUT).await; + assert!( + matches!(outcome, AckWriteOutcome::Failed(_)), + "expected Failed, got {outcome:?}" + ); + } + + /// The permit must be released before the ack write, so a stuck peer cannot hold a + /// `RequestLimiter` slot and starve other connections (OBE-11555). This models the ordering + /// `handle_stream` uses: take the permit, then perform the (stalling) write. + #[tokio::test(start_paused = true)] + async fn permit_is_released_before_a_stalled_ack_write() { + let limiter = RequestLimiter::new(1, 1); + // The limiter starts at its floor of 2 permits; hold every one so the next acquire blocks. + let held = limiter.acquire().await; + let mut permit = Some(limiter.acquire().await); + assert!( + tokio::time::timeout(Duration::from_millis(50), limiter.acquire()) + .await + .is_err(), + "all permits are held, so a further acquire must block" + ); + + // Ordering under test: release, then write to a peer that never reads. + let _ = permit.take(); + + let (_client, mut server) = tokio::io::duplex(1); + let write = tokio::spawn(async move { + write_ack(&mut server, &vec![0u8; 1024], ACK_WRITE_TIMEOUT).await + }); + + // While the write is stalled, another connection must still get a permit. + let second = tokio::time::timeout(Duration::from_secs(1), limiter.acquire()).await; + assert!( + second.is_ok(), + "permit must be available while the ack write is stalled" + ); + + assert!(matches!(write.await.unwrap(), AckWriteOutcome::TimedOut)); + drop(held); + } + + #[test] + fn ack_write_timeout_is_thirty_seconds() { + assert_eq!(ACK_WRITE_TIMEOUT, Duration::from_secs(30)); + } +} + +/// How long to wait for an ack to reach the peer before giving up on the connection. +const ACK_WRITE_TIMEOUT: Duration = Duration::from_secs(30); + +/// Result of attempting to write an ack back to the peer. +#[derive(Debug)] +enum AckWriteOutcome { + Written, + /// The write failed; the connection should be torn down. + Failed(std::io::Error), + /// The peer never drained its receive window within the timeout. + TimedOut, +} + +/// Writes `ack_bytes` to `stream`, bounded by `timeout`. +/// +/// Without the timeout a peer that stops reading parks this write forever. That matters because +/// the caller has already released its `RequestLimiterPermit` by this point (OBE-11555) — the +/// connection itself still needs to be reclaimed. +async fn write_ack(stream: &mut S, ack_bytes: &[u8], timeout: Duration) -> AckWriteOutcome +where + S: AsyncWrite + Unpin + ?Sized, +{ + match tokio::time::timeout(timeout, stream.write_all(ack_bytes)).await { + Ok(Ok(())) => AckWriteOutcome::Written, + Ok(Err(error)) => AckWriteOutcome::Failed(error), + Err(_elapsed) => AckWriteOutcome::TimedOut, + } +} + fn close_socket(socket: &MaybeTlsIncomingStream) -> bool { debug!("Start graceful shutdown."); // Close our write part of TCP socket to signal the other side diff --git a/website/cue/reference/components/sources/base/logstash.cue b/website/cue/reference/components/sources/base/logstash.cue index 920f859b4..819e51f2f 100644 --- a/website/cue/reference/components/sources/base/logstash.cue +++ b/website/cue/reference/components/sources/base/logstash.cue @@ -46,6 +46,20 @@ base: components: sources: logstash: configuration: { type: uint: unit: "seconds" } } + max_decompressed_bytes: { + description: """ + Maximum size in bytes that a compressed frame payload is allowed to expand to. + Guards against decompression bomb (zip bomb) attacks. + + This bound applies per frame, so peak memory scales with the number of concurrent + connections. Raise it only alongside a finite `connection_limit`. + """ + required: false + type: uint: { + default: 268435456 + unit: "bytes" + } + } permit_origin: { description: "List of allowed origin IP networks. IP addresses must be in CIDR notation." required: false