relay_server/utils/
multipart.rs

1use std::future::Future;
2use std::io;
3
4use axum::extract::Request;
5use bytes::Bytes;
6use futures::TryStreamExt;
7use multer::{Field, Multipart};
8use relay_config::ConfigSnapshot;
9use relay_quotas::DataCategory;
10use relay_system::Addr;
11use serde::{Deserialize, Serialize};
12use tokio::io::AsyncReadExt;
13use tokio_util::io::StreamReader;
14
15use crate::endpoints::common::BadStoreRequest;
16use crate::envelope::{AttachmentType, ContentType, Item, ItemType, Items};
17use crate::extractors::RequestMeta;
18use crate::managed::Managed;
19use crate::services::outcome::{
20    DiscardAttachmentType, DiscardItemType, DiscardReason, Outcome, TrackOutcome,
21};
22
23/// Type used for encoding string lengths.
24type Len = u32;
25
26/// Serializes a Pascal-style string with a 4 byte little-endian length prefix.
27fn write_string<W>(mut writer: W, string: &str) -> io::Result<()>
28where
29    W: io::Write,
30{
31    writer.write_all(&(string.len() as Len).to_le_bytes())?;
32    writer.write_all(string.as_bytes())?;
33
34    Ok(())
35}
36
37/// Safely consumes a slice of the given length.
38fn split_front<'a>(data: &mut &'a [u8], len: usize) -> Option<&'a [u8]> {
39    if data.len() < len {
40        *data = &[];
41        return None;
42    }
43
44    let (slice, rest) = data.split_at(len);
45    *data = rest;
46    Some(slice)
47}
48
49/// Consumes the 4-byte length prefix of a string.
50fn consume_len(data: &mut &[u8]) -> Option<usize> {
51    let len = std::mem::size_of::<Len>();
52    let slice = split_front(data, len)?;
53    let bytes = slice.try_into().ok();
54    bytes.map(|b| Len::from_le_bytes(b) as usize)
55}
56
57/// Consumes a Pascal-style string with a 4 byte little-endian length prefix.
58fn consume_string<'a>(data: &mut &'a [u8]) -> Option<&'a str> {
59    let len = consume_len(data)?;
60    let bytes = split_front(data, len)?;
61    std::str::from_utf8(bytes).ok()
62}
63
64/// An entry in a serialized form data item.
65#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
66pub struct FormDataEntry<'a>(&'a str, &'a str);
67
68impl<'a> FormDataEntry<'a> {
69    pub fn new(key: &'a str, value: &'a str) -> Self {
70        Self(key, value)
71    }
72
73    pub fn key(&self) -> &'a str {
74        self.0
75    }
76
77    pub fn value(&self) -> &'a str {
78        self.1
79    }
80
81    fn to_writer<W: io::Write>(&self, mut writer: W) {
82        write_string(&mut writer, self.key()).ok();
83        write_string(&mut writer, self.value()).ok();
84    }
85
86    fn read(data: &mut &'a [u8]) -> Option<Self> {
87        let key = consume_string(data)?;
88        let value = consume_string(data)?;
89        Some(Self::new(key, value))
90    }
91}
92
93/// A writer for serialized form data.
94///
95/// This writer is used to serialize multiple plain fields from a multipart form data request into a
96/// single envelope item. Use `FormDataIter` to iterate all entries.
97pub struct FormDataWriter {
98    data: Vec<u8>,
99}
100
101impl FormDataWriter {
102    pub fn new() -> Self {
103        Self { data: Vec::new() }
104    }
105
106    pub fn append(&mut self, key: &str, value: &str) {
107        let entry = FormDataEntry::new(key, value);
108        entry.to_writer(&mut self.data);
109    }
110
111    pub fn into_inner(self) -> Vec<u8> {
112        self.data
113    }
114
115    pub fn len(&self) -> usize {
116        self.data.len()
117    }
118}
119
120/// Iterates through serialized form data written with `FormDataWriter`.
121pub struct FormDataIter<'a> {
122    data: &'a [u8],
123}
124
125impl<'a> FormDataIter<'a> {
126    pub fn new(data: &'a [u8]) -> Self {
127        Self { data }
128    }
129}
130
131impl<'a> Iterator for FormDataIter<'a> {
132    type Item = FormDataEntry<'a>;
133
134    fn next(&mut self) -> Option<Self::Item> {
135        while !self.data.is_empty() {
136            match FormDataEntry::read(&mut self.data) {
137                Some(entry) => return Some(entry),
138                None => relay_log::error!("form data deserialization failed"),
139            }
140        }
141
142        None
143    }
144}
145
146/// Looks for a multipart boundary at the beginning of the data
147/// and returns it as a `&str` if it is found
148///
149/// A multipart boundary starts at the beginning of the data (possibly
150/// after some blank lines) and it is prefixed by '--' (two dashes)
151///
152/// ```ignore
153/// let boundary = get_multipart_boundary(b"--The boundary\r\n next line");
154/// assert_eq!(Some("The boundary"), boundary);
155///
156/// let invalid_boundary = get_multipart_boundary(b"The boundary\r\n next line");
157/// assert_eq!(None, invalid_boundary);
158/// ```
159pub fn get_multipart_boundary(data: &[u8]) -> Option<&str> {
160    data.split(|&byte| byte == b'\r' || byte == b'\n')
161        // Get the first non-empty line
162        .find(|slice| !slice.is_empty())
163        // Check for the form boundary indicator
164        .filter(|slice| slice.len() > 2 && slice.starts_with(b"--"))
165        // Form boundaries must be valid UTF-8 strings
166        .and_then(|slice| std::str::from_utf8(&slice[2..]).ok())
167}
168
169/// Strategy for how to infer attachment type and add a multipart attachment to an envelope item.
170///
171/// This enables different endpoints to have different ways of dealing with multipart attachments,
172/// for instance, one endpoint can upload attachments and add a ref to the item, while another
173/// endpoint can add attachments to items directly.
174pub trait AttachmentStrategy {
175    fn infer_type(&self, field: &Field) -> AttachmentType;
176
177    /// Defines how individual multipart items should be handled.
178    ///
179    /// Returns
180    ///  - `Ok(Some(item))` if everything was successful.
181    ///  - `Ok(None)` if there was an error adding the attachment, but the rest of the request
182    ///    should still be handled.
183    ///  - `Err(..)` if there was an unexpected error adding the attachment and the request should
184    ///    be cancelled.
185    fn add_to_item(
186        &self,
187        field: Field<'static>,
188        item: Managed<Item>,
189        config: &ConfigSnapshot,
190    ) -> impl Future<Output = Result<Option<Managed<Item>>, BadStoreRequest>> + Send;
191}
192
193pub fn read_bytes_into_item(
194    bytes: Bytes,
195    mut item: Managed<Item>,
196    content_type: Option<ContentType>,
197) -> Managed<Item> {
198    item.modify(|inner, records| {
199        if let Some(content_type) = content_type {
200            inner.set_payload(content_type, bytes);
201        } else {
202            inner.set_payload_without_content_type(bytes);
203        };
204        records.lenient(DataCategory::Attachment);
205    });
206
207    item
208}
209
210async fn read_field_data(field: Field<'static>, limit: usize) -> Result<Vec<u8>, multer::Error> {
211    let mut buf = Vec::new();
212    StreamReader::new(field.map_err(io::Error::other))
213        .take((limit) as u64)
214        .read_to_end(&mut buf)
215        .await
216        .map_err(|e| multer::Error::StreamReadFailed(Box::new(e)))?;
217    Ok(buf)
218}
219
220pub async fn read_field_into_item(
221    field: Field<'static>,
222    mut item: Managed<Item>,
223    config: &ConfigSnapshot,
224) -> Result<Managed<Item>, multer::Error> {
225    let content_type = field
226        .content_type()
227        .map(|ct| ct.as_ref().parse().unwrap_or(ContentType::OctetStream));
228    let field_name = field.name().map(String::from);
229    let limit = config.max_attachment_size();
230    let bytes = Bytes::from(read_field_data(field, limit + 1).await?); // Extra byte needed to determine if limit was exceeded.
231    let n_bytes = bytes.len();
232    item.modify(|inner, records| {
233        if let Some(content_type) = content_type {
234            inner.set_payload(content_type, bytes);
235        } else {
236            inner.set_payload_without_content_type(bytes);
237        };
238        records.lenient(DataCategory::Attachment);
239    });
240
241    if n_bytes > limit {
242        let attachment_type = item.attachment_type().unwrap_or(AttachmentType::Attachment);
243        let item_type = DiscardItemType::Attachment(DiscardAttachmentType::from(attachment_type));
244        let _ = item.reject_err(Outcome::Invalid(DiscardReason::ItemTooLarge(item_type)));
245
246        Err(multer::Error::FieldSizeExceeded {
247            limit: limit as u64,
248            field_name,
249        })
250    } else {
251        Ok(item)
252    }
253}
254
255pub async fn multipart_items(
256    mut multipart: Multipart<'static>,
257    config: &ConfigSnapshot,
258    attachment_strategy: impl AttachmentStrategy,
259    request_meta: &RequestMeta,
260    outcome_aggregator: &Addr<TrackOutcome>,
261) -> Result<Managed<Items>, BadStoreRequest> {
262    let mut items =
263        Managed::with_meta_from_request_meta(request_meta, outcome_aggregator, Items::new());
264    let mut form_data = FormDataWriter::new();
265    let mut attachments_size = 0;
266
267    while let Some(field) = multipart.next_field().await? {
268        if let Some(file_name) = field.file_name() {
269            let mut item = Item::new(ItemType::Attachment);
270            let attachment_type = attachment_strategy.infer_type(&field);
271            item.set_attachment_type(attachment_type);
272            item.set_filename(file_name);
273            let item = items.wrap(item);
274            let item = attachment_strategy
275                .add_to_item(field, item, config)
276                .await
277                .inspect_err(|e| {
278                    if matches!(
279                        e,
280                        BadStoreRequest::InvalidMultipart(multer::Error::FieldSizeExceeded { .. })
281                    ) {
282                        let attachment_type = DiscardAttachmentType::from(attachment_type);
283                        let item_type = DiscardItemType::Attachment(attachment_type);
284                        let discard_reason = DiscardReason::ItemTooLarge(item_type);
285                        let _ = items.reject_err(Outcome::Invalid(discard_reason));
286                    }
287                })?;
288            if let Some(item) = item {
289                // This increases the attachments byte count even if the item is an attachment ref.
290                // This is by design as the total number of bytes read into memory should be
291                // constrained.
292                attachments_size += item.len();
293                items.merge_with(item, |items, item, _| items.push(item));
294                if attachments_size > config.max_attachments_size() {
295                    let item_type = DiscardItemType::Attachment(DiscardAttachmentType::Attachment);
296                    let _ =
297                        items.reject_err(Outcome::Invalid(DiscardReason::ItemTooLarge(item_type)));
298                    return Err(BadStoreRequest::RequestTooLarge);
299                }
300            }
301        } else if let Some(field_name) = field.name().map(str::to_owned) {
302            // Since the FormData ends up in an event use the event size limit.
303            let limit = config.max_event_size();
304            // Extra byte needed to determine if limit was exceeded.
305            let data = read_field_data(field, limit + 1).await?;
306            // Ensure to decode this SAFELY to match Django's POST data behavior. This allows us to
307            // process sentry event payloads even if they contain invalid encoding.
308            let string = String::from_utf8_lossy(&data);
309            form_data.append(&field_name, &string);
310
311            if form_data.len() > limit {
312                return Err(items
313                    .reject_err(BadStoreRequest::ItemTooLarge(DiscardItemType::FormData))
314                    .into());
315            }
316        } else {
317            relay_log::trace!("multipart content without name or file_name");
318        }
319    }
320
321    let form_data = form_data.into_inner();
322    if !form_data.is_empty() {
323        let mut item = Item::new(ItemType::FormData);
324        // Content type is `Text` (since it is not a json object but multiple
325        // json arrays serialized one after the other).
326        item.set_payload(ContentType::Text, form_data);
327        items.merge_with(items.wrap(item), |items, item, _| items.push(item));
328    }
329
330    Ok(items)
331}
332
333pub fn multipart_from_request(request: Request) -> Result<Multipart<'static>, BadStoreRequest> {
334    let content_type = request
335        .headers()
336        .get("content-type")
337        .and_then(|v| v.to_str().ok())
338        .unwrap_or("");
339    let boundary =
340        multer::parse_boundary(content_type).map_err(BadStoreRequest::InvalidMultipart)?;
341    Ok(Multipart::new(
342        request.into_body().into_data_stream(),
343        boundary,
344    ))
345}
346
347#[cfg(test)]
348mod tests {
349    use std::convert::Infallible;
350
351    use relay_config::Config;
352
353    use super::*;
354
355    fn mock_request_meta() -> RequestMeta {
356        let dsn = "https://a94ae32be2582e0bbd7a4cbb95971fee:@sentry.io/42"
357            .parse()
358            .unwrap();
359        RequestMeta::new(dsn)
360    }
361
362    #[test]
363    fn test_get_boundary() {
364        let examples: &[(&[u8], Option<&str>)] = &[
365            (b"--some_val", Some("some_val")),
366            (b"--\nsecond line", None),
367            (b"\n\r--some_val", Some("some_val")),
368            (b"\n\r--some_val\nadfa", Some("some_val")),
369            (b"\n\r--some_val\rfasdf", Some("some_val")),
370            (b"\n\r--some_val\r\nfasdf", Some("some_val")),
371            (b"\n\rsome_val", None),
372            (b"", None),
373            (b"--", None),
374        ];
375
376        for (input, expected) in examples {
377            let boundary = get_multipart_boundary(input);
378            assert_eq!(*expected, boundary);
379        }
380    }
381
382    #[test]
383    fn test_formdata() {
384        let mut writer = FormDataWriter::new();
385        writer.append("foo", "foo");
386        writer.append("bar", "");
387        writer.append("blub", "blub");
388
389        let payload = writer.into_inner();
390        let iter = FormDataIter::new(&payload);
391        let entries: Vec<_> = iter.collect();
392
393        assert_eq!(
394            entries,
395            vec![
396                FormDataEntry::new("foo", "foo"),
397                FormDataEntry::new("bar", ""),
398                FormDataEntry::new("blub", "blub"),
399            ]
400        );
401    }
402
403    #[test]
404    fn test_empty_formdata() {
405        let writer = FormDataWriter::new();
406        let payload = writer.into_inner();
407
408        let iter = FormDataIter::new(&payload);
409        let entries: Vec<_> = iter.collect();
410
411        assert_eq!(entries, vec![]);
412    }
413
414    /// Regression test for multipart payloads without a trailing newline.
415    #[tokio::test]
416    async fn missing_trailing_newline() {
417        let data = "--X-BOUNDARY\r\nContent-Disposition: form-data; \
418        name=\"my_text_field\"\r\n\r\nabcd\r\n--X-BOUNDARY--"; // No trailing newline
419
420        let stream = futures::stream::once(async { Ok::<_, Infallible>(data) });
421        let mut multipart = Multipart::new(stream, "X-BOUNDARY");
422
423        assert!(multipart.next_field().await.unwrap().is_some());
424        assert!(multipart.next_field().await.unwrap().is_none());
425    }
426
427    #[tokio::test]
428    async fn test_individual_size_limit_exceeded() {
429        let data = "--X-BOUNDARY\r\n\
430              Content-Disposition: form-data; name=\"file\"; filename=\"large.txt\"\r\n\
431              Content-Type: text/plain\r\n\
432              \r\n\
433              content too large for limit\r\n\
434              --X-BOUNDARY\r\n\
435              Content-Disposition: form-data; name=\"small_file\"; filename=\"small.txt\"\r\n\
436              Content-Type: text/plain\r\n\
437              \r\n\
438              ok\r\n\
439              --X-BOUNDARY--\r\n";
440
441        let stream = futures::stream::once(async move { Ok::<_, Infallible>(data) });
442        let multipart = Multipart::new(stream, "X-BOUNDARY");
443
444        let config = Config::from_json_value(serde_json::json!({
445            "limits": {
446                "max_attachment_size": 5
447            }
448        }))
449        .unwrap()
450        .current();
451
452        struct MockAttachmentStrategy;
453        impl AttachmentStrategy for MockAttachmentStrategy {
454            async fn add_to_item(
455                &self,
456                field: Field<'static>,
457                item: Managed<Item>,
458                config: &ConfigSnapshot,
459            ) -> Result<Option<Managed<Item>>, BadStoreRequest> {
460                Ok(Some(read_field_into_item(field, item, config).await?))
461            }
462
463            fn infer_type(&self, _: &Field) -> AttachmentType {
464                AttachmentType::Attachment
465            }
466        }
467
468        let res = multipart_items(
469            multipart,
470            &config,
471            MockAttachmentStrategy,
472            &mock_request_meta(),
473            &Addr::dummy(),
474        )
475        .await;
476        assert!(res.is_err_and(|x| matches!(
477            x,
478            BadStoreRequest::InvalidMultipart(multer::Error::FieldSizeExceeded { .. })
479        )));
480    }
481
482    #[tokio::test]
483    async fn test_collective_size_limit_exceeded() {
484        let data = "--X-BOUNDARY\r\n\
485              Content-Disposition: form-data; name=\"file\"; filename=\"large.txt\"\r\n\
486              Content-Type: text/plain\r\n\
487              \r\n\
488              content too large for limit\r\n\
489              --X-BOUNDARY\r\n\
490              Content-Disposition: form-data; name=\"small_file\"; filename=\"small.txt\"\r\n\
491              Content-Type: text/plain\r\n\
492              \r\n\
493              ok\r\n\
494              --X-BOUNDARY--\r\n";
495
496        let stream = futures::stream::once(async move { Ok::<_, Infallible>(data) });
497
498        let config = Config::from_json_value(serde_json::json!({
499            "limits": {
500                "max_attachments_size": 5
501            }
502        }))
503        .unwrap()
504        .current();
505
506        let multipart = Multipart::new(stream, "X-BOUNDARY");
507
508        struct MockAttachmentStrategy;
509        impl AttachmentStrategy for MockAttachmentStrategy {
510            async fn add_to_item(
511                &self,
512                field: Field<'static>,
513                item: Managed<Item>,
514                config: &ConfigSnapshot,
515            ) -> Result<Option<Managed<Item>>, BadStoreRequest> {
516                Ok(Some(read_field_into_item(field, item, config).await?))
517            }
518
519            fn infer_type(&self, _: &Field) -> AttachmentType {
520                AttachmentType::Attachment
521            }
522        }
523
524        let result = multipart_items(
525            multipart,
526            &config,
527            MockAttachmentStrategy,
528            &mock_request_meta(),
529            &Addr::dummy(),
530        )
531        .await;
532
533        // Should be warned if the overall stream limit is being breached.
534        assert!(result.is_err_and(|x| matches!(x, BadStoreRequest::RequestTooLarge)));
535    }
536}
relay_server/utils/multipart.rs

relay_server/utils/
multipart.rs
