feat(replays): Add inbound filters for Annotated<Replay> types. (#3420)

Adds inbound-filters for replay.

Related: getsentry/sentry#44700

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+**Features:**
+
+- Add inbound filters for Annotated<Replay> types. ([#3420](https://github.com/getsentry/relay/pull/3420))
+
 **Internal:**
 
 - Emit negative outcomes in metric stats for metrics. ([#3436](https://github.com/getsentry/relay/pull/3436))

relay-event-schema/src/protocol/replay.rs

@@ -27,11 +27,13 @@
 
 #[cfg(feature = "jsonschema")]
 use relay_jsonschema_derive::JsonSchema;
-use relay_protocol::{Annotated, Array, Empty, FromValue, IntoValue};
+use relay_protocol::{Annotated, Array, Empty, FromValue, Getter, IntoValue, Val};
 
 use crate::processor::ProcessValue;
 use crate::protocol::{
-    ClientSdkInfo, Contexts, EventId, LenientString, Request, Tags, Timestamp, User,
+    AppContext, BrowserContext, ClientSdkInfo, Contexts, DefaultContext, DeviceContext, EventId,
+    LenientString, OsContext, ProfileContext, Request, ResponseContext, Tags, Timestamp,
+    TraceContext, User,
 };
 use uuid::Uuid;
 
@@ -220,6 +222,170 @@ pub struct Replay {
     pub sdk: Annotated<ClientSdkInfo>,
 }
 
+impl Replay {
+    /// Returns a reference to the context if it exists in its default key.
+    pub fn context<C: DefaultContext>(&self) -> Option<&C> {
+        self.contexts.value()?.get()
+    }
+
+    /// Returns the raw user agent string.
+    ///
+    /// Returns `Some` if the event's request interface contains a `user-agent` header. Returns
+    /// `None` otherwise.
+    pub fn user_agent(&self) -> Option<&str> {
+        let headers = self.request.value()?.headers.value()?;
+
+        for item in headers.iter() {
+            if let Some((ref o_k, ref v)) = item.value() {
+                if let Some(k) = o_k.as_str() {
+                    if k.to_lowercase() == "user-agent" {
+                        return v.as_str();
+                    }
+                }
+            }
+        }
+
+        None
+    }
+}
+
+impl Getter for Replay {
+    fn get_value(&self, path: &str) -> Option<Val<'_>> {
+        Some(match path.strip_prefix("event.")? {
+            // Simple fields
+            "release" => self.release.as_str()?.into(),
+            "dist" => self.dist.as_str()?.into(),
+            "environment" => self.environment.as_str()?.into(),
+            "platform" => self.platform.as_str().unwrap_or("other").into(),
+
+            // Fields in top level structures (called "interfaces" in Sentry)
+            "user.email" => or_none(&self.user.value()?.email)?.into(),
+            "user.id" => or_none(&self.user.value()?.id)?.into(),
+            "user.ip_address" => self.user.value()?.ip_address.as_str()?.into(),
+            "user.name" => self.user.value()?.name.as_str()?.into(),
+            "user.segment" => or_none(&self.user.value()?.segment)?.into(),
+            "user.geo.city" => self.user.value()?.geo.value()?.city.as_str()?.into(),
+            "user.geo.country_code" => self
+                .user
+                .value()?
+                .geo
+                .value()?
+                .country_code
+                .as_str()?
+                .into(),
+            "user.geo.region" => self.user.value()?.geo.value()?.region.as_str()?.into(),
+            "user.geo.subdivision" => self.user.value()?.geo.value()?.subdivision.as_str()?.into(),
+            "request.method" => self.request.value()?.method.as_str()?.into(),
+            "request.url" => self.request.value()?.url.as_str()?.into(),
+            "sdk.name" => self.sdk.value()?.name.as_str()?.into(),
+            "sdk.version" => self.sdk.value()?.version.as_str()?.into(),
+
+            // Computed fields (after normalization).
+            "sentry_user" => self.user.value()?.sentry_user.as_str()?.into(),
+
+            // Partial implementation of contexts.
+            "contexts.app.in_foreground" => {
+                self.context::<AppContext>()?.in_foreground.value()?.into()
+            }
+            "contexts.device.arch" => self.context::<DeviceContext>()?.arch.as_str()?.into(),
+            "contexts.device.battery_level" => self
+                .context::<DeviceContext>()?
+                .battery_level
+                .value()?
+                .into(),
+            "contexts.device.brand" => self.context::<DeviceContext>()?.brand.as_str()?.into(),
+            "contexts.device.charging" => self.context::<DeviceContext>()?.charging.value()?.into(),
+            "contexts.device.family" => self.context::<DeviceContext>()?.family.as_str()?.into(),
+            "contexts.device.model" => self.context::<DeviceContext>()?.model.as_str()?.into(),
+            "contexts.device.locale" => self.context::<DeviceContext>()?.locale.as_str()?.into(),
+            "contexts.device.online" => self.context::<DeviceContext>()?.online.value()?.into(),
+            "contexts.device.orientation" => self
+                .context::<DeviceContext>()?
+                .orientation
+                .as_str()?
+                .into(),
+            "contexts.device.name" => self.context::<DeviceContext>()?.name.as_str()?.into(),
+            "contexts.device.screen_density" => self
+                .context::<DeviceContext>()?
+                .screen_density
+                .value()?
+                .into(),
+            "contexts.device.screen_dpi" => {
+                self.context::<DeviceContext>()?.screen_dpi.value()?.into()
+            }
+            "contexts.device.screen_width_pixels" => self
+                .context::<DeviceContext>()?
+                .screen_width_pixels
+                .value()?
+                .into(),
+            "contexts.device.screen_height_pixels" => self
+                .context::<DeviceContext>()?
+                .screen_height_pixels
+                .value()?
+                .into(),
+            "contexts.device.simulator" => {
+                self.context::<DeviceContext>()?.simulator.value()?.into()
+            }
+            "contexts.os.build" => self.context::<OsContext>()?.build.as_str()?.into(),
+            "contexts.os.kernel_version" => {
+                self.context::<OsContext>()?.kernel_version.as_str()?.into()
+            }
+            "contexts.os.name" => self.context::<OsContext>()?.name.as_str()?.into(),
+            "contexts.os.version" => self.context::<OsContext>()?.version.as_str()?.into(),
+            "contexts.browser.name" => self.context::<BrowserContext>()?.name.as_str()?.into(),
+            "contexts.browser.version" => {
+                self.context::<BrowserContext>()?.version.as_str()?.into()
+            }
+            "contexts.profile.profile_id" => self
+                .context::<ProfileContext>()?
+                .profile_id
+                .value()?
+                .0
+                .into(),
+            "contexts.device.uuid" => self.context::<DeviceContext>()?.uuid.value()?.into(),
+            "contexts.trace.status" => self
+                .context::<TraceContext>()?
+                .status
+                .value()?
+                .as_str()
+                .into(),
+            "contexts.trace.op" => self.context::<TraceContext>()?.op.as_str()?.into(),
+            "contexts.response.status_code" => self
+                .context::<ResponseContext>()?
+                .status_code
+                .value()?
+                .into(),
+            "contexts.unreal.crash_type" => match self.contexts.value()?.get_key("unreal")? {
+                super::Context::Other(context) => context.get("crash_type")?.value()?.into(),
+                _ => return None,
+            },
+
+            // Dynamic access to certain data bags
+            path => {
+                if let Some(rest) = path.strip_prefix("tags.") {
+                    self.tags.value()?.get(rest)?.into()
+                } else if let Some(rest) = path.strip_prefix("request.headers.") {
+                    self.request
+                        .value()?
+                        .headers
+                        .value()?
+                        .get_header(rest)?
+                        .into()
+                } else {
+                    return None;
+                }
+            }
+        })
+    }
+}
+
+fn or_none(string: &Annotated<impl AsRef<str>>) -> Option<&str> {
+    match string.as_str() {
+        None | Some("") => None,
+        Some(other) => Some(other),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use chrono::{TimeZone, Utc};

relay-filter/src/interface.rs

@@ -2,7 +2,7 @@
 //! the implementation for [`Event`].
 use url::Url;
 
-use relay_event_schema::protocol::{Csp, Event, EventType, Exception, LogEntry, Values};
+use relay_event_schema::protocol::{Csp, Event, EventType, Exception, LogEntry, Replay, Values};
 
 /// A data item to which filters can be applied.
 pub trait Filterable {
@@ -72,3 +72,39 @@ impl Filterable for Event {
         self.user_agent()
     }
 }
+
+impl Filterable for Replay {
+    fn csp(&self) -> Option<&Csp> {
+        None
+    }
+
+    fn exceptions(&self) -> Option<&Values<Exception>> {
+        None
+    }
+
+    fn ip_addr(&self) -> Option<&str> {
+        let user = self.user.value()?;
+        Some(user.ip_address.value()?.as_ref())
+    }
+
+    fn logentry(&self) -> Option<&LogEntry> {
+        None
+    }
+
+    fn release(&self) -> Option<&str> {
+        self.release.as_str()
+    }
+
+    fn transaction(&self) -> Option<&str> {
+        None
+    }
+
+    fn url(&self) -> Option<Url> {
+        let url_str = self.request.value()?.url.value()?;
+        Url::parse(url_str).ok()
+    }
+
+    fn user_agent(&self) -> Option<&str> {
+        self.user_agent()
+    }
+}

relay-server/src/services/processor.rs

@@ -462,11 +462,14 @@ pub enum ProcessingError {
 
     #[error("invalid replay")]
     InvalidReplay(DiscardReason),
+
+    #[error("replay filtered with reason: {0:?}")]
+    ReplayFiltered(FilterStatKey),
 }
 
 impl ProcessingError {
     fn to_outcome(&self) -> Option<Outcome> {
-        match *self {
+        match self {
             // General outcomes for invalid events
             Self::PayloadTooLarge => Some(Outcome::Invalid(DiscardReason::TooLarge)),
             Self::InvalidJson(_) => Some(Outcome::Invalid(DiscardReason::InvalidJson)),
@@ -505,7 +508,8 @@ impl ProcessingError {
             Self::EventFiltered(_) => None,
             Self::InvalidProcessingGroup(_) => None,
 
-            Self::InvalidReplay(reason) => Some(Outcome::Invalid(reason)),
+            Self::InvalidReplay(reason) => Some(Outcome::Invalid(*reason)),
+            Self::ReplayFiltered(key) => Some(Outcome::Filtered(key.clone())),
         }
     }
 
@@ -1533,7 +1537,11 @@ impl EnvelopeProcessorService {
         &self,
         state: &mut ProcessEnvelopeState<ReplayGroup>,
     ) -> Result<(), ProcessingError> {
-        replay::process(state, &self.inner.config)?;
+        replay::process(
+            state,
+            &self.inner.config,
+            &self.inner.global_config.current(),
+        )?;
         if_processing!(self.inner.config, {
             self.enforce_quotas(state)?;
         });