diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e90d164ee4..107bf89aadf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ - Reshape Nintendo Switch crashes so the issue title falls back to the crashing function instead of the raw abort result code, and raise their level to `fatal`. ([#6253](https://github.com/getsentry/relay/pull/6253)) - Reject Nintendo Switch dying message attachments with an invalid magic number instead of panicking on a short payload. ([#6253](https://github.com/getsentry/relay/pull/6253)) - Downgrade Kafka to prevent producers from getting stuck. ([#6336](https://github.com/getsentry/relay/pull/6336)) +- Use async instead of sync decompression in minidump endpoint. ([#6341](https://github.com/getsentry/relay/pull/6341)) - Prevent memory bomb in PII processor's `split_chunks`. ([#6343](https://github.com/getsentry/relay/pull/6343)) **Internal**: diff --git a/relay-server/src/endpoints/minidump.rs b/relay-server/src/endpoints/minidump.rs index 0d19c3927ee..d3b1a6f0d6b 100644 --- a/relay-server/src/endpoints/minidump.rs +++ b/relay-server/src/endpoints/minidump.rs @@ -2,10 +2,7 @@ use axum::extract::{DefaultBodyLimit, Request}; use axum::response::IntoResponse; use axum::routing::{MethodRouter, post}; use bytes::Bytes; -use bzip2::read::BzDecoder; -use flate2::read::GzDecoder; use futures::{self, Stream, StreamExt, TryStreamExt}; -use liblzma::read::XzDecoder; use multer::{Field, Multipart}; use relay_config::ConfigSnapshot; use relay_dynamic_config::Feature; @@ -15,12 +12,9 @@ use relay_system::Addr; use smallvec::smallvec; use std::convert::Infallible; use std::error::Error; -use std::io::Cursor; -use std::io::Read; use tokio::io::BufReader; use tokio_util::io::{ReaderStream, StreamReader}; use tower_http::limit::RequestBodyLimitLayer; -use zstd::stream::Decoder as ZstdDecoder; use crate::constants::{ITEM_NAME_BREADCRUMBS1, ITEM_NAME_BREADCRUMBS2, ITEM_NAME_EVENT}; use crate::endpoints::common::{self, BadStoreRequest, TextResponse, upload_stream}; @@ -32,7 +26,6 @@ use crate::service::ServiceState; use crate::services::outcome::{DiscardAttachmentType, DiscardItemType, DiscardReason, Outcome}; use crate::services::projects::project::ProjectState; use crate::services::upload::{ByteStream, ProjectContext, Upload}; -use crate::statsd::RelayCounters; use crate::utils::{ self, AttachmentStrategy, SizeSplit, find_error_source, is_length_limit_error, peek_n, read_bytes_into_item, read_field_into_item, @@ -127,16 +120,6 @@ fn validate_minidump(data: &[u8]) -> Result<(), BadStoreRequest> { Ok(()) } -/// Convenience wrapper to let a decoder decode its full input into a buffer. -/// -/// Stops reading once `max_size` is exceeded and returns an error. This prevents -/// decompression bombs from exhausting memory. -fn run_decoder(mut decoder: impl Read) -> std::io::Result> { - let mut buffer = Vec::new(); - decoder.read_to_end(&mut buffer)?; - Ok(buffer) -} - /// Types of compression we support for minidump payloads. enum Compression { None, @@ -162,46 +145,24 @@ impl Compression { } } -/// Creates a decoder based on the magic bytes in the minidump payload. -fn decoder_from(minidump_data: Bytes) -> Option> { - match Compression::from(&minidump_data) { - Compression::None => None, - Compression::Gzip => Some(Box::new(GzDecoder::new(Cursor::new(minidump_data)))), - Compression::Xz => Some(Box::new(XzDecoder::new(Cursor::new(minidump_data)))), - Compression::Bzip2 => Some(Box::new(BzDecoder::new(Cursor::new(minidump_data)))), - Compression::Zstd => match ZstdDecoder::new(Cursor::new(minidump_data)) { - Ok(decoder) => Some(Box::new(decoder)), - Err(ref err) => { - relay_log::error!(error = err as &dyn Error, "failed to create ZstdDecoder"); - None - } - }, - } -} - /// Tries to decode a minidump using any of the supported compression formats /// or returns the provided minidump payload untouched if no format where detected. /// /// Returns an `Overflow` error if the decompressed size exceeds `max_size`. -fn decode_minidump(minidump_data: Bytes, max_size: usize) -> Result { - let Some(decoder) = decoder_from(minidump_data.clone()) else { - // this means we haven't detected any compression container - // proceed to process the payload untouched (as a plain minidump). +async fn decode_minidump(minidump_data: Bytes, max_size: usize) -> Result { + if matches!(Compression::from(&minidump_data), Compression::None) { return Ok(minidump_data); - }; - - // Determine if this is a niche use-case of if this happens frequently. - relay_statsd::metric!(counter(RelayCounters::CompressedMinidump) += 1); - - let decoder = decoder.take(max_size.saturating_add(1) as u64); + } + let stream = futures::stream::once(async move { Ok::<_, Infallible>(minidump_data) }); + let decoded = decode_stream(stream) + .await + .map_err(BadStoreRequest::InvalidCompression)?; - match run_decoder(decoder) { - Ok(decoded) => { - if decoded.len() > max_size { - let item_type = DiscardItemType::Attachment(DiscardAttachmentType::Minidump); - return Err(BadStoreRequest::ItemTooLarge(item_type)); - } - Ok(Bytes::from(decoded)) + match utils::stream::split_by_size(decoded, max_size.saturating_add(1)).await { + Ok(SizeSplit::Small(decoded)) => Ok(decoded), + Ok(SizeSplit::Large(_)) => { + let item_type = DiscardItemType::Attachment(DiscardAttachmentType::Minidump); + Err(BadStoreRequest::ItemTooLarge(item_type)) } Err(err) => { // we detected a compression container but failed to decode it @@ -471,7 +432,9 @@ async fn multipart_to_items( .await .reject(&items)? .unwrap_or(payload); - let payload = decode_minidump(payload, config.max_attachment_size()).reject(&items)?; + let payload = decode_minidump(payload, config.max_attachment_size()) + .await + .reject(&items)?; items.try_modify(|items, records| -> Result<(), BadStoreRequest> { let minidump_item = items @@ -586,8 +549,10 @@ async fn raw_minidump_to_item( .map_err(|e| BadStoreRequest::InvalidBody(std::io::Error::other(e)))? { SizeSplit::Small(bytes) => { + let payload = decode_minidump(bytes, state.config().max_attachment_size()) + .await + .reject(&item)?; item.try_modify(|inner, records| -> Result<(), BadStoreRequest> { - let payload = decode_minidump(bytes, state.config().max_attachment_size())?; inner.set_payload(ContentType::Minidump, payload); records.lenient(DataCategory::Attachment); // decoding changes its size validate_minidump(&inner.payload())?; @@ -621,8 +586,10 @@ async fn raw_minidump_to_item( None => BadStoreRequest::InvalidBody(std::io::Error::other(e)), })?; + let payload = decode_minidump(minidump_data, state.config().max_attachment_size()) + .await + .reject(&item)?; item.try_modify(|inner, records| -> Result<(), BadStoreRequest> { - let payload = decode_minidump(minidump_data, state.config().max_attachment_size())?; inner.set_payload(ContentType::Minidump, payload); records.lenient(DataCategory::Attachment); // decoding the minidump changes its size validate_minidump(&inner.payload())?; @@ -788,31 +755,6 @@ mod tests { Ok(Bytes::from(compressed)) } - #[test] - fn test_validate_encoded_minidump() -> Result<(), Box> { - let encoders: Vec = vec![encode_gzip, encode_zst, encode_bzip, encode_xz]; - for encoder in &encoders { - let be_minidump = b"PMDMxxxxxx"; - let compressed = encoder(be_minidump)?; - let decoder = decoder_from(compressed).unwrap(); - assert!(run_decoder(decoder).is_ok()); - - let le_minidump = b"MDMPxxxxxx"; - let compressed = encoder(le_minidump)?; - let decoder = decoder_from(compressed).unwrap(); - assert!(run_decoder(decoder).is_ok()); - - let garbage = b"xxxxxx"; - let compressed = encoder(garbage)?; - let decoder = decoder_from(compressed).unwrap(); - let decoded = run_decoder(decoder); - assert!(decoded.is_ok()); - assert!(validate_minidump(&decoded.unwrap()).is_err()); - } - - Ok(()) - } - fn stream_of(data: Bytes) -> impl Stream> + Send + 'static { futures::stream::once(async move { Ok(data) }) } @@ -870,19 +812,19 @@ mod tests { Ok(()) } - #[test] - fn test_decode_minidump_size_limit() -> Result<(), Box> { + #[tokio::test] + async fn test_decode_minidump_size_limit() -> Result<(), Box> { // Create a minidump that will decompress to 100 bytes let minidump_data = b"xxxxxxxxxx".repeat(10); let compressed = encode_gzip(&minidump_data)?; // With a limit larger than the decompressed size, decoding should succeed - let result = decode_minidump(compressed.clone(), 200); + let result = decode_minidump(compressed.clone(), 200).await; assert!(result.is_ok()); assert_eq!(result.unwrap().len(), 100); // With a limit smaller than the decompressed size, decoding should fail with Overflow - let result = decode_minidump(compressed, 50); + let result = decode_minidump(compressed, 50).await; assert!(matches!(result, Err(BadStoreRequest::ItemTooLarge(_)))); Ok(()) diff --git a/relay-server/src/statsd.rs b/relay-server/src/statsd.rs index 01c796a7e10..48ae540be32 100644 --- a/relay-server/src/statsd.rs +++ b/relay-server/src/statsd.rs @@ -1015,8 +1015,6 @@ pub enum RelayCounters { /// This metric is tagged with: /// - `expansion`: What expansion was used to expand the error (e.g. unreal). ErrorProcessed, - /// The number of times that relay receives a compressed minidump. - CompressedMinidump, /// The number of times a trace metric has a nil trace ID. /// /// This metric is tagged with: @@ -1091,7 +1089,6 @@ impl CounterMetric for RelayCounters { RelayCounters::EnvelopeWithLogs => "logs.envelope", RelayCounters::ProfileChunksWithoutPlatform => "profile_chunk.no_platform", RelayCounters::ErrorProcessed => "event.error.processed", - RelayCounters::CompressedMinidump => "minidump.compressed.count", RelayCounters::TraceMetricNilTraceId => "trace_metric.nil_trace_id", RelayCounters::StandaloneAttachment => "processing.standalone_attachment", }