From 02aa49c963eeb49f8d1d957565fb3cd7f7c0abbc Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Fri, 7 Aug 2026 23:27:39 +0530 Subject: [PATCH 1/4] OBE-11232: bound delimited framing by default and reset the stream on an over-long frame --- .../decoding/framing/character_delimited.rs | 350 ++++++++++++++---- .../src/decoding/framing/newline_delimited.rs | 37 +- lib/codecs/src/lib.rs | 1 + lib/codecs/src/max_length.rs | 56 +++ src/cli.rs | 91 +++++ src/sources/socket/mod.rs | 126 +++++-- 6 files changed, 557 insertions(+), 104 deletions(-) create mode 100644 lib/codecs/src/max_length.rs diff --git a/lib/codecs/src/decoding/framing/character_delimited.rs b/lib/codecs/src/decoding/framing/character_delimited.rs index 60ecc601c7..f698507c8d 100644 --- a/lib/codecs/src/decoding/framing/character_delimited.rs +++ b/lib/codecs/src/decoding/framing/character_delimited.rs @@ -1,10 +1,60 @@ use bytes::{Buf, Bytes, BytesMut}; use memchr::memchr; use tokio_util::codec::Decoder; -use tracing::{trace, warn}; +use tracing::trace; use vector_config::configurable_component; -use super::BoxedFramingError; +use super::{BoxedFramingError, FramingError}; +use crate::decoding::StreamDecodingError; +use crate::max_length::max_frame_length_bytes; + +/// A frame exceeded `max_length`. +/// +/// Always fatal (`can_continue() == false`): the buffer is dropped and the transport resets the +/// connection. A peer that sends one illegal frame gives us no reason to trust the rest of its +/// stream, so the frame is not skipped over even when its delimiter is present and we could. +/// +/// `terminated` records whether the delimiter had arrived, purely so the log says which of the two +/// situations occurred — it does not change the outcome. +#[derive(Debug)] +pub struct FrameTooLong { + frame_length: usize, + max_length: usize, + terminated: bool, +} + +impl std::fmt::Display for FrameTooLong { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.terminated { + write!( + f, + "frame exceeds max_length: {} bytes, limit is {}; resetting the connection", + self.frame_length, self.max_length + ) + } else { + write!( + f, + "frame exceeds max_length: buffered {} bytes with no delimiter, limit is {}; \ + resetting the connection", + self.frame_length, self.max_length + ) + } + } +} + +impl std::error::Error for FrameTooLong {} + +impl StreamDecodingError for FrameTooLong { + fn can_continue(&self) -> bool { + false + } +} + +impl FramingError for FrameTooLong { + fn as_any(&self) -> &dyn std::any::Any { + self as &dyn std::any::Any + } +} /// Config used to build a `CharacterDelimitedDecoder`. #[configurable_component] @@ -22,7 +72,7 @@ impl CharacterDelimitedDecoderConfig { } } /// Build the `CharacterDelimitedDecoder` from this configuration. - pub const fn build(&self) -> CharacterDelimitedDecoder { + pub fn build(&self) -> CharacterDelimitedDecoder { if let Some(max_length) = self.character_delimited.max_length { CharacterDelimitedDecoder::new_with_max_length( self.character_delimited.delimiter, @@ -47,13 +97,12 @@ pub struct CharacterDelimitedDecoderOptions { /// /// This length does *not* include the trailing delimiter. /// - /// By default, there is no maximum length enforced. If events are malformed, this can lead to - /// additional resource usage as events continue to be buffered in memory, and can potentially - /// lead to memory exhaustion in extreme cases. + /// Defaults to the global frame length cap, set by `--max-frame-length-bytes` (or + /// `VECTOR_MAX_FRAME_LENGTH_BYTES`), which is 100 KiB unless overridden. Set this to override + /// the cap for this component alone. /// - /// If there is a risk of processing malformed data, such as logs with user-controlled input, - /// consider setting the maximum length to a reasonably large value as a safety net. This - /// ensures that processing is not actually unbounded. + /// A frame longer than the limit is a fatal decode error and the connection is reset, whether + /// or not its delimiter had arrived. #[serde(skip_serializing_if = "vector_core::serde::is_default")] pub max_length: Option, } @@ -78,21 +127,20 @@ pub struct CharacterDelimitedDecoder { } impl CharacterDelimitedDecoder { - /// Creates a `CharacterDelimitedDecoder` with the specified delimiter. - pub const fn new(delimiter: u8) -> Self { - CharacterDelimitedDecoder { - delimiter, - max_length: usize::MAX, - } + /// Creates a `CharacterDelimitedDecoder` with the specified delimiter, using the global frame + /// length cap (see [`crate::max_length`]). + pub fn new(delimiter: u8) -> Self { + Self::new_with_max_length(delimiter, max_frame_length_bytes()) } /// Creates a `CharacterDelimitedDecoder` with a maximum frame length limit. /// - /// Any frames longer than `max_length` bytes will be discarded entirely. + /// A frame longer than `max_length` is a fatal decode error and the connection is reset — see + /// [`FrameTooLong`]. pub const fn new_with_max_length(delimiter: u8, max_length: usize) -> Self { CharacterDelimitedDecoder { + delimiter, max_length, - ..CharacterDelimitedDecoder::new(delimiter) } } @@ -107,36 +155,50 @@ impl Decoder for CharacterDelimitedDecoder { type Error = BoxedFramingError; fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { - loop { - // This function has the following goal: we are searching for - // sub-buffers delimited by `self.delimiter` with size no more than - // `self.max_length`. If a sub-buffer is found that exceeds - // `self.max_length` we discard it, else we return it. At the end of - // the buffer if the delimiter is not present the remainder of the - // buffer is discarded. - match memchr(self.delimiter, buf) { - None => return Ok(None), - Some(next_delimiter_idx) => { - if next_delimiter_idx > self.max_length { - // The discovered sub-buffer is too big, so we discard - // it, taking care to also discard the delimiter. - warn!( - message = "Discarding frame larger than max_length.", - buf_len = buf.len(), - max_length = self.max_length, - internal_log_rate_limit = true - ); - buf.advance(next_delimiter_idx + 1); - } else { - let frame = buf.split_to(next_delimiter_idx).freeze(); - trace!( - message = "Decoding the frame.", - bytes_processed = frame.len() - ); - buf.advance(1); // scoot past the delimiter - return Ok(Some(frame)); + // A frame longer than `max_length` fails the stream, whether or not its delimiter has + // arrived. See `FrameTooLong`. + match memchr(self.delimiter, buf) { + None => { + // `memchr` searched all of `buf` and found no delimiter, and any frame emitted or + // rejected previously was removed from `buf` at that point. So whatever is here is + // exactly one incomplete frame — `buf.len()` is that frame's length so far, not + // the size of the last socket read. A read much larger than `max_length` is fine + // as long as it contains delimiters: those frames are handled by the arm below. + // + // `>` and not `>=`: at exactly `max_length` bytes the next byte could still be the + // delimiter, which would make it a legal frame of the maximum size. + if buf.len() > self.max_length { + let frame_length = buf.len(); + buf.clear(); + return Err(FrameTooLong { + frame_length, + max_length: self.max_length, + terminated: false, } + .into()); } + Ok(None) + } + Some(next_delimiter_idx) => { + if next_delimiter_idx > self.max_length { + // We could resync here — the delimiter marks exactly where this frame ended — + // but an over-long frame fails the stream outright, so everything buffered + // goes with it. + buf.clear(); + return Err(FrameTooLong { + frame_length: next_delimiter_idx, + max_length: self.max_length, + terminated: true, + } + .into()); + } + let frame = buf.split_to(next_delimiter_idx).freeze(); + trace!( + message = "Decoding the frame.", + bytes_processed = frame.len() + ); + buf.advance(1); // scoot past the delimiter + Ok(Some(frame)) } } } @@ -147,15 +209,9 @@ impl Decoder for CharacterDelimitedDecoder { None => { if buf.is_empty() { Ok(None) - } else if buf.len() > self.max_length { - warn!( - message = "Discarding frame larger than max_length.", - buf_len = buf.len(), - max_length = self.max_length, - internal_log_rate_limit = true - ); - Ok(None) } else { + // `decode` returned `Ok(None)`, so the remainder is within `max_length`; + // anything longer would already have errored above. let bytes: Bytes = buf.split_to(buf.len()).freeze(); Ok(Some(bytes)) } @@ -181,31 +237,198 @@ mod tests { assert_eq!(Some("abc".into()), codec.decode(buf).unwrap()); } + /// A peer that never sends the delimiter must not be able to grow the buffer without bound. + /// `max_length` previously only applied once a delimiter had been found, so this stream was + /// unbounded even with the limit set explicitly. #[test] - fn decode_max_length() { - const MAX_LENGTH: usize = 6; + fn incomplete_frame_over_max_length_is_a_fatal_error() { + const MAX_LENGTH: usize = 10; let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); let buf = &mut BytesMut::new(); + buf.put_slice(&[b'x'; 100]); + + let error = codec.decode(buf).unwrap_err(); + assert!( + !error.can_continue(), + "an over-long incomplete frame must be fatal so the stream closes", + ); + assert!( + error.to_string().contains("exceeds max_length"), + "error should name the limit, got: {error}", + ); + assert!(buf.is_empty(), "the pending bytes must be released"); + } - // limit is 6 so it will skip longer lines - buf.put_slice(b"1234567\n123456\n123412314\n123"); + /// The bound must not fire while the frame is still within the limit — that would reject + /// ordinary streaming reads that simply have not seen their delimiter yet. + #[test] + fn incomplete_frame_within_max_length_waits_for_more_bytes() { + const MAX_LENGTH: usize = 100; - assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456"))); + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + + buf.put_slice(b"partial"); + assert_eq!(codec.decode(buf).unwrap(), None); + assert_eq!(buf.len(), 7, "pending bytes must be kept for the next read"); + + buf.put_slice(b" frame\n"); + assert_eq!( + codec.decode(buf).unwrap(), + Some(Bytes::from("partial frame")) + ); + } + + /// A frame of exactly `max_length` is accepted; one byte more is not. Pins the boundary so a + /// later refactor cannot silently turn `>` into `>=`. + #[test] + fn max_length_boundary_is_exact() { + const MAX_LENGTH: usize = 10; + + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + buf.put_slice(&[b'x'; MAX_LENGTH]); + buf.put_slice(b"\n"); + assert_eq!( + codec.decode(buf).unwrap(), + Some(Bytes::from(vec![b'x'; MAX_LENGTH])), + "a frame of exactly max_length must be accepted", + ); + + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + buf.put_slice(&[b'x'; MAX_LENGTH + 1]); + assert!( + codec.decode(buf).is_err(), + "one byte past max_length must be rejected", + ); + } + + /// A single socket read is routinely far larger than `max_length`. That must be fine as long + /// as it contains delimiters: the limit bounds one *frame*, not the read. This is the + /// regression guard against measuring the wrong thing. + #[test] + fn read_much_larger_than_max_length_is_fine_when_it_contains_frames() { + const MAX_LENGTH: usize = 10; + + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + + // 8 KiB in one read, made of 1000 short valid frames — 800x the limit in total. + for _ in 0..1_000 { + buf.put_slice(b"abcdefg\n"); + } + let total = buf.len(); + assert!( + total > MAX_LENGTH * 100, + "test should exceed the limit many times over" + ); + + for i in 0..1_000 { + assert_eq!( + codec.decode(buf).unwrap(), + Some(Bytes::from("abcdefg")), + "frame {i} of a large multi-frame read should decode", + ); + } assert_eq!(codec.decode(buf).unwrap(), None); + assert!(buf.is_empty()); + } + + /// The same, but the large read ends mid-frame: the complete frames decode and the short tail + /// waits for more bytes rather than being judged against the total read size. + #[test] + fn large_read_with_trailing_partial_frame_waits_instead_of_erroring() { + const MAX_LENGTH: usize = 10; + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); let buf = &mut BytesMut::new(); + for _ in 0..500 { + buf.put_slice(b"abcdefg\n"); + } + buf.put_slice(b"tail"); // 4 bytes, under the limit, no delimiter yet - // limit is 6 so it will skip longer lines - buf.put_slice(b"1234567\n123456\n123412314\n123"); + for _ in 0..500 { + assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("abcdefg"))); + } + assert_eq!( + codec.decode(buf).unwrap(), + None, + "a short trailing partial must wait, not error", + ); + assert_eq!(buf.len(), 4, "the partial frame must be retained"); + } - assert_eq!(codec.decode_eof(buf).unwrap(), Some(Bytes::from("123456"))); + /// Exactly `max_length` bytes with no delimiter must wait: the very next byte could be the + /// delimiter, making it a legal maximum-size frame. One byte more cannot be legal. + #[test] + fn exactly_max_length_without_delimiter_still_waits() { + const MAX_LENGTH: usize = 10; + + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + buf.put_slice(&[b'x'; MAX_LENGTH]); + assert_eq!( + codec.decode(buf).unwrap(), + None, + "at exactly max_length the frame may still be completed by the next byte", + ); + + buf.put_slice(b"\n"); + assert_eq!( + codec.decode(buf).unwrap(), + Some(Bytes::from(vec![b'x'; MAX_LENGTH])), + ); + } + + /// `new()` must pick up the global cap rather than the old unbounded default. + #[test] + fn new_uses_the_global_frame_length_cap() { + let codec = CharacterDelimitedDecoder::new(b'\n'); + assert_eq!( + codec.max_length(), + crate::max_length::max_frame_length_bytes() + ); + assert_ne!(codec.max_length(), usize::MAX); + } + + #[test] + fn decode_max_length() { + const MAX_LENGTH: usize = 6; + + // A terminated frame longer than the limit is fatal, exactly as an over-long incomplete + // frame is. It used to be skipped so that following frames still decoded; that split + // behaviour is gone, so nothing after the offending frame is read. + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + buf.put_slice(b"1234567\n123456\n"); + let error = codec.decode(buf).unwrap_err(); + assert!(!error.can_continue()); + assert!( + buf.is_empty(), + "the stream is abandoned, not resynchronized" + ); + + // Frames within the limit are untouched, including one of exactly `max_length`. + let mut codec = CharacterDelimitedDecoder::new_with_max_length(b'\n', MAX_LENGTH); + let buf = &mut BytesMut::new(); + buf.put_slice(b"123456\n12345\n123"); + assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("123456"))); + assert_eq!(codec.decode(buf).unwrap(), Some(Bytes::from("12345"))); + assert_eq!(codec.decode(buf).unwrap(), None); assert_eq!(codec.decode_eof(buf).unwrap(), Some(Bytes::from("123"))); assert_eq!(codec.decode_eof(buf).unwrap(), None); } // Regression test for [infinite loop bug](https://github.com/vectordotdev/vector/issues/2564) // Derived from https://github.com/tokio-rs/tokio/issues/1483 + // + // The guarantee this pins is "no spinning on the same bytes". It used to be met by returning + // `Ok(None)` for an over-long incomplete frame, which meant the caller kept reading and the + // buffer kept growing — bounded progress, but unbounded memory. It is now met by failing the + // frame outright and releasing the buffer, so there is still no repeated work on the same + // bytes and memory is bounded as well. #[test] fn decode_discard_repeat() { const MAX_LENGTH: usize = 1; @@ -215,7 +438,10 @@ mod tests { buf.reserve(200); buf.put(&b"aa"[..]); - assert!(codec.decode(buf).unwrap().is_none()); + assert!(codec.decode(buf).is_err()); + assert!(buf.is_empty(), "the rejected frame must be released"); + + // No spin: the decoder makes progress rather than re-failing on the same bytes. buf.put(&b"a"[..]); assert!(codec.decode(buf).unwrap().is_none()); } diff --git a/lib/codecs/src/decoding/framing/newline_delimited.rs b/lib/codecs/src/decoding/framing/newline_delimited.rs index 7bdc3a6088..a4b1e1e6de 100644 --- a/lib/codecs/src/decoding/framing/newline_delimited.rs +++ b/lib/codecs/src/decoding/framing/newline_delimited.rs @@ -23,13 +23,12 @@ pub struct NewlineDelimitedDecoderOptions { /// /// This length does *not* include the trailing delimiter. /// - /// By default, there is no maximum length enforced. If events are malformed, this can lead to - /// additional resource usage as events continue to be buffered in memory, and can potentially - /// lead to memory exhaustion in extreme cases. + /// Defaults to the global frame length cap, set by `--max-frame-length-bytes` (or + /// `VECTOR_MAX_FRAME_LENGTH_BYTES`), which is 100 KiB unless overridden. Set this to override + /// the cap for this component alone. /// - /// If there is a risk of processing malformed data, such as logs with user-controlled input, - /// consider setting the maximum length to a reasonably large value as a safety net. This - /// ensures that processing is not actually unbounded. + /// A frame longer than the limit is a fatal decode error and the connection is reset, whether + /// or not its delimiter had arrived. #[serde(skip_serializing_if = "vector_core::serde::is_default")] pub max_length: Option, } @@ -57,7 +56,7 @@ impl NewlineDelimitedDecoderConfig { } /// Build the `NewlineDelimitedDecoder` from this configuration. - pub const fn build(&self) -> NewlineDelimitedDecoder { + pub fn build(&self) -> NewlineDelimitedDecoder { if let Some(max_length) = self.newline_delimited.max_length { NewlineDelimitedDecoder::new_with_max_length(max_length) } else { @@ -72,7 +71,7 @@ pub struct NewlineDelimitedDecoder(CharacterDelimitedDecoder); impl NewlineDelimitedDecoder { /// Creates a new `NewlineDelimitedDecoder`. - pub const fn new() -> Self { + pub fn new() -> Self { Self(CharacterDelimitedDecoder::new(b'\n')) } @@ -132,12 +131,16 @@ mod tests { #[test] fn decode_bytes_with_newlines_and_max_length() { + // An over-long line is now fatal rather than skipped, so "baz" behind it is never read. let mut input = BytesMut::from("foo\nbarbara\nbaz\n"); let mut decoder = NewlineDelimitedDecoder::new_with_max_length(3); assert_eq!(decoder.decode(&mut input).unwrap().unwrap(), "foo"); - assert_eq!(decoder.decode(&mut input).unwrap().unwrap(), "baz"); - assert_eq!(decoder.decode(&mut input).unwrap(), None); + assert!(decoder.decode(&mut input).is_err()); + assert!( + input.is_empty(), + "the stream is abandoned, not resynchronized" + ); } #[test] @@ -167,6 +170,20 @@ mod tests { let mut decoder = NewlineDelimitedDecoder::new_with_max_length(3); assert_eq!(decoder.decode_eof(&mut input).unwrap().unwrap(), "foo"); + assert!(decoder.decode_eof(&mut input).is_err()); + assert!(input.is_empty()); + } + + /// Lines within the limit are unaffected by the cap, including at EOF without a trailing + /// delimiter — the accept half of the pair above. + #[test] + fn decode_bytes_within_max_length_are_unaffected() { + let mut input = BytesMut::from("foo\nbar\nbaz"); + let mut decoder = NewlineDelimitedDecoder::new_with_max_length(3); + + assert_eq!(decoder.decode(&mut input).unwrap().unwrap(), "foo"); + assert_eq!(decoder.decode(&mut input).unwrap().unwrap(), "bar"); + assert_eq!(decoder.decode(&mut input).unwrap(), None); assert_eq!(decoder.decode_eof(&mut input).unwrap().unwrap(), "baz"); assert_eq!(decoder.decode_eof(&mut input).unwrap(), None); } diff --git a/lib/codecs/src/lib.rs b/lib/codecs/src/lib.rs index ae22f9c18d..592c9284d3 100644 --- a/lib/codecs/src/lib.rs +++ b/lib/codecs/src/lib.rs @@ -8,6 +8,7 @@ mod common; pub mod decoding; pub mod encoding; pub mod gelf; +pub mod max_length; pub use decoding::{ BytesDecoder, BytesDecoderConfig, BytesDeserializer, BytesDeserializerConfig, diff --git a/lib/codecs/src/max_length.rs b/lib/codecs/src/max_length.rs new file mode 100644 index 0000000000..59b08d2e3a --- /dev/null +++ b/lib/codecs/src/max_length.rs @@ -0,0 +1,56 @@ +//! Global cap on the length of a single delimited frame. +//! +//! A delimited framer accumulates bytes until it sees its delimiter. A peer that never sends one +//! would otherwise grow the per-connection buffer with everything it sends, so the framers consult +//! this cap while a frame is still incomplete and reject anything past it. +//! +//! The cap is process-wide and set once at startup from `--max-frame-length-bytes` (or +//! `VECTOR_MAX_FRAME_LENGTH_BYTES`). Individual framers may still override it per component via +//! their own `max_length` option; this only supplies the default. + +use std::sync::OnceLock; + +/// Default cap on the length of a single delimited frame. +/// +/// Matches the limit the `file`, `syslog`, `stdin`, `file_descriptor` and `socket` mode `udp` +/// sources have always applied, so a frame length that is acceptable to one of those is acceptable +/// to every delimited framer. +pub const DEFAULT_MAX_FRAME_LENGTH_BYTES: usize = 100 * 1024; + +static MAX_FRAME_LENGTH_BYTES: OnceLock = OnceLock::new(); + +/// Override the global frame length cap. Must be called before any sources start. +/// +/// # Panics +/// +/// Panics if called more than once, as the global cap may only be initialized a single time. +pub fn set_max_frame_length_bytes(size: usize) { + MAX_FRAME_LENGTH_BYTES + .set(size) + .expect("max_frame_length_bytes already set"); +} + +/// Returns the currently configured frame length cap. +#[must_use] +pub fn max_frame_length_bytes() -> usize { + *MAX_FRAME_LENGTH_BYTES + .get() + .unwrap_or(&DEFAULT_MAX_FRAME_LENGTH_BYTES) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Without an explicit override the getter must report the documented default, since that is + /// what every framer built by `new()` will enforce. + #[test] + fn defaults_to_the_documented_limit() { + assert_eq!(DEFAULT_MAX_FRAME_LENGTH_BYTES, 102_400); + // `set_*` is process-global and may have been called by another test in this binary, so + // only assert the default when it has not been overridden. + if MAX_FRAME_LENGTH_BYTES.get().is_none() { + assert_eq!(max_frame_length_bytes(), DEFAULT_MAX_FRAME_LENGTH_BYTES); + } + } +} diff --git a/src/cli.rs b/src/cli.rs index 6ab179ddfa..c86c325969 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -236,6 +236,45 @@ pub struct RootOpts { /// `--watch-config`. #[arg(long, env = "VECTOR_ALLOW_EMPTY_CONFIG", default_value = "false")] pub allow_empty_config: bool, + + /// Maximum length, in bytes, of a single delimited frame. + /// + /// Delimited framers buffer bytes until they see their delimiter, so a peer that never sends + /// one would otherwise grow the per-connection buffer without bound. A frame that reaches this + /// limit while still incomplete is rejected and the stream is closed. + /// + /// Defaults to 102400 (100 KiB). Raise this only when sources routinely receive legitimately + /// long frames; individual components can also override it with their own `max_length` option. + /// + /// Must be at least 1024; a cap below that would reject essentially all traffic rather than + /// just runaway frames. + #[arg( + long, + env = "VECTOR_MAX_FRAME_LENGTH_BYTES", + default_value_t = vector_lib::codecs::max_length::DEFAULT_MAX_FRAME_LENGTH_BYTES, + value_parser = parse_max_frame_length_bytes, + )] + pub max_frame_length_bytes: usize, +} + +/// Lower bound for `--max-frame-length-bytes`. +/// +/// Guards against a value (notably `0`) that would silently reject all delimited ingestion, which +/// looks identical to a broken pipeline from the outside. +const MIN_MAX_FRAME_LENGTH_BYTES: usize = 1024; + +fn parse_max_frame_length_bytes(raw: &str) -> Result { + let value: usize = raw + .parse() + .map_err(|_| format!("`{raw}` is not a valid number of bytes"))?; + + if value < MIN_MAX_FRAME_LENGTH_BYTES { + return Err(format!( + "`{raw}` is below the minimum of {MIN_MAX_FRAME_LENGTH_BYTES} bytes" + )); + } + + Ok(value) } impl RootOpts { @@ -262,6 +301,8 @@ impl RootOpts { } crate::metrics::init_global().expect("metrics initialization failed"); + + vector_lib::codecs::max_length::set_max_frame_length_bytes(self.max_frame_length_bytes); } } @@ -395,3 +436,53 @@ pub fn handle_config_errors(errors: Vec) -> exitcode::ExitCode { exitcode::CONFIG } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn max_frame_length_accepts_reasonable_values() { + assert_eq!(parse_max_frame_length_bytes("102400"), Ok(102_400)); + assert_eq!( + parse_max_frame_length_bytes("1024"), + Ok(MIN_MAX_FRAME_LENGTH_BYTES) + ); + } + + /// A zero (or near-zero) cap would reject essentially all delimited ingestion while looking + /// like a silently broken pipeline, so it must fail loudly at startup instead. + #[test] + fn max_frame_length_rejects_values_below_the_floor() { + for raw in ["0", "1", "512", "1023"] { + assert!( + parse_max_frame_length_bytes(raw).is_err(), + "{raw} should have been rejected" + ); + } + } + + #[test] + fn max_frame_length_rejects_non_numeric() { + assert!(parse_max_frame_length_bytes("100KiB").is_err()); + assert!(parse_max_frame_length_bytes("-1").is_err()); + assert!(parse_max_frame_length_bytes("").is_err()); + } + + /// The clap default must track the constant the framers actually enforce. + #[test] + fn cli_default_matches_the_enforced_default() { + let opts = RootOpts::parse_from(["vector"]); + assert_eq!( + opts.max_frame_length_bytes, + vector_lib::codecs::max_length::DEFAULT_MAX_FRAME_LENGTH_BYTES + ); + } + + /// The flag must actually be settable, otherwise the escape hatch for long frames is absent. + #[test] + fn cli_flag_overrides_the_default() { + let opts = RootOpts::parse_from(["vector", "--max-frame-length-bytes", "1048576"]); + assert_eq!(opts.max_frame_length_bytes, 1_048_576); + } +} diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs index 9159161b26..786ad6e53a 100644 --- a/src/sources/socket/mod.rs +++ b/src/sources/socket/mod.rs @@ -458,6 +458,62 @@ mod test { .await; } + /// End-to-end proof that the frame length cap actually terminates the connection rather than + /// just erroring internally: a peer that streams past the limit without ever sending a newline + /// must have its socket closed by the server, so the buffer cannot keep growing. + #[tokio::test] + async fn tcp_over_long_frame_closes_the_connection() { + use tokio::io::AsyncWriteExt as _; + + let (tx, _rx) = SourceSender::new_test(); + let addr = next_addr(); + + let mut conf = TcpConfig::from_address(addr.into()); + conf.set_framing(Some( + NewlineDelimitedDecoderConfig::new_with_max_length(1024).into(), + )); + + let server = SocketConfig::from(conf) + .build(SourceContext::new_test(tx, None)) + .await + .unwrap(); + tokio::spawn(server); + wait_for_tcp(addr).await; + + let mut stream = TcpStream::connect(addr).await.unwrap(); + + // Never send a newline. Write until the peer closes on us, or we have clearly exceeded + // the limit without being closed (which is the failure this test guards against). + let chunk = vec![b'x'; 4096]; + let mut written = 0usize; + let mut closed = false; + for _ in 0..64 { + match stream.write_all(&chunk).await { + Ok(()) => written += chunk.len(), + Err(_) => { + closed = true; + break; + } + } + } + + if !closed { + // The write side may still buffer locally after the server drops the connection, so + // confirm via the read side: a closed connection reads EOF. + let mut buf = [0u8; 1]; + closed = matches!( + timeout(Duration::from_secs(10), stream.read(&mut buf)).await, + Ok(Ok(0)) | Ok(Err(_)) + ); + } + + assert!( + closed, + "connection stayed open after streaming {written} bytes with no delimiter \ + against a 1024-byte max_length", + ); + } + #[tokio::test] async fn tcp_it_includes_vector_namespaced_fields() { assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async { @@ -557,44 +613,50 @@ mod test { } #[tokio::test] - async fn tcp_continue_after_long_line() { - assert_source_compliance(&SOCKET_PUSH_SOURCE_TAGS, async { - let (tx, mut rx) = SourceSender::new_test(); - let addr = next_addr(); + /// A *terminated* over-long line is fatal too: even though its delimiter tells us exactly + /// where it ended and we could resync, the connection is reset. Asserted on the socket itself + /// rather than on event delivery — clearing the buffer would drop later events either way, so + /// only checking for EOF distinguishes "connection reset" from "frame skipped". + async fn tcp_resets_connection_after_long_line() { + use tokio::io::AsyncWriteExt as _; - let mut config = TcpConfig::from_address(addr.into()); - config.set_framing(Some( - NewlineDelimitedDecoderConfig::new_with_max_length(10).into(), - )); + let (tx, mut rx) = SourceSender::new_test(); + let addr = next_addr(); - let server = SocketConfig::from(config) - .build(SourceContext::new_test(tx, None)) - .await - .unwrap(); - tokio::spawn(server); + let mut config = TcpConfig::from_address(addr.into()); + config.set_framing(Some( + NewlineDelimitedDecoderConfig::new_with_max_length(10).into(), + )); - let lines = vec![ - "short".to_owned(), - "this is too long".to_owned(), - "more short".to_owned(), - ]; + let server = SocketConfig::from(config) + .build(SourceContext::new_test(tx, None)) + .await + .unwrap(); + tokio::spawn(server); + wait_for_tcp(addr).await; - wait_for_tcp(addr).await; - send_lines(addr, lines.into_iter()).await.unwrap(); + let mut stream = TcpStream::connect(addr).await.unwrap(); - let event = rx.next().await.unwrap(); - assert_eq!( - event.as_log()[log_schema().message_key().unwrap().to_string()], - "short".into() - ); + // A line within the limit is delivered as normal. + stream.write_all(b"short\n").await.unwrap(); + let event = rx.next().await.unwrap(); + assert_eq!( + event.as_log()[log_schema().message_key().unwrap().to_string()], + "short".into() + ); - let event = rx.next().await.unwrap(); - assert_eq!( - event.as_log()[log_schema().message_key().unwrap().to_string()], - "more short".into() - ); - }) - .await; + // A terminated line over the limit resets the connection. + stream.write_all(b"this is too long\n").await.unwrap(); + + let mut buf = [0u8; 1]; + let closed = matches!( + timeout(Duration::from_secs(10), stream.read(&mut buf)).await, + Ok(Ok(0)) | Ok(Err(_)) + ); + assert!( + closed, + "connection should have been reset by the over-long terminated line", + ); } #[tokio::test] From a68c71c8ad59aa529d37f3578b797e6f015a550a Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Fri, 7 Aug 2026 23:45:58 +0530 Subject: [PATCH 2/4] OBE-11566 - netflow framer: scope template caches per exporter --- lib/codecs/src/decoding/framing/netflow.rs | 268 ++++++++++++++++++++- src/sources/socket/tcp.rs | 11 +- src/sources/socket/udp.rs | 11 +- 3 files changed, 283 insertions(+), 7 deletions(-) diff --git a/lib/codecs/src/decoding/framing/netflow.rs b/lib/codecs/src/decoding/framing/netflow.rs index 66bd58a309..ed8f72cdce 100644 --- a/lib/codecs/src/decoding/framing/netflow.rs +++ b/lib/codecs/src/decoding/framing/netflow.rs @@ -1,5 +1,7 @@ use std::collections::BTreeMap; use std::io; +use std::net::IpAddr; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use byteorder::{ByteOrder, NetworkEndian}; @@ -19,8 +21,35 @@ use vrl::core::Value; use vrl::value::KeyString; use crate::decoding::BoxedFramingError; +use indexmap::IndexMap; use vector_config::configurable_component; +/// Maximum number of exporters whose templates are cached at once. +/// +/// The map is keyed by attacker-influenceable data (the source address of a datagram), so it needs +/// a ceiling of its own: without one, spraying spoofed source addresses grows it without bound. +/// When full the least recently used exporter is evicted, which costs that exporter a template +/// refresh rather than dropping its data permanently. +const MAX_TRACKED_EXPORTERS: usize = 1024; + +static NEXT_CONNECTION_ID: AtomicU64 = AtomicU64::new(0); + +/// Which template cache a decoder reads and writes. +/// +/// NetFlow v9/IPFIX templates are defined by the exporter and are only meaningful in its own +/// context (RFC 3954 section 5.2, RFC 7011 section 8): template id 260 from one exporter says +/// nothing about template id 260 from another. Sharing one cache across peers lets any host that +/// can reach the collector redefine the layout that another exporter's data records are decoded +/// with, so each scope gets its own parser. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +enum ExporterScope { + /// A datagram peer. Keyed on the address only, deliberately not the port: an exporter's source + /// port can change between datagrams, and including it would hide templates it already sent. + Datagram(IpAddr), + /// One stream connection. Templates live and die with the connection. + Connection(u64), +} + /// Config used to build a `NetflowDecoderDecoder`. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct NetflowDecoderConfig { @@ -70,7 +99,19 @@ impl NetflowDecoderOptions { pub struct NetflowDecoder { /// The maximum length of the byte buffer. pub max_length: usize, - parser: Arc>, + /// One stateful parser per exporter, shared across clones. UDP clones this decoder per + /// datagram, so the cache cannot live in the clone: templates arrive in one datagram and the + /// data records referencing them in later ones. + parsers: Arc>>, + /// Which entry of `parsers` this clone uses. Defaults to a fresh connection scope so a caller + /// that forgets to set it gets isolation rather than a shared cache. + scope: ExporterScope, +} + +#[derive(Debug)] +struct TrackedParser { + parser: NetflowParser, + last_used: u64, } impl NetflowDecoder { @@ -88,15 +129,77 @@ impl NetflowDecoder { pub fn new_with_max_length(max_length: usize) -> Self { Self { max_length, - parser: Arc::new(Mutex::new(NetflowParser::default())), + parsers: Arc::new(Mutex::new(IndexMap::new())), + scope: ExporterScope::Connection(NEXT_CONNECTION_ID.fetch_add(1, Ordering::Relaxed)), } } + /// Scopes this decoder's templates to a datagram peer. + /// + /// Call on the per-datagram clone before decoding, so each exporter resolves its data records + /// against templates it sent itself. + pub fn set_datagram_peer(&mut self, peer: IpAddr) { + self.scope = ExporterScope::Datagram(peer); + } + + /// Scopes this decoder's templates to a fresh connection. + /// + /// Call once per accepted connection. Without this every connection built from the same + /// configured decoder would inherit one scope and share a template cache. + pub fn set_new_connection_scope(&mut self) { + self.scope = ExporterScope::Connection(NEXT_CONNECTION_ID.fetch_add(1, Ordering::Relaxed)); + } + /// Returns the maximum frame length when decoding. pub const fn max_length(&self) -> usize { self.max_length } + /// Returns the parser for `scope`, creating it if this exporter has not been seen before and + /// evicting the least recently used exporter if the cache is full. + fn parser_for<'a>( + parsers: &'a mut IndexMap, + scope: &ExporterScope, + ) -> &'a mut NetflowParser { + // Monotonic tick rather than a clock: only the ordering matters, and this keeps the hot + // path free of a syscall. + static TICK: AtomicU64 = AtomicU64::new(0); + let now = TICK.fetch_add(1, Ordering::Relaxed); + + if !parsers.contains_key(scope) { + if parsers.len() >= MAX_TRACKED_EXPORTERS { + // Eviction is O(n) but only runs when the cache is full; lookups stay O(1). + if let Some((index, _, _)) = parsers + .iter() + .enumerate() + .min_by_key(|(_, (_, tracked))| tracked.last_used) + .map(|(index, (key, tracked))| (index, key.clone(), tracked.last_used)) + { + parsers.shift_remove_index(index); + warn!( + message = "Netflow template cache is full; evicted the least recently \ + used exporter, which must resend its templates.", + max_tracked_exporters = MAX_TRACKED_EXPORTERS, + internal_log_rate_limit = true + ); + } + } + parsers.insert( + scope.clone(), + TrackedParser { + parser: NetflowParser::default(), + last_used: now, + }, + ); + } + + let tracked = parsers + .get_mut(scope) + .expect("entry was just inserted if absent"); + tracked.last_used = now; + &mut tracked.parser + } + fn insert_v9_header_fields(v9pkt: &V9) -> BTreeMap { let mut pkt: BTreeMap = BTreeMap::new(); pkt.insert( @@ -216,7 +319,8 @@ impl Decoder for NetflowDecoder { } let mut packets = Vec::new(); - let mut parser = self.parser.lock().expect("Failed to lock NetflowParser"); + let mut parsers = self.parsers.lock().expect("Failed to lock NetflowParser"); + let parser = Self::parser_for(&mut parsers, &self.scope); let parse_results = parser.parse_bytes(src.as_mut()); let mut had_fatal_error = false; @@ -1003,4 +1107,162 @@ mod tests { let data_records = collect_records(&data_results); assert!(data_records.iter().any(|r| r["template_type"] == "data")); } + + + // ---- template scoping (OBE-11566) --------------------------------------------------------- + // + // Minimal hand-built v9 packets rather than base64 captures: what matters here is exactly + // which template id each exporter defined, which a capture blob would hide. + + /// v9 header: version, flowset count, uptime, epoch, sequence, source id. + fn v9_header(flowset_count: u16) -> Vec { + let mut p = Vec::new(); + p.extend_from_slice(&9u16.to_be_bytes()); + p.extend_from_slice(&flowset_count.to_be_bytes()); + p.extend_from_slice(&0u32.to_be_bytes()); // sys_up_time + p.extend_from_slice(&1u32.to_be_bytes()); // unix_secs + p.extend_from_slice(&0u32.to_be_bytes()); // sequence + p.extend_from_slice(&0u32.to_be_bytes()); // source_id + p + } + + /// Template flowset (flowset id 0) defining `template_id` as the given (field_type, length)s. + fn v9_template(template_id: u16, fields: &[(u16, u16)]) -> Vec { + let mut body = Vec::new(); + body.extend_from_slice(&template_id.to_be_bytes()); + body.extend_from_slice(&(fields.len() as u16).to_be_bytes()); + for (ty, len) in fields { + body.extend_from_slice(&ty.to_be_bytes()); + body.extend_from_slice(&len.to_be_bytes()); + } + let mut p = v9_header(1); + p.extend_from_slice(&0u16.to_be_bytes()); + p.extend_from_slice(&((body.len() + 4) as u16).to_be_bytes()); + p.extend_from_slice(&body); + p + } + + /// Data flowset referencing `template_id`. + fn v9_data(template_id: u16, record: &[u8]) -> Vec { + let mut p = v9_header(1); + p.extend_from_slice(&template_id.to_be_bytes()); + p.extend_from_slice(&((record.len() + 4) as u16).to_be_bytes()); + p.extend_from_slice(record); + p + } + + /// IPV4_SRC_ADDR then IPV4_DST_ADDR — what the legitimate exporter registers. + fn honest_fields() -> Vec<(u16, u16)> { + vec![(8, 4), (12, 4)] + } + + /// The same template id with a different layout — what a spoofing peer registers so the + /// victim's records decode against the wrong offsets. + fn poisoned_fields() -> Vec<(u16, u16)> { + vec![(2, 4), (1, 4)] // IN_PKTS, IN_BYTES + } + + const RECORD: [u8; 8] = [10, 0, 0, 9, 10, 0, 0, 8]; + + fn decode_from(decoder: &NetflowDecoder, peer: &str, packet: &[u8]) -> String { + let mut decoder = decoder.clone(); + decoder.set_datagram_peer(peer.parse().expect("valid ip")); + let mut buf = BytesMut::from(packet); + decoder + .decode(&mut buf) + .expect("decode should not fail") + .map(|b| String::from_utf8_lossy(&b).into_owned()) + .unwrap_or_default() + } + + /// The feature itself: a template learned from one datagram must still decode data records + /// arriving in a later datagram from the same exporter. Without this, "isolation" could be + /// satisfied trivially by never caching anything. + #[test] + fn templates_persist_across_datagrams_from_the_same_exporter() { + let decoder = NetflowDecoder::new(); + + decode_from(&decoder, "10.0.0.1", &v9_template(256, &honest_fields())); + let text = decode_from(&decoder, "10.0.0.1", &v9_data(256, &RECORD)); + + assert!( + text.contains("10.0.0.9"), + "data should decode against the template this exporter sent, got: {text}" + ); + } + + /// The finding: another peer redefining the same template id must not change how this + /// exporter's records are decoded. + #[test] + fn one_exporter_cannot_poison_another_exporters_template() { + let decoder = NetflowDecoder::new(); + + decode_from(&decoder, "10.0.0.1", &v9_template(256, &honest_fields())); + decode_from(&decoder, "10.0.0.2", &v9_template(256, &poisoned_fields())); + + let text = decode_from(&decoder, "10.0.0.1", &v9_data(256, &RECORD)); + assert!( + text.contains("10.0.0.9"), + "another exporter poisoned this exporter's template 256, got: {text}" + ); + } + + /// Templates must not leak the other way either: an exporter that never sent one must not + /// decode against a template another exporter happened to register. + #[test] + fn an_exporter_cannot_borrow_another_exporters_template() { + let decoder = NetflowDecoder::new(); + + decode_from(&decoder, "10.0.0.1", &v9_template(256, &honest_fields())); + + let text = decode_from(&decoder, "10.0.0.3", &v9_data(256, &RECORD)); + assert!( + !text.contains("10.0.0.9"), + "an exporter that sent no template decoded against another's, got: {text}" + ); + } + + /// Stream peers are isolated too, and each connection scope is distinct from every other. + #[test] + fn each_connection_scope_is_distinct() { + let decoder = NetflowDecoder::new(); + + let mut first = decoder.clone(); + first.set_new_connection_scope(); + let mut buf = BytesMut::from(&v9_template(256, &honest_fields())[..]); + first.decode(&mut buf).expect("template should decode"); + + let mut second = decoder.clone(); + second.set_new_connection_scope(); + let mut buf = BytesMut::from(&v9_data(256, &RECORD)[..]); + let text = second + .decode(&mut buf) + .expect("decode should not fail") + .map(|b| String::from_utf8_lossy(&b).into_owned()) + .unwrap_or_default(); + + assert!( + !text.contains("10.0.0.9"), + "a second connection reused the first connection's template, got: {text}" + ); + } + + /// The cache is keyed on the datagram source address, which an attacker controls, so it needs + /// a ceiling of its own. + #[test] + fn exporter_cache_is_bounded() { + let decoder = NetflowDecoder::new(); + + for i in 0..(MAX_TRACKED_EXPORTERS + 50) { + let peer = format!("10.{}.{}.{}", (i >> 16) & 0xff, (i >> 8) & 0xff, i & 0xff); + decode_from(&decoder, &peer, &v9_template(256, &honest_fields())); + } + + let tracked = decoder.parsers.lock().expect("lock").len(); + assert!( + tracked <= MAX_TRACKED_EXPORTERS, + "cache grew to {tracked} entries, above the {MAX_TRACKED_EXPORTERS} ceiling", + ); + } + } diff --git a/src/sources/socket/tcp.rs b/src/sources/socket/tcp.rs index 1aea8b0332..9cdf9d12bc 100644 --- a/src/sources/socket/tcp.rs +++ b/src/sources/socket/tcp.rs @@ -4,7 +4,7 @@ use vector_lib::ipallowlist::IpAllowlistConfig; use chrono::Utc; use serde_with::serde_as; use smallvec::SmallVec; -use vector_lib::codecs::decoding::{DeserializerConfig, FramingConfig}; +use vector_lib::codecs::decoding::{DeserializerConfig, Framer, FramingConfig}; use vector_lib::config::{LegacyKey, LogNamespace}; use vector_lib::configurable::configurable_component; use vector_lib::lookup::{lookup_v2::OptionalValuePath, owned_value_path, path}; @@ -213,7 +213,14 @@ impl TcpSource for RawTcpSource { type Acker = TcpNullAcker; fn decoder(&self) -> Self::Decoder { - self.decoder.clone() + // Called once per accepted connection. Netflow templates are scoped to the connection so + // one peer cannot redefine the template ids another peer's records decode against; + // without this every connection would inherit the configured decoder's single scope. + let mut decoder = self.decoder.clone(); + if let Framer::Netflow(netflow) = &mut decoder.framer { + netflow.set_new_connection_scope(); + } + decoder } fn handle_events(&self, events: &mut [Event], host: std::net::SocketAddr) { diff --git a/src/sources/socket/udp.rs b/src/sources/socket/udp.rs index 173f7ad034..120a5607c6 100644 --- a/src/sources/socket/udp.rs +++ b/src/sources/socket/udp.rs @@ -6,7 +6,7 @@ use ipnet::IpNet; use listenfd::ListenFd; use tokio_util::codec::FramedRead; use vector_lib::codecs::{ - decoding::{DeserializerConfig, FramingConfig}, + decoding::{DeserializerConfig, Framer, FramingConfig}, StreamDecodingError, }; use vector_lib::configurable::configurable_component; @@ -219,7 +219,14 @@ pub(super) fn udp( bytes_received.emit(ByteSize(byte_size)); let payload = buf.split_to(byte_size); let truncated = byte_size == max_length + 1; - let mut stream = FramedRead::new(payload.as_ref(), decoder.clone()).peekable(); + // Scope netflow templates to the sending exporter. Without this every peer + // shares one template cache, so any host able to reach this port could + // redefine the template ids another exporter's data records decode against. + let mut datagram_decoder = decoder.clone(); + if let Framer::Netflow(netflow) = &mut datagram_decoder.framer { + netflow.set_datagram_peer(address.ip()); + } + let mut stream = FramedRead::new(payload.as_ref(), datagram_decoder).peekable(); while let Some(result) = stream.next().await { let last = Pin::new(&mut stream).peek().await.is_none(); From 2ec9893f331830b8e096adb35d5cab29be0c803a Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Sat, 8 Aug 2026 00:36:19 +0530 Subject: [PATCH 3/4] OBE-11567 - json_paths decoder: give each stream its own parser and recover after a bad frame --- lib/codecs/src/decoding/format/json_paths.rs | 200 +++++++++++++++++-- 1 file changed, 185 insertions(+), 15 deletions(-) diff --git a/lib/codecs/src/decoding/format/json_paths.rs b/lib/codecs/src/decoding/format/json_paths.rs index 4e78bffa9b..a10dc314e1 100644 --- a/lib/codecs/src/decoding/format/json_paths.rs +++ b/lib/codecs/src/decoding/format/json_paths.rs @@ -185,14 +185,36 @@ pub struct JsonPathDeserializerOptions { /// /// This deserializer maintains parser state across calls to support streaming, /// allowing JSON to be split across multiple `parse()` calls. -#[derive(Clone)] pub struct JsonPathDeserializer { path_map: BTreeMap, lossy: bool, - /// Streaming parser state (wrapped in Arc for interior mutability) + feeder_capacity: usize, + /// Streaming parser state (wrapped in Arc for interior mutability). + /// + /// A JSON document may span several frames, so this has to persist between `parse` calls — + /// but only within one stream. See the `Clone` impl. parser_state: std::sync::Arc>, } +/// Cloning yields an independent parser, not a second handle to the same one. +/// +/// `Decoder` is cloned once per accepted connection (and per datagram for message-based sources), +/// so sharing the `Arc` would feed every client's bytes into a single streaming document: one +/// client's frame could complete another's half-open object, and one malformed frame would leave +/// the shared parser in an error state that no later frame from any client could recover from. +impl Clone for JsonPathDeserializer { + fn clone(&self) -> Self { + Self { + path_map: self.path_map.clone(), + lossy: self.lossy, + feeder_capacity: self.feeder_capacity, + parser_state: std::sync::Arc::new(std::sync::Mutex::new(Self::new_parser_state( + self.feeder_capacity, + ))), + } + } +} + /// State maintained across parse() calls for streaming struct StreamingParserState { parser: JsonParser, @@ -202,19 +224,29 @@ struct StreamingParserState { impl JsonPathDeserializer { /// Creates a new `JsonPathDeserializer`. pub fn new(config: PathOperationConfig, options: JsonPathDeserializerOptions) -> Self { - let feeder = PushJsonFeeder::with_capacity(options.feeder_capacity); - let parser_options = JsonParserOptionsBuilder::default() - .with_streaming(true) - .build(); - let parser = JsonParser::new_with_options(feeder, parser_options); - Self { path_map: config.as_map(), lossy: options.lossy, - parser_state: std::sync::Arc::new(std::sync::Mutex::new(StreamingParserState { - parser, - state: ParserState::new(), - })), + feeder_capacity: options.feeder_capacity, + parser_state: std::sync::Arc::new(std::sync::Mutex::new(Self::new_parser_state( + options.feeder_capacity, + ))), + } + } + + /// Builds a parser with no residual state. + /// + /// Used for a new deserializer, for each clone, and to recover after a parse error — actson's + /// push parser cannot resynchronise once it reports a syntax error, so the only way back is a + /// new parser. + fn new_parser_state(feeder_capacity: usize) -> StreamingParserState { + let feeder = PushJsonFeeder::with_capacity(feeder_capacity); + let parser_options = JsonParserOptionsBuilder::default() + .with_streaming(true) + .build(); + StreamingParserState { + parser: JsonParser::new_with_options(feeder, parser_options), + state: ParserState::new(), } } } @@ -290,24 +322,48 @@ impl Deserializer for JsonPathDeserializer { .lock() .unwrap_or_else(|poison| poison.into_inner()); + // Any failure below leaves actson in an error state and `ParserState.path` / + // `value_stack` holding fragments of the half-parsed document. actson's push parser cannot + // resynchronise after a syntax error, and `Error::ParsingError` is continuable, so without + // this the source would keep calling a permanently wrecked parser: every later frame on + // this stream fails or decodes against stale path residue. Discard the state so the next + // frame starts from a clean parser. + let result = self.parse_frame(&mut streaming_state, &bytes_slice, log_namespace); + if result.is_err() { + *streaming_state = Self::new_parser_state(self.feeder_capacity); + } + result + } +} + +impl JsonPathDeserializer { + fn parse_frame( + &self, + streaming_state: &mut StreamingParserState, + bytes_slice: &[u8], + log_namespace: LogNamespace, + ) -> vector_common::Result> { let mut byte_offset = 0usize; let total_len = bytes_slice.len(); while byte_offset < total_len { if streaming_state.parser.feeder.is_full() { - self.drain_parser_events(&mut streaming_state)?; + self.drain_parser_events(streaming_state)?; if streaming_state.parser.feeder.is_full() { return Err("JSON feeder is full and cannot accept more bytes".into()); } } - let pushed = streaming_state.parser.feeder.push_bytes(&bytes_slice[byte_offset..]); + let pushed = streaming_state + .parser + .feeder + .push_bytes(&bytes_slice[byte_offset..]); if pushed == 0 { return Err("JSON feeder could not accept bytes (0 bytes pushed)".into()); } byte_offset += pushed; } - self.drain_parser_events(&mut streaming_state)?; + self.drain_parser_events(streaming_state)?; let mut result = SmallVec::new(); let events_to_emit = std::mem::take(&mut streaming_state.state.events); @@ -1290,4 +1346,118 @@ codec = "json_paths" assert_eq!(arr[3], Value::Integer(4)); assert_eq!(arr[4], Value::Integer(5)); } + + // ---- parser isolation and recovery (OBE-11567) -------------------------------------------- + + fn test_deserializer() -> JsonPathDeserializer { + let config = PathOperationConfig::new(owned_value_path!("meta"), PathOperation::Identity); + JsonPathDeserializer::new(config, JsonPathDeserializerOptions::default()) + } + + fn parse_ok(d: &JsonPathDeserializer, raw: &str) -> usize { + d.parse(Bytes::from(raw.to_owned()), LogNamespace::Vector) + .expect("should parse") + .len() + } + + /// A clone is a separate parser, not another handle to the same one. + /// + /// `Decoder` is cloned per connection, so a shared parser would let one client's bytes land in + /// the middle of another client's document. + #[test] + fn clones_do_not_share_parser_state() { + let first = test_deserializer(); + let second = first.clone(); + + // `first` is left mid-document, with an unterminated object. + assert_eq!(parse_ok(&first, r#"{"meta": {"source": "#), 0); + + // `second` must see a clean parser: a whole document decodes normally. + assert_eq!( + parse_ok(&second, r#"{"meta": {"source": "foo"}}"#), + 1, + "a clone inherited another parser's half-open document" + ); + } + + /// The mirror of the above: one client's bytes must not be able to complete another's + /// half-open object, which would forge an event out of two senders' data. + #[test] + fn a_clone_cannot_complete_another_clones_document() { + let victim = test_deserializer(); + let attacker = victim.clone(); + + assert_eq!(parse_ok(&victim, r#"{"meta": {"source": "#), 0); + + // On a shared parser these bytes would close the victim's half-open object and emit an + // event stitched from two senders. On its own the fragment is not valid JSON, so an error + // is the expected outcome; what matters is that no event is produced either way. + let emitted = attacker + .parse(Bytes::from(r#""spoofed"}}"#), LogNamespace::Vector) + .map(|events| events.len()) + .unwrap_or(0); + assert_eq!( + emitted, 0, + "one clone completed another clone's document, forging an event" + ); + } + + /// A malformed frame must not wedge the parser: actson cannot resynchronise after a syntax + /// error, and parse errors are continuable, so without a reset every later frame on this + /// stream would fail forever. + #[test] + fn parser_recovers_after_a_malformed_frame() { + let deserializer = test_deserializer(); + + assert_eq!(parse_ok(&deserializer, r#"{"meta": {"source": "foo"}}"#), 1); + + assert!( + deserializer + .parse(Bytes::from("xxx not json xxx"), LogNamespace::Vector) + .is_err(), + "a malformed frame should be reported as an error" + ); + + assert_eq!( + parse_ok(&deserializer, r#"{"meta": {"source": "bar"}}"#), + 1, + "the parser stayed wedged after a malformed frame" + ); + } + + /// Residue from an abandoned document must not leak into the next one — a disconnect + /// mid-object leaves `path` and `value_stack` populated. + #[test] + fn abandoned_document_does_not_corrupt_the_next_one() { + let deserializer = test_deserializer(); + + assert_eq!(parse_ok(&deserializer, r#"{"meta": {"source": "#), 0); + assert!(deserializer + .parse(Bytes::from("!!!"), LogNamespace::Vector) + .is_err()); + + let events = deserializer + .parse( + Bytes::from(r#"{"meta": {"source": "clean"}}"#), + LogNamespace::Vector, + ) + .expect("should parse after the reset"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].as_log()["expr"], "meta".into()); + } + + /// The feature must survive the fix: a document split across frames still decodes within one + /// parser. Isolation must not be achieved by simply resetting on every call. + #[test] + fn a_document_split_across_frames_still_decodes() { + let deserializer = test_deserializer(); + + assert_eq!(parse_ok(&deserializer, r#"{"meta": {"sou"#), 0); + assert_eq!( + parse_ok(&deserializer, r#"rce": "foo"}}"#), + 1, + "streaming across frames broke" + ); + } + } From 5d187a3565a410ca67ff7677bd0385e3ae382d96 Mon Sep 17 00:00:00 2001 From: Harshvardhan Shrivastava Date: Sat, 8 Aug 2026 00:48:58 +0530 Subject: [PATCH 4/4] OBE-11568 - influxdb decoder: bound the tag amplification of a single line --- lib/codecs/src/decoding/format/influxdb.rs | 204 ++++++++++++++++----- 1 file changed, 159 insertions(+), 45 deletions(-) diff --git a/lib/codecs/src/decoding/format/influxdb.rs b/lib/codecs/src/decoding/format/influxdb.rs index 2b059e2160..badc80bb3b 100644 --- a/lib/codecs/src/decoding/format/influxdb.rs +++ b/lib/codecs/src/decoding/format/influxdb.rs @@ -84,6 +84,20 @@ impl InfluxdbDeserializer { } } +/// Maximum number of (tag, value) pairs one line may materialise across all of its metrics. +/// +/// A line with T tags and F fields produces F metrics that each own a full copy of the T tags, so +/// the memory it costs is T x F -- quadratic in the length of a single line. Hoisting the tag map +/// out of the field loop removes the repeated string conversions but not this: `Metric::with_tags` +/// takes `MetricTags` by value, so every metric genuinely owns its tags. Only a ceiling on the +/// product bounds it. +/// +/// The frame length cap bounds a line to 100 KiB by default, which still leaves room for roughly +/// 12k tags x 12k fields -- on the order of 10 GB once materialised -- so the frame cap alone does +/// not close this. Real line protocol carries a handful of tags and at most a few hundred fields, +/// so this ceiling is far above legitimate traffic. +const MAX_TAG_ENTRIES_PER_LINE: usize = 100_000; + impl Deserializer for InfluxdbDeserializer { fn parse( &self, @@ -96,51 +110,62 @@ impl Deserializer for InfluxdbDeserializer { }; let parsed_line = influxdb_line_protocol::parse_lines(&line); - let res = parsed_line - .collect::, _>>()? - .iter() - .flat_map(|line| { - let ParsedLine { - series, - field_set, - timestamp, - } = line; - - field_set - .iter() - .filter_map(|f| { - let measurement = series.measurement.clone(); - let tags = series.tag_set.as_ref(); - let val = match f.1 { - FieldValue::I64(v) => v as f64, - FieldValue::U64(v) => v as f64, - FieldValue::F64(v) => v, - FieldValue::Boolean(v) => { - if v { - 1.0 - } else { - 0.0 - } - } - FieldValue::String(_) => return None, // String values cannot be modelled in our schema - }; - Some(Event::Metric( - Metric::new( - format!("{0}_{1}", measurement, f.0), - MetricKind::Absolute, - MetricValue::Gauge { value: val }, - ) - .with_tags(tags.map(|ts| { - MetricTags::from_iter( - ts.iter().map(|t| (t.0.to_string(), t.1.to_string())), - ) - })) - .with_timestamp(timestamp.map(DateTime::from_timestamp_nanos)), - )) - }) - .collect::>() - }) - .collect(); + let mut res: SmallVec<[Event; 1]> = SmallVec::new(); + + for line in parsed_line.collect::, _>>()? { + let ParsedLine { + series, + field_set, + timestamp, + } = &line; + + // Checked before anything is materialised: the point is never to allocate the + // quadratic amount in the first place. + let tag_count = series.tag_set.as_ref().map_or(0, |ts| ts.len()); + let tag_entries = tag_count.saturating_mul(field_set.len()); + if tag_entries > MAX_TAG_ENTRIES_PER_LINE { + return Err(format!( + "influxdb line would expand to {tag_entries} tag entries ({tag_count} tags \ + x {} fields), above the limit of {MAX_TAG_ENTRIES_PER_LINE}", + field_set.len() + ) + .into()); + } + + // Built once per line rather than once per field. Each emitted metric still owns its + // own copy, since `Metric::with_tags` takes `MetricTags` by value, but this stops the + // tags being re-converted from `&str` for every single field. + let tags = series.tag_set.as_ref().map(|ts| { + MetricTags::from_iter(ts.iter().map(|t| (t.0.to_string(), t.1.to_string()))) + }); + + for f in field_set.iter() { + let val = match f.1 { + FieldValue::I64(v) => v as f64, + FieldValue::U64(v) => v as f64, + FieldValue::F64(v) => v, + FieldValue::Boolean(v) => { + if v { + 1.0 + } else { + 0.0 + } + } + // String values cannot be modelled in our schema + FieldValue::String(_) => continue, + }; + + res.push(Event::Metric( + Metric::new( + format!("{0}_{1}", series.measurement, f.0), + MetricKind::Absolute, + MetricValue::Gauge { value: val }, + ) + .with_tags(tags.clone()) + .with_timestamp(timestamp.map(DateTime::from_timestamp_nanos)), + )); + } + } Ok(res) } @@ -209,4 +234,93 @@ mod tests { let buffer = Bytes::from("some invalid string"); assert!(deser.parse(buffer, LogNamespace::default()).is_err()); } + + // ---- amplification cap (OBE-11568) -------------------------------------------------------- + + /// Builds a line with `tags` tags and `fields` fields. + fn line_with(tags: usize, fields: usize) -> Bytes { + let mut line = String::from("m"); + for i in 0..tags { + line.push_str(&format!(",t{i}=v")); + } + line.push(' '); + for i in 0..fields { + if i > 0 { + line.push(','); + } + line.push_str(&format!("f{i}=1i")); + } + Bytes::from(line) + } + + /// Every field becomes a metric owning a full copy of the line's tags, so one line costs + /// tags x fields. A line whose product is over the ceiling must be refused before any of it is + /// materialised. + #[test] + fn line_over_the_tag_entry_limit_is_rejected() { + let deser = InfluxdbDeserializer::new(true); + + // 400 x 400 = 160,000 entries, above the 100,000 ceiling. + let error = deser + .parse(line_with(400, 400), LogNamespace::default()) + .expect_err("an over-amplifying line must be rejected"); + + let text = error.to_string(); + assert!( + text.contains("tag entries") && text.contains("100000"), + "error should name the limit, got: {text}" + ); + } + + /// The ceiling must sit far above real line protocol: a wide-but-ordinary line still decodes. + #[test] + fn ordinary_wide_line_is_still_accepted() { + let deser = InfluxdbDeserializer::new(true); + + // 20 tags x 200 fields = 4,000 entries — generous for real telemetry, well under the cap. + let events = deser + .parse(line_with(20, 200), LogNamespace::default()) + .expect("an ordinary wide line must still decode"); + + assert_eq!(events.len(), 200); + let tags = events[0].as_metric().tags().expect("tags"); + assert_eq!(tags.iter_all().count(), 20); + } + + /// A line sitting just under the ceiling is accepted, pinning the boundary so the check cannot + /// drift into rejecting legitimate traffic. + #[test] + fn line_just_under_the_limit_is_accepted() { + let deser = InfluxdbDeserializer::new(true); + + // 100 x 1000 = 100,000 entries, exactly the ceiling. + let events = deser + .parse(line_with(100, 1000), LogNamespace::default()) + .expect("a line exactly at the limit must be accepted"); + assert_eq!(events.len(), 1000); + } + + /// Tags are built once per line and cloned into each metric, so every metric must still carry + /// the complete tag set — the hoist must not have changed what is emitted. + #[test] + fn every_metric_carries_the_full_tag_set() { + let deser = InfluxdbDeserializer::new(true); + + let events = deser + .parse( + Bytes::from("cpu,host=A,region=west a=1i,b=2i,c=3i"), + LogNamespace::default(), + ) + .expect("should parse"); + + assert_eq!(events.len(), 3); + for event in &events { + let tags = event + .as_metric() + .tags() + .expect("every metric keeps its tags"); + assert_eq!(tags.get("host"), Some("A")); + assert_eq!(tags.get("region"), Some("west")); + } + } }