fix: accept parameterized OTLP content types

src/handlers/http/modal/utils/ingest_utils.rs

@@ -17,6 +17,7 @@
  */
 
 use actix_web::http::header::USER_AGENT;
+use actix_web::mime::Mime;
 use actix_web::{HttpRequest, web};
 use chrono::Utc;
 use opentelemetry_proto::tonic::{
@@ -121,6 +122,33 @@ pub async fn process_otel_content(
     }
 }
 
+/// Payload encodings accepted by the OTEL HTTP endpoints.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum OtelContentType {
+    Json,
+    Protobuf,
+}
+
+/// Classifies a raw `Content-Type` header value by its media-type essence
+/// (`type/subtype`) only. Everything from the first `;` onwards is ignored,
+/// including malformed parameter text, so `application/json; charset=utf-8`
+/// is treated as JSON. Only HTTP optional whitespace (SP/HTAB) around the
+/// essence is trimmed; type and subtype are compared case-insensitively.
+/// Structured-syntax suffixes such as `application/otlp+json` are not aliases.
+fn classify_otel_content_type(content_type: &str) -> Option<OtelContentType> {
+    let essence = content_type
+        .split(';')
+        .next()
+        .unwrap_or_default()
+        .trim_matches([' ', '\t']);
+    let mime: Mime = essence.parse().ok()?;
+    match mime.essence_str() {
+        CONTENT_TYPE_JSON => Some(OtelContentType::Json),
+        CONTENT_TYPE_PROTOBUF => Some(OtelContentType::Protobuf),
+        _ => None,
+    }
+}
+
 #[allow(clippy::too_many_arguments)]
 fn handle_otel_ingestion(
     body: web::Bytes,
@@ -132,8 +160,8 @@ fn handle_otel_ingestion(
     tenant_id: Option<String>,
 ) -> Result<(), PostError> {
     match content_type {
-        Some(content_type) => {
-            if content_type == CONTENT_TYPE_JSON {
+        Some(content_type) => match classify_otel_content_type(&content_type) {
+            Some(OtelContentType::Json) => {
                 let json: serde_json::Value = match serde_json::from_slice(&body) {
                     Ok(v) => v,
                     Err(e) => {
@@ -155,22 +183,24 @@ fn handle_otel_ingestion(
                     error!("Ingestion failed for stream {stream_name}: {e}");
                     return Err(e);
                 }
-            } else if content_type == CONTENT_TYPE_PROTOBUF {
+            }
+            Some(OtelContentType::Protobuf) => {
                 error!(
                     "Ingestion failed for stream {stream_name}: Protobuf ingestion is not supported in Parseable OSS"
                 );
                 return Err(PostError::Invalid(anyhow::anyhow!(
                     "Ingestion failed for stream {stream_name}: Protobuf ingestion is not supported in Parseable OSS"
                 )));
-            } else {
+            }
+            None => {
                 error!(
                     "Ingestion failed for stream {stream_name}: Unsupported Content-Type: {content_type}. Expected application/json or application/x-protobuf"
                 );
                 return Err(PostError::Invalid(anyhow::anyhow!(
                     "Ingestion failed for stream {stream_name}: Unsupported Content-Type: {content_type}. Expected application/json or application/x-protobuf"
                 )));
             }
-        }
+        },
         None => {
             error!(
                 "Ingestion failed for stream {stream_name}: Missing Content-Type header. Expected application/json or application/x-protobuf"
@@ -628,6 +658,216 @@ mod tests {
         assert_eq!(custom_fields.get(SOURCE_IP_KEY).unwrap(), "unknown");
     }
 
+    #[test]
+    fn test_classify_otel_content_type_by_essence_only() {
+        use OtelContentType::{Json, Protobuf};
+
+        let cases: &[(&str, Option<OtelContentType>)] = &[
+            // bare media types
+            ("application/json", Some(Json)),
+            ("application/x-protobuf", Some(Protobuf)),
+            // charset parameter, plain and quoted
+            ("application/json; charset=utf-8", Some(Json)),
+            ("application/json;charset=utf-8", Some(Json)),
+            ("application/json; charset=\"utf-8\"", Some(Json)),
+            // unknown and multiple parameters
+            ("application/json; foo=bar", Some(Json)),
+            ("application/json; charset=utf-8; foo=bar", Some(Json)),
+            // type/subtype are case-insensitive
+            ("APPLICATION/JSON", Some(Json)),
+            ("Application/Json; Charset=UTF-8", Some(Json)),
+            // outer SP/HTAB around the whole value
+            ("  application/json  ", Some(Json)),
+            ("\tapplication/json\t", Some(Json)),
+            // SP/HTAB around the parameter separator
+            ("application/json ; charset=utf-8", Some(Json)),
+            ("application/json\t;\tcharset=utf-8", Some(Json)),
+            (" \tapplication/json \t; charset=utf-8", Some(Json)),
+            // malformed parameter text is ignored (essence-only validation)
+            ("application/json;", Some(Json)),
+            ("application/json;;", Some(Json)),
+            ("application/json; charset", Some(Json)),
+            ("application/json; charset=", Some(Json)),
+            ("application/json; charset=\"utf-8", Some(Json)),
+            ("application/json; =utf-8", Some(Json)),
+            ("application/json; a b c", Some(Json)),
+            // parameterised and case-varied protobuf still classifies as protobuf
+            ("application/x-protobuf; charset=utf-8", Some(Protobuf)),
+            ("Application/X-Protobuf", Some(Protobuf)),
+            ("APPLICATION/X-PROTOBUF ; proto=foo", Some(Protobuf)),
+            // whitespace inside the essence is rejected
+            ("application /json", None),
+            ("application/ json", None),
+            ("application\t/json", None),
+            ("application/js on", None),
+            // malformed essences
+            ("", None),
+            ("   ", None),
+            (";charset=utf-8", None),
+            ("application", None),
+            ("application/", None),
+            ("/json", None),
+            ("application/json/extra", None),
+            ("\"application/json\"", None),
+            // unsupported media types and suffix aliases
+            ("*/*", None),
+            ("text/plain", None),
+            ("application/xml", None),
+            ("application/protobuf", None),
+            ("application/jsonl", None),
+            ("application/otlp+json", None),
+            ("application/vnd.api+json", None),
+        ];
+
+        for (input, expected) in cases {
+            assert_eq!(
+                classify_otel_content_type(input),
+                *expected,
+                "classification of {input:?}"
+            );
+        }
+    }
+
+    const OTEL_TEST_STREAM: &str = "otel-content-type-test";
+    const OTEL_SIGNALS: [(TelemetryType, LogSource); 3] = [
+        (TelemetryType::Logs, LogSource::OtelLogs),
+        (TelemetryType::Metrics, LogSource::OtelMetrics),
+        (TelemetryType::Traces, LogSource::OtelTraces),
+    ];
+
+    // Malformed JSON body: an accepted JSON content type fails in serde
+    // before any PARSEABLE/storage access, so no global state is needed.
+    async fn run_otel_content(
+        content_type: Option<actix_web::http::header::HeaderValue>,
+        telemetry_type: TelemetryType,
+        log_source: &LogSource,
+    ) -> PostError {
+        let mut req = TestRequest::default();
+        if let Some(value) = content_type {
+            req = req.insert_header((actix_web::http::header::CONTENT_TYPE, value));
+        }
+        let req = req.to_http_request();
+        process_otel_content(
+            &req,
+            web::Bytes::from_static(b"{not valid json"),
+            OTEL_TEST_STREAM,
+            log_source,
+            telemetry_type,
+        )
+        .await
+        .expect_err("malformed body or rejected content type must fail")
+    }
+
+    async fn assert_bad_request_body(err: &PostError, expected_body: &str) {
+        use actix_web::ResponseError;
+
+        assert_eq!(err.status_code(), actix_web::http::StatusCode::BAD_REQUEST);
+        let resp = err.error_response();
+        assert_eq!(resp.status(), actix_web::http::StatusCode::BAD_REQUEST);
+        assert_eq!(
+            resp.headers()
+                .get(actix_web::http::header::CONTENT_TYPE)
+                .unwrap(),
+            "text/plain; charset=utf-8"
+        );
+        let body = actix_web::body::to_bytes(resp.into_body()).await.unwrap();
+        assert_eq!(std::str::from_utf8(&body).unwrap(), expected_body);
+    }
+
+    fn header(value: &'static str) -> Option<actix_web::http::header::HeaderValue> {
+        Some(actix_web::http::header::HeaderValue::from_static(value))
+    }
+
+    #[tokio::test]
+    async fn test_process_otel_content_accepts_parameterised_json() {
+        let accepted = [
+            "application/json",
+            "application/json; charset=utf-8",
+            "application/json; charset=\"utf-8\"",
+            "Application/JSON ; foo=bar; charset=utf-8",
+            "\tapplication/json\t;\tcharset=utf-8",
+            "application/json; charset=\"unterminated",
+        ];
+        for (telemetry_type, log_source) in OTEL_SIGNALS {
+            for value in accepted {
+                let err = run_otel_content(header(value), telemetry_type, &log_source).await;
+                assert!(
+                    matches!(err, PostError::SerdeError(_)),
+                    "{log_source} with {value:?} should reach JSON decoding, got {err:?}"
+                );
+                let expected = err.to_string();
+                assert!(expected.starts_with("Could not deserialize into JSON object, "));
+                assert_bad_request_body(&err, &expected).await;
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_process_otel_content_missing_or_non_text_content_type() {
+        let non_text =
+            actix_web::http::header::HeaderValue::from_bytes(b"application/json\xff").unwrap();
+        for (telemetry_type, log_source) in OTEL_SIGNALS {
+            for content_type in [None, Some(non_text.clone())] {
+                let err = run_otel_content(content_type, telemetry_type, &log_source).await;
+                assert!(matches!(err, PostError::Invalid(_)), "{err:?}");
+                assert_bad_request_body(
+                    &err,
+                    &format!(
+                        "Invalid Request: Ingestion failed for stream {OTEL_TEST_STREAM}: Missing Content-Type header. Expected application/json or application/x-protobuf"
+                    ),
+                )
+                .await;
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_process_otel_content_unsupported_reports_original_value() {
+        let unsupported = [
+            "text/plain; charset=utf-8",
+            "application/otlp+json",
+            "application /json; charset=utf-8",
+            "  Application/XML ; q=1",
+            "",
+        ];
+        for (telemetry_type, log_source) in OTEL_SIGNALS {
+            for value in unsupported {
+                let err = run_otel_content(header(value), telemetry_type, &log_source).await;
+                assert!(matches!(err, PostError::Invalid(_)), "{err:?}");
+                assert_bad_request_body(
+                    &err,
+                    &format!(
+                        "Invalid Request: Ingestion failed for stream {OTEL_TEST_STREAM}: Unsupported Content-Type: {value}. Expected application/json or application/x-protobuf"
+                    ),
+                )
+                .await;
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn test_process_otel_content_protobuf_variants_still_unsupported() {
+        let protobuf = [
+            "application/x-protobuf",
+            "application/x-protobuf; charset=utf-8",
+            "Application/X-Protobuf",
+            " APPLICATION/X-PROTOBUF\t;proto=foo",
+        ];
+        for (telemetry_type, log_source) in OTEL_SIGNALS {
+            for value in protobuf {
+                let err = run_otel_content(header(value), telemetry_type, &log_source).await;
+                assert!(matches!(err, PostError::Invalid(_)), "{err:?}");
+                assert_bad_request_body(
+                    &err,
+                    &format!(
+                        "Invalid Request: Ingestion failed for stream {OTEL_TEST_STREAM}: Protobuf ingestion is not supported in Parseable OSS"
+                    ),
+                )
+                .await;
+            }
+        }
+    }
+
     #[test]
     fn test_get_custom_fields_from_header_with_peer_address() {
         let req = TestRequest::default()