feat(splunk_hec source): Allow content-type header if it includes application/json (#23024)

* #23022 Allow content-type header if it only includes application/json. Do not require exact match

* Formatting

* Ensuring response with invalid Content-Type header value will return same error response

* Adding changelog

* Adding newline

* Fixed failing tests and added new unit test

Author: tot19

changelog.d/23022_hec_allow_content_header_contains.enhancement.md

@@ -0,0 +1,3 @@
+Updated the Splunk HEC source to accept requests that contain the header content-type with any value containing "application/json," not the exact value of "application/json." This matches the behavior of a true Splunk HEC. Allows sources from AWS to successfully send events to the Splunk HEC source without additional proxying to update headers.
+
+authors: Tot19

src/sources/splunk_hec/mod.rs

@@ -13,6 +13,7 @@ use flate2::read::MultiGzDecoder;
 use futures::FutureExt;
 use http::StatusCode;
 use hyper::{service::make_service_fn, Server};
+use serde::de::DeserializeOwned;
 use serde::Serialize;
 use serde_json::{
     de::{Read as JsonRead, StrRead},
@@ -35,6 +36,7 @@ use vector_lib::{
 use vector_lib::{configurable::configurable_component, tls::MaybeTlsIncomingStream};
 use vrl::path::OwnedTargetPath;
 use vrl::value::{kind::Collection, Kind};
+use warp::http::header::{HeaderValue, CONTENT_TYPE};
 use warp::{filters::BoxedFilter, path, reject::Rejection, reply::Response, Filter, Reply};
 
 use self::{
@@ -529,26 +531,50 @@ impl SplunkSource {
             .boxed()
     }
 
+    fn lenient_json_content_type_check<T>() -> impl Filter<Extract = (T,), Error = Rejection> + Clone
+    where
+        T: Send + DeserializeOwned + 'static,
+    {
+        warp::header::optional::<HeaderValue>(CONTENT_TYPE.as_str())
+            .and(warp::body::bytes())
+            .and_then(
+                |ctype: Option<HeaderValue>, body: bytes::Bytes| async move {
+                    let ok = ctype
+                        .as_ref()
+                        .and_then(|v| v.to_str().ok())
+                        .map(|h| h.to_ascii_lowercase().contains("application/json"))
+                        .unwrap_or(true);
+
+                    if !ok {
+                        return Err(warp::reject::custom(ApiError::UnsupportedContentType));
+                    }
+
+                    let value = serde_json::from_slice::<T>(&body)
+                        .map_err(|_| warp::reject::custom(ApiError::BadRequest))?;
+
+                    Ok(value)
+                },
+            )
+    }
+
     fn ack_service(&self) -> BoxedFilter<(Response,)> {
         let idx_ack = self.idx_ack.clone();
+
         warp::post()
-            .and(path!("ack"))
+            .and(warp::path!("ack"))
             .and(self.authorization())
             .and(SplunkSource::required_channel())
-            .and(warp::body::json())
-            .and_then(move |_, channel_id: String, body: HecAckStatusRequest| {
+            .and(Self::lenient_json_content_type_check::<HecAckStatusRequest>())
+            .and_then(move |_, channel: String, req: HecAckStatusRequest| {
                 let idx_ack = idx_ack.clone();
                 async move {
                     if let Some(idx_ack) = idx_ack {
-                        let ack_statuses = idx_ack
-                            .get_acks_status_from_channel(channel_id, &body.acks)
+                        let acks = idx_ack
+                            .get_acks_status_from_channel(channel, &req.acks)
                             .await?;
-                        Ok(
-                            warp::reply::json(&HecAckStatusResponse { acks: ack_statuses })
-                                .into_response(),
-                        )
+                        Ok(warp::reply::json(&HecAckStatusResponse { acks }).into_response())
                     } else {
-                        Err(Rejection::from(ApiError::AckIsDisabled))
+                        Err(warp::reject::custom(ApiError::AckIsDisabled))
                     }
                 }
             })
@@ -1094,6 +1120,7 @@ pub(crate) enum ApiError {
     MissingAuthorization,
     InvalidAuthorization,
     UnsupportedEncoding,
+    UnsupportedContentType,
     MissingChannel,
     NoData,
     InvalidDataFormat { event: usize },
@@ -1188,6 +1215,14 @@ fn finish_ok(maybe_ack_id: Option<u64>) -> Response {
     response_json(StatusCode::OK, body)
 }
 
+fn response_plain(code: StatusCode, msg: &'static str) -> Response {
+    warp::reply::with_status(
+        warp::reply::with_header(msg, http::header::CONTENT_TYPE, "text/plain; charset=utf-8"),
+        code,
+    )
+    .into_response()
+}
+
 async fn finish_err(rejection: Rejection) -> Result<(Response,), Rejection> {
     if let Some(&error) = rejection.find::<ApiError>() {
         emit!(SplunkHecRequestError { error });
@@ -1200,6 +1235,10 @@ async fn finish_err(rejection: Rejection) -> Result<(Response,), Rejection> {
                 splunk_response::INVALID_AUTHORIZATION,
             ),
             ApiError::UnsupportedEncoding => empty_response(StatusCode::UNSUPPORTED_MEDIA_TYPE),
+            ApiError::UnsupportedContentType => response_plain(
+                StatusCode::UNSUPPORTED_MEDIA_TYPE,
+                "The request's content-type is not supported",
+            ),
             ApiError::MissingChannel => {
                 response_json(StatusCode::BAD_REQUEST, splunk_response::NO_CHANNEL)
             }
@@ -2485,6 +2524,51 @@ mod tests {
         assert!(ack_res.acks.get(&event_res.ack_id).unwrap());
     }
 
+    #[tokio::test]
+    async fn ack_service_accepts_parameterized_content_type() {
+        let ack_config = HecAcknowledgementsConfig {
+            enabled: Some(true),
+            ..Default::default()
+        };
+        let (source, address) = source(Some(ack_config)).await;
+        let opts = SendWithOpts {
+            channel: Some(Channel::Header("guid")),
+            forwarded_for: None,
+        };
+
+        let event_res = send_with_response(
+            address,
+            "services/collector/event",
+            r#"{"event":"param-test"}"#,
+            TOKEN,
+            &opts,
+        )
+        .await
+        .json::<HecAckEventResponse>()
+        .await
+        .unwrap();
+        let _ = collect_n(source, 1).await;
+
+        let body = serde_json::to_string(&HecAckStatusRequest {
+            acks: vec![event_res.ack_id],
+        })
+        .unwrap();
+
+        let res = reqwest::Client::new()
+            .post(format!("http://{}/services/collector/ack", address))
+            .header("Authorization", format!("Splunk {}", TOKEN))
+            .header("x-splunk-request-channel", "guid")
+            .header("Content-Type", "application/json; some-random-text; hello")
+            .body(body)
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(200, res.status().as_u16());
+
+        let _parsed: HecAckStatusResponse = res.json().await.unwrap();
+    }
+
     #[tokio::test]
     async fn event_service_acknowledgements_enabled_channel_required() {
         let message = r#"{"event":"first", "color": "blue"}"#;