fix: reject malformed headers without panicking workers

Cargo.toml

@@ -221,6 +221,7 @@ anyhow = "1.0"
 procfs = { version = "0.18.0", default-features = false }
 
 [dev-dependencies]
+proptest = { version = "=1.9.0", default-features = false, features = ["std"] }
 rstest = "0.26.1"
 arrow = "59.2.0"
 temp-dir = "0.1.14"

src/alerts/mod.rs

@@ -1263,6 +1263,8 @@ pub enum AlertError {
     ValidationFailure(String),
     #[error(transparent)]
     MetastoreError(#[from] MetastoreError),
+    #[error("{0}")]
+    Header(#[from] crate::utils::header_parsing::ParseHeaderError),
 }
 
 impl From<outbound_http_policy::OutboundPolicyError> for AlertError {
@@ -1298,6 +1300,7 @@ impl actix_web::ResponseError for AlertError {
             Self::NotPresentInOSS(_) => StatusCode::BAD_REQUEST,
             Self::MetastoreError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::OutboundPolicy(_) => StatusCode::BAD_REQUEST,
+            Self::Header(_) => StatusCode::BAD_REQUEST,
         }
     }
 
@@ -1972,3 +1975,19 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+mod header_error_tests {
+    use actix_web::ResponseError;
+
+    use super::*;
+    use crate::utils::header_parsing::ParseHeaderError;
+
+    #[test]
+    fn header_error_is_bad_request() {
+        assert_eq!(
+            AlertError::from(ParseHeaderError::InvalidTenantId).status_code(),
+            StatusCode::BAD_REQUEST
+        );
+    }
+}

src/apikeys.rs

@@ -98,8 +98,36 @@ impl actix_web::ResponseError for ApiKeyError {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::{
+        handlers::http::rbac::RBACError,
+        utils::header_parsing::{HeaderBuildError, ParseHeaderError},
+    };
+    use actix_web::{ResponseError, http::StatusCode};
     use serde_json::json;
 
+    #[test]
+    fn sync_header_errors_preserve_status_without_revocation_wrapping() {
+        let error = ApiKeyError::from(RBACError::HeaderBuild(HeaderBuildError::Internal(
+            "configured username",
+        )));
+        assert_eq!(error.status_code(), StatusCode::INTERNAL_SERVER_ERROR);
+        assert_eq!(
+            error.to_string(),
+            "Failed to build internal request headers"
+        );
+
+        let error = ApiKeyError::from(RBACError::HeaderBuild(HeaderBuildError::Input(
+            ParseHeaderError::InvalidTenantId,
+        )));
+        assert_eq!(error.status_code(), StatusCode::BAD_REQUEST);
+
+        let network_failure = ApiKeyError::RevocationSyncFailed("network failure".into());
+        assert_eq!(
+            network_failure.status_code(),
+            StatusCode::SERVICE_UNAVAILABLE
+        );
+    }
+
     #[test]
     fn ttl_accepts_inclusive_boundaries_and_no_expiration() {
         for ttl in [json!(null), json!(1800), json!(7_776_000)] {

src/event/format/known_schema.rs

@@ -43,6 +43,10 @@ pub enum Error {
         "Unsupported log format: '{0}'. This format cannot be parsed by the current version. Please create an issue on our GitHub repository (github.com/parseablehq/parseable) with a sample log event, or reach out in the #support-channel on our Slack community for assistance. Include this error message and your log sample to help us improve compatibility."
     )]
     Unknown(String),
+    #[error(
+        "Event shape is not supported for known log formats. Expected a JSON object or an array of JSON objects."
+    )]
+    InvalidShape,
 }
 
 /// Deserializes a string pattern into a compiled Regex
@@ -193,6 +197,8 @@ impl EventProcessor {
     /// # Returns
     /// * `Ok` - The original JSON will now contain extracted fields
     /// * `Err(Unacceptable)` - JSON provided is acceptable for the known format
+    /// * `Err(Unknown)` - `log_source` is not a known format
+    /// * `Err(InvalidShape)` - JSON is neither an object nor an array
     pub fn extract_from_inline_log(
         &self,
         json: &mut Value,
@@ -227,7 +233,9 @@ impl EventProcessor {
                     p_custom_fields.insert(FORMAT_VERIFY_KEY.to_string(), "false".to_string());
                 }
             }
-            _ => unreachable!("We don't accept events of the form: {json}"),
+            // Scalars (number, string, bool, null) can't carry log fields.
+            // The value is deliberately not echoed back in the error.
+            _ => return Err(Error::InvalidShape),
         }
 
         Ok(fields)
@@ -470,6 +478,145 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_extract_from_inline_log_rejects_scalars() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+
+        for scalar in [
+            json!(42),
+            json!(1.5),
+            json!("[ERROR] [2023-10-10T13:55:36Z] do-not-echo-this"),
+            json!(true),
+            Value::Null,
+        ] {
+            let mut json_value = scalar.clone();
+            let mut p_custom_fields = HashMap::new();
+            let result = processor.extract_from_inline_log(
+                &mut json_value,
+                &mut p_custom_fields,
+                "custom_app_log",
+                Some("raw_log"),
+            );
+
+            let err = result.expect_err("scalar events must be rejected");
+            assert!(matches!(err, Error::InvalidShape), "{scalar} gave {err:?}");
+            assert!(!err.to_string().contains("do-not-echo-this"));
+            assert_eq!(json_value, scalar, "scalar event must be left untouched");
+            assert!(p_custom_fields.is_empty());
+            assert_eq!(
+                actix_web::ResponseError::status_code(
+                    &crate::handlers::http::ingest::PostError::from(err)
+                ),
+                actix_web::http::StatusCode::BAD_REQUEST
+            );
+        }
+    }
+
+    #[test]
+    fn test_extract_from_inline_log_unknown_source_before_shape() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+        let mut json_value = json!(42);
+        let mut p_custom_fields = HashMap::new();
+
+        let result = processor.extract_from_inline_log(
+            &mut json_value,
+            &mut p_custom_fields,
+            "nonexistent_format",
+            Some("raw_log"),
+        );
+
+        assert!(matches!(result, Err(Error::Unknown(name)) if name == "nonexistent_format"));
+    }
+
+    #[test]
+    fn test_extract_from_inline_log_object_via_processor() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+        let mut json_value = json!({
+            "raw_log": "[ERROR] [2023-10-10T13:55:36Z] Failed to connect to database"
+        });
+        let mut p_custom_fields = HashMap::new();
+
+        let fields = processor
+            .extract_from_inline_log(
+                &mut json_value,
+                &mut p_custom_fields,
+                "custom_app_log",
+                Some("raw_log"),
+            )
+            .unwrap();
+
+        assert!(fields.contains("level"));
+        assert_eq!(json_value["level"], "ERROR");
+        assert!(p_custom_fields.is_empty());
+    }
+
+    #[test]
+    fn test_extract_from_inline_log_object_no_match_marks_unverified() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+        let mut json_value = json!({ "raw_log": "not a known format" });
+        let mut p_custom_fields = HashMap::new();
+
+        let fields = processor
+            .extract_from_inline_log(
+                &mut json_value,
+                &mut p_custom_fields,
+                "custom_app_log",
+                Some("raw_log"),
+            )
+            .unwrap();
+
+        assert!(fields.is_empty());
+        assert_eq!(
+            p_custom_fields.get(FORMAT_VERIFY_KEY).map(String::as_str),
+            Some("false")
+        );
+    }
+
+    #[test]
+    fn test_extract_from_inline_log_array_skips_non_objects() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+        let mut json_value = json!([
+            42,
+            "plain string",
+            null,
+            { "raw_log": "[INFO] [2023-10-10T13:55:40Z] Application started" }
+        ]);
+        let mut p_custom_fields = HashMap::new();
+
+        let fields = processor
+            .extract_from_inline_log(
+                &mut json_value,
+                &mut p_custom_fields,
+                "custom_app_log",
+                Some("raw_log"),
+            )
+            .unwrap();
+
+        assert!(fields.contains("message"));
+        assert_eq!(json_value[0], json!(42));
+        assert_eq!(json_value[3]["level"], "INFO");
+        assert!(p_custom_fields.is_empty());
+    }
+
+    #[test]
+    fn test_extract_from_inline_log_empty_array() {
+        let processor = EventProcessor::new(TEST_CONFIG);
+        let mut json_value = json!([]);
+        let mut p_custom_fields = HashMap::new();
+
+        let fields = processor
+            .extract_from_inline_log(
+                &mut json_value,
+                &mut p_custom_fields,
+                "custom_app_log",
+                Some("raw_log"),
+            )
+            .unwrap();
+
+        assert!(fields.is_empty());
+        assert!(p_custom_fields.is_empty());
+    }
+
     #[test]
     fn test_unknown_log_format() {
         let processor = EventProcessor::new(TEST_CONFIG);

src/handlers/http/alert_target_policy.rs

@@ -13,7 +13,7 @@ use crate::utils::get_tenant_id_from_request;
 use actix_web::HttpRequest;
 
 pub async fn get(req: HttpRequest) -> Result<impl Responder, AlertError> {
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     let policy = active_policy(&tenant_id).await?;
     Ok(web::Json(policy))
 }
@@ -23,7 +23,7 @@ pub async fn put(
     Json(policy): Json<AlertTargetPolicyConfig>,
 ) -> Result<impl Responder, AlertError> {
     // validate before replacing so a bad policy never becomes active
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     validate_policy(&policy)?;
     replace_policy(&tenant_id, policy.clone()).await?;
     Ok(web::Json(policy))

src/handlers/http/alerts.rs

@@ -260,7 +260,7 @@ pub async fn post(
     req: HttpRequest,
     Json(alert): Json<AlertRequest>,
 ) -> Result<impl Responder, AlertError> {
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     let session_key = extract_session_key_from_req(&req)?;
     Ok(web::Json(
         post_internal(alert, &session_key, &tenant_id).await?,
@@ -356,7 +356,7 @@ pub async fn post_internal(
 pub async fn get(req: HttpRequest, alert_id: Path<Ulid>) -> Result<impl Responder, AlertError> {
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     Ok(web::Json(
         get_internal(&session_key, alert_id, &tenant_id).await?,
     ))
@@ -388,7 +388,7 @@ pub async fn delete(req: HttpRequest, alert_id: Path<Ulid>) -> Result<impl Respo
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
     let _lifecycle = crate::alerts::promql_alerts::lock_lifecycle(alert_id).await;
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     let guard = ALERTS.write().await;
     let alerts = if let Some(alerts) = guard.as_ref() {
         alerts
@@ -435,7 +435,7 @@ pub async fn update_notification_state(
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
     let _lifecycle = crate::alerts::promql_alerts::lock_lifecycle(alert_id).await;
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     let new_notification_state = match new_notification_state.state.as_str() {
         "notify" => NotificationState::Notify,
         "indefinite" => NotificationState::Mute("indefinite".into()),
@@ -492,7 +492,7 @@ pub async fn disable_alert(
 ) -> Result<impl Responder, AlertError> {
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     Ok(web::Json(
         disable_alert_internal(&session_key, alert_id, &tenant_id).await?,
     ))
@@ -534,7 +534,7 @@ pub async fn enable_alert(
 ) -> Result<impl Responder, AlertError> {
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     Ok(web::Json(
         enable_alert_internal(&session_key, alert_id, &tenant_id).await?,
     ))
@@ -591,7 +591,7 @@ pub async fn modify_alert(
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
     let _lifecycle = crate::alerts::promql_alerts::lock_lifecycle(alert_id).await;
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     // Get alerts manager reference without holding the global lock
     let alerts = {
         let guard = ALERTS.read().await;
@@ -688,7 +688,7 @@ pub async fn evaluate_alert(
 ) -> Result<impl Responder, AlertError> {
     let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     Ok(Json(
         evaluate_alert_internal(&session_key, alert_id, &tenant_id).await?,
     ))
@@ -724,7 +724,7 @@ pub async fn evaluate_alert_internal(
 }
 
 pub async fn list_tags(req: HttpRequest) -> Result<impl Responder, AlertError> {
-    let tenant_id = get_tenant_id_from_request(&req);
+    let tenant_id = get_tenant_id_from_request(&req)?;
     Ok(web::Json(list_tags_internal(&tenant_id).await?))
 }
 

src/handlers/http/apikeys.rs

@@ -331,7 +331,7 @@ pub async fn create_api_key(
     // can be slow and must not block other create/delete requests.
     drop(guard);
 
-    // Best-effort sync to other live nodes (reuses the standard user sync).
+    // Node delivery is best effort; invalid forwarding headers still fail the request.
     let caller_userid = user
         .as_api_key()
         .map(|k| k.created_by.clone())
@@ -345,6 +345,9 @@ pub async fn create_api_key(
     )
     .await
     {
+        if matches!(&e, RBACError::HeaderBuild(_)) {
+            return Err(e.into());
+        }
         tracing::error!("Failed to sync API-key user creation: {e}");
     }
 
@@ -393,6 +396,9 @@ pub async fn delete_api_key(
     )
     .await
     {
+        if matches!(&e, RBACError::HeaderBuild(_)) {
+            return Err(e.into());
+        }
         tracing::error!("Failed to sync API-key user deletion: {e}");
         return Err(ApiKeyError::RevocationSyncFailed(e.to_string()));
     }