fix(playstation): Add overall stream size limit

CHANGELOG.md

@@ -13,6 +13,7 @@
 - Emit outcomes in both `log_byte` and `log_item` categories when logs are dropped. ([#5766](https://github.com/getsentry/relay/pull/5766))
 - Propagate an event's retention policy to its attachments ([#5774](https://github.com/getsentry/relay/pull/5774))
 - Prevent panic in span description normalization. ([#5781](https://github.com/getsentry/relay/pull/5781))
+- Limit overall stream size in playstation multiparts. ([#5795](https://github.com/getsentry/relay/pull/5795))
 
 **Features**:
 

relay-server/src/endpoints/attachments.rs

@@ -1,7 +1,7 @@
-use axum::extract::Path;
+use axum::extract::{Path, Request};
 use axum::http::StatusCode;
 use axum::response::IntoResponse;
-use multer::Field;
+use multer::{Field, Multipart};
 use relay_config::Config;
 use relay_event_schema::protocol::EventId;
 use serde::Deserialize;
@@ -10,7 +10,7 @@ use crate::endpoints::common::{self, BadStoreRequest};
 use crate::envelope::{AttachmentType, Envelope, Item};
 use crate::extractors::RequestMeta;
 use crate::service::ServiceState;
-use crate::utils::{AttachmentStrategy, ConstrainedMultipart, read_attachment_bytes_into_item};
+use crate::utils::{self, AttachmentStrategy, read_attachment_bytes_into_item};
 
 #[derive(Debug, Deserialize)]
 pub struct AttachmentPath {
@@ -37,12 +37,10 @@ impl AttachmentStrategy for AttachmentsAttachmentStrategy {
 async fn extract_envelope(
     meta: RequestMeta,
     path: AttachmentPath,
-    multipart: ConstrainedMultipart,
+    multipart: Multipart<'static>,
     config: &Config,
 ) -> Result<Box<Envelope>, BadStoreRequest> {
-    let items = multipart
-        .items(config, AttachmentsAttachmentStrategy)
-        .await?;
+    let items = utils::multipart_items(multipart, config, AttachmentsAttachmentStrategy).await?;
 
     let mut envelope = Envelope::from_request(Some(path.event_id), meta);
     for item in items {
@@ -56,8 +54,9 @@ pub async fn handle(
     state: ServiceState,
     meta: RequestMeta,
     Path(path): Path<AttachmentPath>,
-    multipart: ConstrainedMultipart,
+    request: Request,
 ) -> axum::response::Result<impl IntoResponse> {
+    let multipart = utils::multipart_from_request(request, state.config().max_attachments_size())?;
     let envelope = extract_envelope(meta, path, multipart, state.config()).await?;
     common::handle_envelope(&state, envelope)
         .await?

relay-server/src/endpoints/minidump.rs

@@ -23,9 +23,7 @@ use crate::extractors::{RawContentType, RequestMeta};
 use crate::middlewares;
 use crate::service::ServiceState;
 use crate::services::outcome::{DiscardAttachmentType, DiscardItemType};
-use crate::utils::{
-    self, AttachmentStrategy, ConstrainedMultipart, read_attachment_bytes_into_item,
-};
+use crate::utils::{self, AttachmentStrategy, read_attachment_bytes_into_item};
 
 /// The field name of a minidump in the multipart form-data upload.
 ///
@@ -192,11 +190,11 @@ impl AttachmentStrategy for MinidumpAttachmentStrategy {
 }
 
 async fn extract_multipart(
-    multipart: ConstrainedMultipart,
+    multipart: Multipart<'static>,
     meta: RequestMeta,
     config: &Config,
 ) -> Result<Box<Envelope>, BadStoreRequest> {
-    let mut items = multipart.items(config, MinidumpAttachmentStrategy).await?;
+    let mut items = utils::multipart_items(multipart, config, MinidumpAttachmentStrategy).await?;
 
     let minidump_item = items
         .iter_mut()
@@ -262,7 +260,8 @@ async fn handle(
     let envelope = if MINIDUMP_RAW_CONTENT_TYPES.contains(&content_type.as_ref()) {
         extract_raw_minidump(request.extract().await?, meta, config.max_attachment_size())?
     } else {
-        let multipart = request.extract_with_state(&state).await?;
+        let multipart =
+            utils::multipart_from_request(request, state.config().max_attachments_size())?;
         extract_multipart(multipart, meta, config).await?
     };
 
@@ -453,11 +452,8 @@ mod tests {
 
         let config = Config::default();
 
-        let multipart = ConstrainedMultipart(
-            utils::multipart_from_request(request, multer::Constraints::new()).unwrap(),
-        );
-        let items = multipart
-            .items(&config, MinidumpAttachmentStrategy)
+        let multipart = utils::multipart_from_request(request, multipart_body.len()).unwrap();
+        let items = utils::multipart_items(multipart, &config, MinidumpAttachmentStrategy)
             .await
             .unwrap();
 

relay-server/src/endpoints/mod.rs

@@ -3,11 +3,12 @@
 //! This module contains implementations for all supported relay endpoints, as well as a generic
 //! `forward` endpoint that sends unknown requests to the upstream.
 
+pub(crate) mod common;
+
 mod attachments;
 mod autoscaling;
 mod batch_metrics;
 mod batch_outcomes;
-mod common;
 mod envelope;
 mod forward;
 mod health_check;

relay-server/src/endpoints/playstation.rs

@@ -30,7 +30,7 @@ use crate::services::outcome::DiscardReason;
 use crate::services::projects::cache::Project;
 use crate::services::upload::{Create, Stream, Upload};
 use crate::utils::{self, MeteredStream};
-use crate::utils::{AttachmentStrategy, BadMultipart, BoundedStream};
+use crate::utils::{AttachmentStrategy, BoundedStream};
 
 /// The extension of a prosperodump in the multipart form-data upload.
 const PROSPERODUMP_EXTENSION: &str = ".prosperodmp";
@@ -216,14 +216,15 @@ async fn handle(
         return Ok(axum::Json(create_data_request_response()).into_response());
     }
 
+    let config = state.config();
     let project = state
         .project_cache_handle()
-        .ready(meta.public_key(), state.config().query_timeout())
+        .ready(meta.public_key(), config.query_timeout())
         .await
         .ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
 
-    let multipart = utils::multipart_from_request(request, multer::Constraints::new())
-        .map_err(BadMultipart::Multipart)?;
+    let stream_size_limit = config.max_upload_size() + config.max_attachments_size();
+    let multipart = utils::multipart_from_request(request, stream_size_limit)?;
     let mut envelope = extract_multipart(multipart, meta, &state, &project).await?;
     envelope.require_feature(Feature::PlaystationIngestion);
 

relay-server/src/extractors/mod.rs

@@ -3,7 +3,6 @@ mod forwarded_for;
 mod integration_builder;
 mod mime;
 mod received_at;
-mod remote;
 mod request_meta;
 mod signature;
 mod signed_json;
@@ -13,6 +12,5 @@ pub use self::forwarded_for::*;
 pub use self::integration_builder::*;
 pub use self::mime::*;
 pub use self::received_at::*;
-pub use self::remote::*;
 pub use self::request_meta::*;
 pub use self::signed_json::*;

relay-server/src/utils/multipart.rs

@@ -1,21 +1,16 @@
-use std::convert::Infallible;
 use std::future::Future;
 use std::io;
 use std::task::Poll;
 
-use axum::extract::{FromRequest, Request};
-use axum::http::StatusCode;
-use axum::response::{IntoResponse, Response};
+use axum::extract::Request;
 use bytes::{Bytes, BytesMut};
 use futures::{StreamExt, TryStreamExt};
-use multer::{Field, Multipart};
+use multer::{Constraints, Field, Multipart, SizeLimit};
 use relay_config::Config;
 use serde::{Deserialize, Serialize};
 
+use crate::endpoints::common::BadStoreRequest;
 use crate::envelope::{AttachmentType, ContentType, Item, ItemType, Items};
-use crate::extractors::{BadEventMeta, Remote};
-use crate::service::ServiceState;
-use crate::utils::ApiErrorResponse;
 
 /// Type used for encoding string lengths.
 type Len = u32;
@@ -159,34 +154,6 @@ pub fn get_multipart_boundary(data: &[u8]) -> Option<&str> {
         .and_then(|slice| std::str::from_utf8(&slice[2..]).ok())
 }
 
-#[cfg_attr(not(sentry), allow(dead_code))]
-#[derive(Debug, thiserror::Error)]
-pub enum BadMultipart {
-    #[error("event metadata error: {0}")]
-    EventMeta(#[from] BadEventMeta),
-    #[error("multipart error: {0}")]
-    Multipart(#[from] multer::Error),
-}
-
-impl From<Infallible> for BadMultipart {
-    fn from(infallible: Infallible) -> Self {
-        match infallible {}
-    }
-}
-
-impl IntoResponse for BadMultipart {
-    fn into_response(self) -> Response {
-        let status_code = match self {
-            BadMultipart::Multipart(
-                multer::Error::FieldSizeExceeded { .. } | multer::Error::StreamSizeExceeded { .. },
-            ) => StatusCode::PAYLOAD_TOO_LARGE,
-            _ => StatusCode::BAD_REQUEST,
-        };
-
-        (status_code, ApiErrorResponse::from_error(&self)).into_response()
-    }
-}
-
 /// Strategy for how to infer attachment type and add a multipart attachment to an envelope item.
 ///
 /// This enables different endpoints to have different ways of dealing with multipart attachments,
@@ -357,53 +324,25 @@ impl futures::Stream for LimitedField<'_> {
     }
 }
 
-/// Wrapper around [`multer::Multipart`] that prevents the total stream size from exceeding
-/// [`Config::max_attachments_size`].
-///
-/// Although [`LimitedField`] prevents reading large fields into memory, it still reads
-/// the entire field before returning an error. This led to the incident described in PR
-/// [#4836](https://github.com/getsentry/relay/pull/4836) which necessitated introducing
-/// [`ConstrainedMultipart`]'s stream size limit.
-pub struct ConstrainedMultipart(pub Multipart<'static>);
-
-impl FromRequest<ServiceState> for ConstrainedMultipart {
-    type Rejection = Remote<multer::Error>;
-
-    async fn from_request(request: Request, state: &ServiceState) -> Result<Self, Self::Rejection> {
-        let limits =
-            multer::SizeLimit::new().whole_stream(state.config().max_attachments_size() as u64);
-
-        multipart_from_request(request, multer::Constraints::new().size_limit(limits))
-            .map(Self)
-            .map_err(Remote)
-    }
-}
-
-impl ConstrainedMultipart {
-    pub async fn items(
-        self,
-        config: &Config,
-        attachment_strategy: impl AttachmentStrategy,
-    ) -> Result<Items, multer::Error> {
-        multipart_items(self.0, config, attachment_strategy).await
-    }
-}
-
 pub fn multipart_from_request(
     request: Request,
-    constraints: multer::Constraints,
-) -> Result<Multipart<'static>, multer::Error> {
+    stream_size_limit: usize,
+) -> Result<Multipart<'static>, BadStoreRequest> {
     let content_type = request
         .headers()
         .get("content-type")
         .and_then(|v| v.to_str().ok())
         .unwrap_or("");
-    let boundary = multer::parse_boundary(content_type)?;
+    let boundary =
+        multer::parse_boundary(content_type).map_err(BadStoreRequest::InvalidMultipart)?;
 
+    // Limits the overall stream size, preventing overly long processing times which can cause
+    // incidents like the one described in [#4836](https://github.com/getsentry/relay/pull/4836).
+    let stream_size_limit = SizeLimit::new().whole_stream(stream_size_limit as u64);
     Ok(Multipart::with_constraints(
         request.into_body().into_data_stream(),
         boundary,
-        constraints,
+        Constraints::new().size_limit(stream_size_limit),
     ))
 }