feat(attachment): Add verification logic for attachment placeholders

CHANGELOG.md

@@ -23,6 +23,7 @@
 - Calculate and track accepted bytes per individual trace metric item via `TraceMetricByte` data category. ([#5744](https://github.com/getsentry/relay/pull/5744), [#5767](https://github.com/getsentry/relay/pull/5767))
 - Use new processor architecture to process standalone profiles. ([#5741](https://github.com/getsentry/relay/pull/5741))
 - TUS: Disallow creation with upload. ([#5734](https://github.com/getsentry/relay/pull/5734))
+- Add logic to verify attachment placeholders and correctly store them. ([#5747](https://github.com/getsentry/relay/pull/5747))
 - Remove continuous-profiling-beta feature flags. ([#5762](https://github.com/getsentry/relay/pull/5762))
 - Playstation: Do not upload attachments if quota is 0. ([#5770](https://github.com/getsentry/relay/pull/5770))
 - Add payload byte size to trace metrics. ([#5764](https://github.com/getsentry/relay/pull/5764))

relay-server/src/envelope/attachment.rs

@@ -1,6 +1,6 @@
 use std::fmt;
 
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
 use crate::envelope::ContentType;
 
@@ -78,8 +78,9 @@ impl fmt::Display for AttachmentType {
 /// Represents the payload of an [attachment placeholder item](
 /// https://develop.sentry.dev/sdk/telemetry/attachments/#attachment-placeholder-item).
 #[cfg_attr(not(sentry), expect(unused))]
-#[derive(Serialize)]
+#[derive(Serialize, Deserialize)]
 pub struct AttachmentPlaceholder<'a> {
+    #[serde(borrow)]
     pub location: &'a str,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub content_type: Option<ContentType>,

relay-server/src/envelope/item.rs

@@ -558,7 +558,7 @@ impl Item {
     }
 
     /// Returns `true` if this item is an attachment placeholder.
-    fn is_attachment_ref(&self) -> bool {
+    pub fn is_attachment_ref(&self) -> bool {
         self.ty() == &ItemType::Attachment
             && self.content_type() == Some(ContentType::AttachmentRef)
     }

relay-server/src/processing/attachments/mod.rs

@@ -88,7 +88,7 @@ impl processing::Processor for AttachmentProcessor {
 
     async fn process(
         &self,
-        attachments: Managed<Self::Input>,
+        mut attachments: Managed<Self::Input>,
         ctx: processing::Context<'_>,
     ) -> Result<processing::Output<Self::Output>, Rejected<Self::Error>> {
         let client_name = crate::utils::client_name_tag(attachments.headers.meta().client_name());
@@ -125,6 +125,7 @@ impl processing::Processor for AttachmentProcessor {
             }
         }
 
+        process::validate_attachments(&mut attachments, ctx);
         let mut attachments = self.limiter.enforce_quotas(attachments, ctx).await?;
         process::scrub(&mut attachments, ctx)?;
 

relay-server/src/processing/attachments/process.rs

@@ -16,3 +16,27 @@ pub fn scrub(
         Ok::<_, Error>(())
     })
 }
+
+/// Validates the attachments and drop any invalid ones.
+///
+/// An attachment might be a placeholder, in which case its signature needs to be verified.
+pub fn validate_attachments(
+    attachments: &mut Managed<SerializedAttachments>,
+    ctx: processing::Context<'_>,
+) {
+    if !ctx.is_processing() {
+        return;
+    }
+
+    attachments.modify(|attachments, records| {
+        attachments.attachments.retain_mut(|attachment| {
+            match utils::attachments::validate(attachment, ctx.config) {
+                Ok(()) => true,
+                Err(err) => {
+                    records.reject_err(err, &*attachment);
+                    false
+                }
+            }
+        });
+    });
+}

relay-server/src/processing/errors/mod.rs

@@ -109,6 +109,8 @@ impl processing::Processor for ErrorsProcessor {
     ) -> Result<Output<Self::Output>, Rejected<Self::Error>> {
         let mut error = process::expand(error, ctx)?;
 
+        process::validate_attachments(&mut error, ctx);
+
         process::process(&mut error)?;
 
         process::finalize(&mut error, ctx)?;

relay-server/src/processing/errors/process.rs

@@ -141,3 +141,24 @@ pub fn scrub(error: &mut Managed<ExpandedError>, ctx: Context<'_>) -> Result<(),
         Ok::<_, Error>(())
     })
 }
+
+/// Validates the attachments and drop any invalid ones.
+///
+/// An attachment might be a placeholder, in which case it needs to be validated.
+pub fn validate_attachments(error: &mut Managed<ExpandedError>, ctx: Context<'_>) {
+    if !ctx.is_processing() {
+        return;
+    }
+
+    error.modify(|error, records| {
+        error.attachments.retain_mut(|attachment| {
+            match processing::utils::attachments::validate(attachment, ctx.config) {
+                Ok(()) => true,
+                Err(err) => {
+                    records.reject_err(err, &*attachment);
+                    false
+                }
+            }
+        });
+    });
+}

relay-server/src/processing/transactions/mod.rs

@@ -134,6 +134,9 @@ impl Processor for TransactionProcessor {
         relay_log::trace!("Expand transaction");
         let mut tx = process::expand(tx)?;
 
+        relay_log::trace!("Validate attachments");
+        process::validate_attachments(&mut tx, ctx);
+
         relay_log::trace!("Prepare transaction data");
         process::prepare_data(&mut tx, &mut ctx, &mut metrics)?;
 

relay-server/src/processing/transactions/process.rs

@@ -445,3 +445,24 @@ impl Counted for IndexedSpans {
         smallvec![(DataCategory::SpanIndexed, self.0)]
     }
 }
+
+/// Validates the attachments and drop any invalid ones.
+///
+/// An attachment might be a placeholder, in which case it needs to be validated.
+pub fn validate_attachments(transaction: &mut Managed<Box<ExpandedTransaction>>, ctx: Context<'_>) {
+    if !ctx.is_processing() {
+        return;
+    }
+
+    transaction.modify(|transaction, records| {
+        transaction.attachments.retain_mut(|attachment| {
+            match utils::attachments::validate(attachment, ctx.config) {
+                Ok(()) => true,
+                Err(err) => {
+                    records.reject_err(err, &*attachment);
+                    false
+                }
+            }
+        });
+    });
+}

relay-server/src/processing/utils/attachments.rs

@@ -1,16 +1,52 @@
 use std::error::Error;
 use std::time::Instant;
 
+use relay_config::Config;
 use relay_pii::{PiiAttachmentsProcessor, SelectorPathItem, SelectorSpec};
 use relay_statsd::metric;
 
+#[cfg(feature = "processing")]
+use crate::envelope::AttachmentPlaceholder;
 use crate::envelope::{AttachmentType, ContentType, Item, ItemType};
 use crate::managed::RecordKeeper;
+use crate::services::processor::ProcessingError;
 use crate::statsd::RelayTimers;
 
 use crate::services::projects::project::ProjectInfo;
 use relay_dynamic_config::Feature;
 
+#[cfg_attr(not(feature = "processing"), expect(unused_variables))]
+pub fn validate(item: &Item, config: &Config) -> Result<(), ProcessingError> {
+    #[cfg(not(feature = "processing"))]
+    return Ok(());
+
+    #[cfg(feature = "processing")]
+    {
+        if !item.is_attachment_ref() {
+            return Ok(());
+        }
+
+        let payload = item.payload();
+        let payload: AttachmentPlaceholder =
+            serde_json::from_slice(&payload).map_err(|_| ProcessingError::InvalidAttachmentRef)?;
+        let signed_location =
+            crate::services::upload::SignedLocation::try_from_str(payload.location)
+                .ok_or(ProcessingError::InvalidAttachmentRef)?;
+        // NOTE: Using the received timestamp here breaks tests without a pop-relay.
+        let location = signed_location
+            .verify(chrono::Utc::now(), config)
+            .map_err(|_| ProcessingError::InvalidAttachmentRef)?;
+        let signed_length = location
+            .length
+            .ok_or(ProcessingError::InvalidAttachmentRef)?;
+
+        match item.attachment_body_size() == signed_length {
+            true => Ok(()),
+            false => Err(ProcessingError::InvalidAttachmentRef),
+        }
+    }
+}
+
 /// Apply data privacy rules to attachments in the envelope.
 ///
 /// This only applies the new PII rules that explicitly select `ValueType::Binary` or one of the

relay-server/src/services/objectstore.rs

@@ -18,7 +18,7 @@ use relay_system::{
 use sentry_protos::snuba::v1::TraceItem;
 
 use crate::constants::DEFAULT_ATTACHMENT_RETENTION;
-use crate::envelope::ItemType;
+use crate::envelope::{Item, ItemType};
 use crate::managed::{
     Counted, Managed, ManagedEnvelope, ManagedResult, OutcomeError, Quantities, Rejected,
 };
@@ -319,8 +319,7 @@ impl ObjectstoreServiceInner {
             }
             Ok(session) => {
                 for attachment in attachments {
-                    // we are not storing zero-size attachments in objectstore
-                    if attachment.is_empty() {
+                    if Self::should_skip_upload(attachment) {
                         continue;
                     }
                     let result = self
@@ -351,8 +350,7 @@ impl ObjectstoreServiceInner {
     /// This mutates the attachment item in-place, setting the `stored_key` field to the key in the
     /// objectstore.
     async fn handle_event_attachment(&self, mut attachment: Managed<StoreAttachment>) {
-        // we are not storing zero-size attachments in objectstore
-        if attachment.attachment.is_empty() {
+        if Self::should_skip_upload(&attachment.attachment) {
             self.store.send(attachment);
             return;
         }
@@ -545,4 +543,13 @@ impl ObjectstoreServiceInner {
 
         Ok(ObjectstoreKey(response.key))
     }
+
+    /// Returns `true` if the item should **not** be uploaded to the objectstore.
+    ///
+    /// This is the case for:
+    /// - Zero-size attachments
+    /// - Attachment placeholders
+    fn should_skip_upload(item: &Item) -> bool {
+        item.is_empty() || item.is_attachment_ref()
+    }
 }

relay-server/src/services/outcome.rs

@@ -485,6 +485,10 @@ pub enum DiscardReason {
     /// (Relay) A trace attachment that has invalid item headers or attachment meta-data.
     InvalidTraceAttachment,
 
+    /// (Relay) An attachment ref that has invalid item headers or payload.
+    #[cfg(feature = "processing")]
+    InvalidAttachmentRef,
+
     /// (Relay) A required feature is not enabled.
     FeatureDisabled(Feature),
 
@@ -555,6 +559,8 @@ impl DiscardReason {
             DiscardReason::InvalidSpan => "invalid_span",
             DiscardReason::InvalidSpanAttachment => "invalid_span_attachment",
             DiscardReason::InvalidTraceAttachment => "invalid_trace_attachment",
+            #[cfg(feature = "processing")]
+            DiscardReason::InvalidAttachmentRef => "invalid_placeholder_attachment",
             DiscardReason::FeatureDisabled(_) => "feature_disabled",
             DiscardReason::TransactionAttachment => "transaction_attachment",
             DiscardReason::InvalidCheckIn => "invalid_check_in",

relay-server/src/services/processor.rs

@@ -620,6 +620,10 @@ pub enum ProcessingError {
     ProcessingGroupMismatch,
     #[error("new processing pipeline failed")]
     ProcessingFailure,
+
+    #[cfg(feature = "processing")]
+    #[error("invalid attachment reference")]
+    InvalidAttachmentRef,
 }
 
 impl ProcessingError {
@@ -662,6 +666,10 @@ impl ProcessingError {
             Self::ProcessingGroupMismatch => Some(Outcome::Invalid(DiscardReason::Internal)),
             // Outcomes are emitted in the new processing pipeline already.
             Self::ProcessingFailure => None,
+            #[cfg(feature = "processing")]
+            Self::InvalidAttachmentRef => {
+                Some(Outcome::Invalid(DiscardReason::InvalidAttachmentRef))
+            }
         }
     }