vector/sources/util/http/
encoding.rs

1use bytes::{Buf, BufMut, Bytes, BytesMut};
2use futures_util::StreamExt;
3use snap::raw::Decoder as SnappyDecoder;
4use warp::http::StatusCode;
5use warp::{Filter, filters::BoxedFilter};
6
7#[cfg(test)]
8use crate::sources::util::decompression::DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES;
9pub use crate::sources::util::decompression::set_max_decompressed_size_bytes;
10use crate::sources::util::decompression::{
11    CappedDecoder, is_decompressed_size_limit_error, max_decompressed_size_bytes,
12};
13use crate::{common::http::ErrorMessage, internal_events::HttpDecompressError};
14
15/// Collects a request body into [`Bytes`] while enforcing an in-memory size cap.
16///
17/// The cap is the global decompressed-size limit ([`max_decompressed_size_bytes`]): it bounds the
18/// raw (still-compressed) body a source buffers before decompression, so a large upload cannot
19/// drive unbounded allocation independently of the decompressed-size cap.
20pub(crate) fn capped_body() -> BoxedFilter<(Bytes,)> {
21    let max_body_size = max_decompressed_size_bytes();
22    let max_body_size_header = u64::try_from(max_body_size).unwrap_or(u64::MAX);
23
24    warp::header::optional::<u64>("content-length")
25        .and_then(move |declared: Option<u64>| async move {
26            if declared.is_some_and(|len| len > max_body_size_header) {
27                Err(warp::reject::custom(request_body_too_large_error(
28                    max_body_size,
29                )))
30            } else {
31                Ok(())
32            }
33        })
34        .untuple_one()
35        .and(warp::body::stream())
36        .and_then(move |body| async move {
37            collect_body_with_limit(body, max_body_size)
38                .await
39                .map_err(warp::reject::custom)
40        })
41        .boxed()
42}
43
44/// Decompresses the body based on the Content-Encoding header.
45///
46/// Supports gzip, deflate, snappy, zstd, and identity (no compression).
47///
48/// Caps the decompressed output at 100 MiB to mitigate decompression-bomb DoS attacks.
49pub fn decompress_body(header: Option<&str>, body: Bytes) -> Result<Bytes, ErrorMessage> {
50    decompress_body_with_limit(header, body, max_decompressed_size_bytes())
51}
52
53/// Like [`decompress_body`], but allows the caller to control the decompressed size cap.
54pub(crate) fn decompress_body_with_limit(
55    header: Option<&str>,
56    mut body: Bytes,
57    max_decompressed_size: usize,
58) -> Result<Bytes, ErrorMessage> {
59    if let Some(encodings) = header {
60        for encoding in encodings.rsplit(',').map(str::trim) {
61            body = match encoding {
62                "identity" => body,
63                "gzip" => CappedDecoder::gzip_with_limit(body.reader(), max_decompressed_size)
64                    .decompress()
65                    .map(Bytes::from)
66                    .map_err(|error| {
67                        emit_decompress_error(encoding, error, max_decompressed_size)
68                    })?,
69                "deflate" => CappedDecoder::zlib_with_limit(body.reader(), max_decompressed_size)
70                    .decompress()
71                    .map(Bytes::from)
72                    .map_err(|error| {
73                        emit_decompress_error(encoding, error, max_decompressed_size)
74                    })?,
75                "snappy" => decompress_snappy(&body, max_decompressed_size)?,
76                "zstd" => CappedDecoder::zstd_http_with_limit(body.reader(), max_decompressed_size)
77                    .map_err(|error| emit_decompress_error(encoding, error, max_decompressed_size))?
78                    .decompress()
79                    .map(Bytes::from)
80                    .map_err(|error| {
81                        emit_decompress_error(encoding, error, max_decompressed_size)
82                    })?,
83                encoding => {
84                    return Err(ErrorMessage::new(
85                        StatusCode::UNSUPPORTED_MEDIA_TYPE,
86                        format!("Unsupported encoding {encoding}"),
87                    ));
88                }
89            }
90        }
91    }
92
93    ensure_body_within_limit(&body, "identity", max_decompressed_size)?;
94    Ok(body)
95}
96
97fn decompress_snappy(body: &Bytes, max_decompressed_size: usize) -> Result<Bytes, ErrorMessage> {
98    // Snappy stores the decompressed length in the frame header, so reject oversized
99    // payloads before allocating the output buffer.
100    let len = snap::raw::decompress_len(body).map_err(|error| {
101        emit_decompress_error(
102            "snappy",
103            std::io::Error::other(error),
104            max_decompressed_size,
105        )
106    })?;
107    if len > max_decompressed_size {
108        return Err(decompressed_too_large_error(
109            "snappy",
110            max_decompressed_size,
111        ));
112    }
113    let decoded = SnappyDecoder::new().decompress_vec(body).map_err(|error| {
114        emit_decompress_error(
115            "snappy",
116            std::io::Error::other(error),
117            max_decompressed_size,
118        )
119    })?;
120    Ok(decoded.into())
121}
122
123/// Spare capacity added to the initial buffer so a third or later chunk can be appended without reallocating right away.
124const ADDITIONAL_CAPACITY_FOR_CHUNKS_BEYOND_FIRST_TWO: usize = 16 * 1024;
125
126/// Collects the body into [`Bytes`] under `max_body_size`, mirroring the fast
127/// paths of hyper `to_bytes`. Single-chunk bodies avoid the `BytesMut`
128/// allocation: a buffer sized for both chunks plus an arbitrary 16 KiB (to try
129/// to avoid having to reallocate multiple times once other chunks arrive) is
130/// only allocated once a second chunk arrives.
131/// (<https://github.com/hyperium/hyper/blob/v0.14.32/src/body/to_bytes.rs>).
132async fn collect_body_with_limit<S, B>(body: S, max_body_size: usize) -> Result<Bytes, ErrorMessage>
133where
134    S: futures_util::Stream<Item = Result<B, warp::Error>>,
135    B: Buf,
136{
137    futures_util::pin_mut!(body);
138
139    let mut total_body_size: usize = 0;
140    let mut admit_chunk_within_limit = |chunk: Result<B, warp::Error>| -> Result<B, ErrorMessage> {
141        let chunk = chunk.map_err(|error| {
142            ErrorMessage::new(
143                StatusCode::BAD_REQUEST,
144                format!("Failed reading request body: {error}"),
145            )
146        })?;
147
148        total_body_size = total_body_size.saturating_add(chunk.remaining());
149        if total_body_size > max_body_size {
150            return Err(request_body_too_large_error(max_body_size));
151        }
152
153        Ok(chunk)
154    };
155
156    let Some(chunk) = body.next().await else {
157        return Ok(Bytes::new());
158    };
159    let mut first = admit_chunk_within_limit(chunk)?;
160
161    let Some(chunk) = body.next().await else {
162        return Ok(first.copy_to_bytes(first.remaining()));
163    };
164    let second = admit_chunk_within_limit(chunk)?;
165
166    let mut bytes = BytesMut::with_capacity(
167        first.remaining() + second.remaining() + ADDITIONAL_CAPACITY_FOR_CHUNKS_BEYOND_FIRST_TWO,
168    );
169    bytes.put(first);
170    bytes.put(second);
171
172    while let Some(chunk) = body.next().await {
173        bytes.put(admit_chunk_within_limit(chunk)?);
174    }
175
176    Ok(bytes.freeze())
177}
178
179fn ensure_body_within_limit(
180    body: &Bytes,
181    encoding: &str,
182    max_decompressed_size: usize,
183) -> Result<(), ErrorMessage> {
184    if body.len() > max_decompressed_size {
185        return Err(decompressed_too_large_error(
186            encoding,
187            max_decompressed_size,
188        ));
189    }
190    Ok(())
191}
192
193fn request_body_too_large_error(max: usize) -> ErrorMessage {
194    ErrorMessage::new(
195        StatusCode::PAYLOAD_TOO_LARGE,
196        format!("Request body exceeds limit of {max} bytes."),
197    )
198}
199
200fn decompressed_too_large_error(encoding: &str, max: usize) -> ErrorMessage {
201    ErrorMessage::new(
202        StatusCode::PAYLOAD_TOO_LARGE,
203        format!("Decompressed {encoding} body exceeds limit of {max} bytes."),
204    )
205}
206
207/// Maps a decompression failure to a response. If `error` is a [`DecompressedSizeLimitExceeded`]
208/// (the decompressed output exceeded the configured size cap), it becomes a `413 Payload Too
209/// Large` reporting `max_decompressed_size` (the cap that was actually enforced), matching the
210/// request-body and snappy size errors. Any other decode failure emits an `HttpDecompressError`
211/// event and becomes a `422 Unprocessable Entity`.
212///
213/// Callers whose error is not already an [`std::io::Error`] (e.g. snappy) wrap it via
214/// [`std::io::Error::other`].
215///
216/// [`DecompressedSizeLimitExceeded`]: crate::sources::util::decompression::DecompressedSizeLimitExceeded
217pub fn emit_decompress_error(
218    encoding: &str,
219    error: std::io::Error,
220    max_decompressed_size: usize,
221) -> ErrorMessage {
222    if is_decompressed_size_limit_error(&error) {
223        return decompressed_too_large_error(encoding, max_decompressed_size);
224    }
225    emit!(HttpDecompressError {
226        encoding,
227        error: &error
228    });
229    ErrorMessage::new(
230        StatusCode::UNPROCESSABLE_ENTITY,
231        format!("Failed decompressing payload with {encoding} decoder."),
232    )
233}
234
235#[cfg(test)]
236mod tests {
237    use std::io::Write;
238
239    use flate2::{Compression, write::GzEncoder};
240    use futures_util::stream;
241    use zstd::stream::Encoder as ZstdEncoder;
242
243    use super::*;
244
245    fn gzip_payload(plaintext: &[u8]) -> Bytes {
246        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
247        encoder.write_all(plaintext).unwrap();
248        encoder.finish().unwrap().into()
249    }
250
251    fn zstd_payload_with_window_log(plaintext: &[u8], window_log: u32) -> Bytes {
252        let mut encoder = ZstdEncoder::new(Vec::new(), 0).unwrap();
253        encoder.window_log(window_log).unwrap();
254        encoder.write_all(plaintext).unwrap();
255        encoder.finish().unwrap().into()
256    }
257
258    #[test]
259    fn gzip_within_limit_succeeds() {
260        let plaintext = vec![0u8; 10_000];
261        let body = gzip_payload(&plaintext);
262
263        let decoded = decompress_body_with_limit(Some("gzip"), body, 100_000).unwrap();
264        assert_eq!(decoded.len(), plaintext.len());
265    }
266
267    #[test]
268    fn gzip_exceeding_limit_returns_413() {
269        // Compress 1 MB of zeros, then cap at 1 KB.
270        let plaintext = vec![0u8; 1_000_000];
271        let body = gzip_payload(&plaintext);
272
273        let err = decompress_body_with_limit(Some("gzip"), body, 1024).expect_err("should reject");
274        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
275    }
276
277    #[test]
278    fn snappy_exceeding_limit_returns_413_before_allocating() {
279        // 2 MB of zeros. Snappy keeps the embedded length in the frame header.
280        let plaintext = vec![0u8; 2 * 1024 * 1024];
281        let compressed = snap::raw::Encoder::new().compress_vec(&plaintext).unwrap();
282
283        let err = decompress_body_with_limit(Some("snappy"), compressed.into(), 1024)
284            .expect_err("should reject");
285        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
286    }
287
288    #[test]
289    fn zstd_exceeding_limit_returns_413() {
290        let plaintext = vec![0u8; 10_000];
291        let compressed = zstd_payload_with_window_log(plaintext.as_slice(), 10);
292
293        let err =
294            decompress_body_with_limit(Some("zstd"), compressed, 1024).expect_err("should reject");
295        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
296    }
297
298    #[test]
299    fn identity_passes_through() {
300        let body: Bytes = Bytes::from_static(b"hello world");
301        let decoded = decompress_body(Some("identity"), body.clone()).unwrap();
302        assert_eq!(decoded, body);
303    }
304
305    #[test]
306    fn identity_exceeding_limit_returns_413() {
307        let body = Bytes::from_static(b"hello world");
308
309        let err = decompress_body_with_limit(Some("identity"), body, 5).expect_err("should reject");
310        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
311    }
312
313    #[test]
314    fn missing_content_encoding_exceeding_limit_returns_413() {
315        let body = Bytes::from_static(b"hello world");
316
317        let err = decompress_body_with_limit(None, body, 5).expect_err("should reject");
318        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
319    }
320
321    #[test]
322    fn zstd_window_log_tracks_limit() {
323        use crate::sources::util::decompression::zstd_window_log_max;
324        // Protocol-neutral: the window tracks the decompressed cap (used by gRPC/OTLP, which
325        // RFC 9659 does not govern).
326        // A zero cap maps to the minimum window log so the allocation guard is never disabled.
327        assert_eq!(zstd_window_log_max(0), Some(10));
328        assert_eq!(zstd_window_log_max(1), Some(10));
329        assert_eq!(zstd_window_log_max(1024), Some(10));
330        assert_eq!(zstd_window_log_max(1025), Some(11));
331        assert_eq!(
332            zstd_window_log_max(DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES),
333            Some(27)
334        );
335    }
336
337    #[test]
338    fn http_zstd_window_log_clamps_to_rfc9659_ceiling() {
339        use crate::sources::util::decompression::http_zstd_window_log_max;
340        // Below the 8 MB (2^23) ceiling the HTTP window still tracks the cap.
341        assert_eq!(http_zstd_window_log_max(1024), Some(10));
342        assert_eq!(http_zstd_window_log_max(8 * 1024 * 1024), Some(23));
343        // At or above the ceiling it clamps to 2^23 instead of tracking the cap, per RFC 9659.
344        assert_eq!(http_zstd_window_log_max(16 * 1024 * 1024), Some(23));
345        assert_eq!(
346            http_zstd_window_log_max(DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES),
347            Some(23)
348        );
349    }
350
351    #[tokio::test]
352    async fn collect_body_with_limit_succeeds_within_limit() {
353        let body = stream::iter([
354            Ok::<_, warp::Error>(Bytes::from_static(b"hello")),
355            Ok::<_, warp::Error>(Bytes::from_static(b" world")),
356        ]);
357
358        let collected = collect_body_with_limit(body, 11).await.unwrap();
359        assert_eq!(collected, Bytes::from_static(b"hello world"));
360    }
361
362    #[tokio::test]
363    async fn collect_body_with_limit_rejects_oversized_stream() {
364        let body = stream::iter([
365            Ok::<_, warp::Error>(Bytes::from_static(b"hello")),
366            Ok::<_, warp::Error>(Bytes::from_static(b" world")),
367        ]);
368
369        let err = collect_body_with_limit(body, 5)
370            .await
371            .expect_err("should reject");
372        assert_eq!(err.status_code(), StatusCode::PAYLOAD_TOO_LARGE);
373    }
374}
vector/sources/util/http/encoding.rs

vector/sources/util/http/
encoding.rs
