fix(relay): Limit maximum number of otel logs deserialized from protos

relay-config/src/config.rs

@@ -639,6 +639,8 @@ pub struct Limits {
     pub max_trace_metric_size: ByteSize,
     /// The maximum payload size for a log.
     pub max_log_size: ByteSize,
+    /// The maximum number of logs that can result from a log expansion.
+    pub max_expanded_log_count: usize,
     /// The maximum payload size for a span.
     pub max_span_size: ByteSize,
     /// The maximum payload size for an item container.
@@ -728,6 +730,7 @@ impl Default for Limits {
             max_profile_size: ByteSize::mebibytes(50),
             max_trace_metric_size: ByteSize::mebibytes(1),
             max_log_size: ByteSize::mebibytes(1),
+            max_expanded_log_count: 1000,
             max_span_size: ByteSize::mebibytes(10),
             max_container_size: ByteSize::mebibytes(12),
             max_statsd_size: ByteSize::mebibytes(1),
@@ -2412,6 +2415,11 @@ impl Config {
         self.values.limits.max_log_size.as_bytes()
     }
 
+    /// Returns the maximum number of logs that can result from a log expansion.
+    pub fn max_expanded_log_count(&self) -> usize {
+        self.values.limits.max_expanded_log_count
+    }
+
     /// Returns the maximum payload size of a span in bytes.
     pub fn max_span_size(&self) -> usize {
         self.values.limits.max_span_size.as_bytes()

relay-server/src/processing/logs/integrations/mod.rs

@@ -8,6 +8,8 @@ use crate::processing::logs::Settings;
 
 mod nel;
 mod otel;
+mod otel_json_deserializer;
+mod otel_proto_deserializer;
 mod vercel;
 
 /// Expands a log [`Integration`] into a list of logs.
@@ -17,6 +19,7 @@ pub fn expand(
     item: Item,
     records: &mut RecordKeeper<'_>,
     headers: &EnvelopeHeaders,
+    max_expanded_log_count: usize,
 ) -> Option<(Settings, ContainerItems<OurLog>)> {
     let integration = match item.integration() {
         Some(Integration::Logs(integration)) => integration,
@@ -46,7 +49,9 @@ pub fn expand(
 
     let settings = match integration {
         LogsIntegration::Nel => nel::expand(&payload, headers, produce),
-        LogsIntegration::OtelV1 { format } => otel::expand(format, &payload, produce),
+        LogsIntegration::OtelV1 { format } => {
+            otel::expand(format, &payload, max_expanded_log_count, produce)
+        }
         LogsIntegration::VercelDrainLog { format } => vercel::expand(format, &payload, produce),
     };
     let settings = match settings {

relay-server/src/processing/logs/integrations/otel.rs

@@ -1,17 +1,23 @@
-use opentelemetry_proto::tonic::logs::v1::LogsData;
-use prost::Message as _;
-use relay_event_schema::protocol::OurLog;
-
 use crate::integrations::OtelFormat;
-use crate::processing::logs::{Error, Result, Settings};
+use crate::processing::logs::{
+    Error, Result, Settings,
+    integrations::{otel_json_deserializer, otel_proto_deserializer},
+};
 use crate::services::outcome::DiscardReason;
+use opentelemetry_proto::tonic::logs::v1::LogsData;
+use relay_event_schema::protocol::OurLog;
 
 /// Expands OTeL logs into the [`OurLog`] format.
-pub fn expand<F>(format: OtelFormat, payload: &[u8], mut produce: F) -> Result<Settings>
+pub fn expand<F>(
+    format: OtelFormat,
+    payload: &[u8],
+    max_logs: usize,
+    mut produce: F,
+) -> Result<Settings>
 where
     F: FnMut(OurLog),
 {
-    let logs = parse_logs_data(format, payload)?;
+    let logs = parse_logs_data(format, payload, max_logs)?;
 
     for resource_logs in logs.resource_logs {
         let resource = resource_logs.resource.as_ref();
@@ -27,21 +33,129 @@ where
     Ok(Settings::default())
 }
 
-fn parse_logs_data(format: OtelFormat, payload: &[u8]) -> Result<LogsData, Error> {
+fn parse_logs_data(format: OtelFormat, payload: &[u8], max_logs: usize) -> Result<LogsData, Error> {
     match format {
-        OtelFormat::Json => serde_json::from_slice(payload).map_err(|e| {
+        OtelFormat::Json => otel_json_deserializer::deserialize(payload, max_logs).map_err(|e| {
             relay_log::debug!(
                 error = &e as &dyn std::error::Error,
                 "Failed to parse logs data as JSON"
             );
             Error::Invalid(DiscardReason::InvalidJson)
         }),
-        OtelFormat::Protobuf => LogsData::decode(payload).map_err(|e| {
-            relay_log::debug!(
-                error = &e as &dyn std::error::Error,
-                "Failed to parse logs data as protobuf"
-            );
-            Error::Invalid(DiscardReason::InvalidProtobuf)
-        }),
+        OtelFormat::Protobuf => {
+            otel_proto_deserializer::deserialize(payload, max_logs).map_err(|e| {
+                relay_log::debug!(
+                    error = &e as &dyn std::error::Error,
+                    "Failed to parse logs data as protobuf"
+                );
+                Error::Invalid(DiscardReason::InvalidProtobuf)
+            })
+        }
+    }
+}
+#[cfg(test)]
+mod tests {
+
+    use opentelemetry_proto::tonic::common::v1::any_value::Value;
+    use opentelemetry_proto::tonic::common::v1::{
+        AnyValue, ArrayValue, InstrumentationScope, KeyValue,
+    };
+    use opentelemetry_proto::tonic::resource::v1::Resource;
+    use relay_ourlogs::otel_logs::{LogRecord, LogsData, ResourceLogs, ScopeLogs};
+
+    use crate::processing::logs::integrations::otel::parse_logs_data;
+
+    #[test]
+    fn test_basic_json() {
+        let log_data = LogsData {
+            resource_logs: vec![ResourceLogs {
+                resource: Some(Resource {
+                    attributes: vec![KeyValue {
+                        key: "service.name".to_owned(),
+                        value: Some(AnyValue {
+                            value: Some(Value::StringValue("test-service".to_owned())),
+                        }),
+                    }],
+                    dropped_attributes_count: 0,
+                    entity_refs: vec![],
+                }),
+                scope_logs: vec![ScopeLogs {
+                    scope: Some(InstrumentationScope {
+                        name: "test-library".to_owned(),
+                        version: "".to_owned(),
+                        attributes: vec![],
+                        dropped_attributes_count: 0,
+                    }),
+                    log_records: vec![LogRecord {
+                        time_unix_nano: 123,
+                        observed_time_unix_nano: 123,
+                        severity_number: 2,
+                        severity_text: "Information".to_owned(),
+                        body: Some(AnyValue {
+                            value: Some(Value::StringValue("a body".to_owned())),
+                        }),
+                        attributes: vec![
+                            KeyValue {
+                                key: "attribute".to_owned(),
+                                value: Some(AnyValue {
+                                    value: Some(Value::StringValue("value".to_owned())),
+                                }),
+                            },
+                            KeyValue {
+                                key: "nested attribute".to_owned(),
+                                value: Some(AnyValue {
+                                    value: Some(Value::ArrayValue(ArrayValue {
+                                        values: vec![AnyValue {
+                                            value: Some(Value::StringValue("value".to_owned())),
+                                        }],
+                                    })),
+                                }),
+                            },
+                        ],
+                        dropped_attributes_count: 0,
+                        flags: 0,
+                        trace_id: "5B8EFFF798038103D269B633813FC60C".into(),
+                        span_id: "EEE19B7EC3C1B174".into(),
+                        event_name: "".to_owned(),
+                    }],
+                    schema_url: "".to_owned(),
+                }],
+                schema_url: "http://example.com".to_owned(),
+            }],
+        };
+
+        let json = serde_json::to_string(&log_data).unwrap();
+
+        let unjson: LogsData = serde_json::from_str(&json).unwrap();
+
+        assert_eq!(unjson, log_data);
+    }
+
+    #[test]
+    fn test_abusive_json() {
+        let mut abusive_log = "{},".repeat(1_001);
+        abusive_log.pop();
+
+        let json = r#"{
+        "resourceLogs": [
+            {
+                "resource": {
+                    "attributes": [
+                        {
+                            "key": "service.name",
+                            "value": {"stringValue": "test-service"}
+                        }
+                    ]
+                },
+                "scopeLogs": [
+                    {
+                        "scope": {"name": "test-library"},
+                        "logRecords": ["#
+            .to_owned();
+        let json = json + &abusive_log + "]}]}]}";
+
+        assert!(
+            parse_logs_data(crate::integrations::OtelFormat::Json, json.as_bytes(), 1000).is_err()
+        );
     }
 }