From 0863a0d69cd5eca059bdba2f793fe8d1ad536dc4 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Thu, 22 Oct 2026 12:43:10 +0000 Subject: [PATCH] fix(alerts): validate and quote group-by fields of custom aggregation alerts (#15649) --- src/core/src/alerts/alert.rs | 94 +++++++++++ src/core/src/alerts/mod.rs | 305 ++++++++++++++++++++++++----------- src/core/src/http.rs | 3 +- 3 files changed, 307 insertions(+), 95 deletions(-) diff --git a/src/core/src/alerts/alert.rs b/src/core/src/alerts/alert.rs index a50a226f3e..8216be5bd6 100644 --- a/src/core/src/alerts/alert.rs +++ b/src/core/src/alerts/alert.rs @@ -358,6 +358,8 @@ pub enum AlertError { RecoveryNeedsRoutingKey { destination: String }, #[error("Error in multi alert grouping: {0}")] MultiAlertGroupingError(String), + #[error("Group by column {column} not found on stream {stream_name}")] + GroupByColumnNotFound { column: String, stream_name: String }, } /// What a manual trigger did: sent, with its messages, or held by an active downtime. @@ -477,6 +479,30 @@ fn validate_multi_alert_config(alert: &Alert) -> Result<(), AlertError> { Ok(()) } +/// Refuses, at save time, a custom-aggregation `group_by` entry that is not a column of the stream. +#[allow(clippy::result_large_err)] +fn validate_custom_group_by( + stream_name: &str, + schema: &arrow_schema::Schema, + alert: &Alert, +) -> Result<(), AlertError> { + if alert.query_condition.query_type != QueryType::Custom { + return Ok(()); + } + let group_by = alert + .query_condition + .aggregation + .as_ref() + .and_then(|agg| agg.group_by.as_deref()) + .unwrap_or(&[]); + super::quoted_group_by(schema, group_by) + .map(drop) + .map_err(|column| AlertError::GroupByColumnNotFound { + column, + stream_name: stream_name.to_owned(), + }) +} + /// Structural half only: `{var}` tokens are unresolved until send time, so a body shaped right but /// malformed still fails at send. A CONTENT template always renders an object. fn platform_body_is_addressable(kind: TemplateKind, body: &str) -> bool { @@ -842,6 +868,7 @@ async fn prepare_alert( stream_name: stream_name.to_owned(), }); } + validate_custom_group_by(stream_name, &schema, alert)?; // Alerts must follow the max_query_range of the stream as set in the schema if let Some(settings) = unwrap_stream_settings(&schema) { @@ -4851,6 +4878,73 @@ mod threshold_validation_tests { } } +#[cfg(test)] +mod group_by_validation_tests { + use arrow_schema::{DataType, Field, Schema}; + use config::{ + meta::alerts::{AggFunction, Aggregation, Condition, Operator, QueryType, alert::Alert}, + utils::json::json, + }; + + use super::{AlertError, validate_custom_group_by}; + + fn schema() -> Schema { + Schema::new(vec![ + Field::new("host", DataType::Utf8, false), + Field::new("latency", DataType::Int64, false), + ]) + } + + fn custom_alert(group_by: &[&str]) -> Alert { + let mut alert = Alert::default(); + alert.query_condition.query_type = QueryType::Custom; + alert.query_condition.aggregation = Some(Aggregation { + group_by: Some(group_by.iter().map(|s| s.to_string()).collect()), + function: AggFunction::Avg, + having: Condition { + column: "latency".into(), + operator: Operator::GreaterThan, + value: json!(100), + ignore_case: false, + }, + warning_value: None, + multi_alert: false, + }); + alert + } + + #[test] + fn test_group_by_columns_in_the_schema_pass() { + assert!(validate_custom_group_by("svc", &schema(), &custom_alert(&["host"])).is_ok()); + assert!(validate_custom_group_by("svc", &schema(), &custom_alert(&[])).is_ok()); + assert!(validate_custom_group_by("svc", &schema(), &Alert::default()).is_ok()); + } + + #[test] + fn test_group_by_column_outside_the_schema_is_rejected_and_named() { + let alert = custom_alert(&["host", "host, (SELECT 1)"]); + let err = validate_custom_group_by("svc", &schema(), &alert).unwrap_err(); + assert!(matches!( + &err, + AlertError::GroupByColumnNotFound { column, stream_name } + if column == "host, (SELECT 1)" && stream_name == "svc" + )); + assert_eq!( + err.to_string(), + "Group by column host, (SELECT 1) not found on stream svc" + ); + let response = axum::response::Response::from(err); + assert_eq!(response.status(), axum::http::StatusCode::BAD_REQUEST); + } + + #[test] + fn test_sql_alerts_derive_group_by_elsewhere_and_are_not_checked_here() { + let mut alert = custom_alert(&["nonexistent"]); + alert.query_condition.query_type = QueryType::SQL; + assert!(validate_custom_group_by("svc", &schema(), &alert).is_ok()); + } +} + #[cfg(test)] mod send_path_tests { use config::meta::destinations::{Endpoint, HTTPType, Template, TemplateKind}; diff --git a/src/core/src/alerts/mod.rs b/src/core/src/alerts/mod.rs index 1b6d9ad053..edddeed91f 100644 --- a/src/core/src/alerts/mod.rs +++ b/src/core/src/alerts/mod.rs @@ -32,6 +32,7 @@ use config::{ utils::{ base64, json::{Map, Value}, + sql::quote_identifier, }, }; use search::utils::is_permissable_function_error; @@ -1561,13 +1562,19 @@ pub async fn build_sql( ))? ) }; - if query_condition.aggregation.is_none() { + let Some(agg) = query_condition.aggregation.as_ref() else { return Ok(format!("SELECT * FROM \"{stream_name}\" {where_sql}")); }; + build_aggregation_sql(stream_name, &schema, agg, &where_sql) +} - // handle aggregation - let mut sql = String::new(); - let agg = query_condition.aggregation.as_ref().unwrap(); +/// Only columns of `schema` reach the SQL; any other `group_by`/`having` entry is refused. +fn build_aggregation_sql( + stream_name: &str, + schema: &Schema, + agg: &config::meta::alerts::Aggregation, + where_sql: &str, +) -> Result { let having_expr = { let data_type = match schema.field_with_name(&agg.having.column) { Ok(field) => field.data_type(), @@ -1581,69 +1588,56 @@ pub async fn build_sql( build_having_expr(agg, data_type)? }; + let col = quote_identifier(&agg.having.column); let func_expr = match agg.function { - AggFunction::Avg => format!("AVG(\"{}\")", agg.having.column), - AggFunction::Max => format!("MAX(\"{}\")", agg.having.column), - AggFunction::Min => format!("MIN(\"{}\")", agg.having.column), - AggFunction::Sum => format!("SUM(\"{}\")", agg.having.column), - AggFunction::Count => format!("COUNT(\"{}\")", agg.having.column), - AggFunction::Median => format!("MEDIAN(\"{}\")", agg.having.column), - AggFunction::P50 => format!( - "approx_percentile_cont(0.5) WITHIN GROUP (ORDER BY \"{}\")", - agg.having.column - ), - AggFunction::P75 => format!( - "approx_percentile_cont(0.75) WITHIN GROUP (ORDER BY \"{}\")", - agg.having.column - ), - AggFunction::P90 => format!( - "approx_percentile_cont(0.9) WITHIN GROUP (ORDER BY \"{}\")", - agg.having.column - ), - AggFunction::P95 => format!( - "approx_percentile_cont(0.95) WITHIN GROUP (ORDER BY \"{}\")", - agg.having.column - ), - AggFunction::P99 => format!( - "approx_percentile_cont(0.99) WITHIN GROUP (ORDER BY \"{}\")", - agg.having.column - ), + AggFunction::Avg => format!("AVG({col})"), + AggFunction::Max => format!("MAX({col})"), + AggFunction::Min => format!("MIN({col})"), + AggFunction::Sum => format!("SUM({col})"), + AggFunction::Count => format!("COUNT({col})"), + AggFunction::Median => format!("MEDIAN({col})"), + AggFunction::P50 => format!("approx_percentile_cont(0.5) WITHIN GROUP (ORDER BY {col})"), + AggFunction::P75 => format!("approx_percentile_cont(0.75) WITHIN GROUP (ORDER BY {col})"), + AggFunction::P90 => format!("approx_percentile_cont(0.9) WITHIN GROUP (ORDER BY {col})"), + AggFunction::P95 => format!("approx_percentile_cont(0.95) WITHIN GROUP (ORDER BY {col})"), + AggFunction::P99 => format!("approx_percentile_cont(0.99) WITHIN GROUP (ORDER BY {col})"), }; - if let Some(group) = agg.group_by.as_ref() - && !group.is_empty() - { - let cols = group.join(", "); - if agg.multi_alert { - // Multi-alerts drop the HAVING filter. A group that falls back - // under the threshold must still be RETURNED: otherwise its - // recovery is indistinguishable from it vanishing, and it would - // only resolve later via the reaper timeout, with a NULL value - // where the real reading should be. - // - // The page stays bounded, so it is ordered worst-first. That - // ordering keeps the rollup level exact and lets the cap admit the - // true top of the distribution rather than an arbitrary slice. The - // group columns are the deterministic tiebreak within a severity - // band. - let severity_order = - config::meta::alerts::aggregation_level::severity_order_sql(agg, "alert_agg_value") - .map_err(|e| anyhow::anyhow!("Invalid aggregation threshold: {e}"))?; - sql = format!( - "SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} ORDER BY {severity_order}, {cols}" - ); - } else { - sql = format!( - "SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} HAVING {having_expr}" - ); - } - } - if sql.is_empty() { - sql = format!( + let group = agg.group_by.as_deref().unwrap_or(&[]); + if group.is_empty() { + return Ok(format!( "SELECT {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} HAVING {having_expr}" - ); + )); } - Ok(sql) + let cols = quoted_group_by(schema, group) + .map_err(|column| { + anyhow::anyhow!("Group by column {column} not found on stream {stream_name}") + })? + .join(", "); + if agg.multi_alert { + // No HAVING: a group back under the threshold must still be returned to show recovery. + // Worst-first order keeps the bounded page exact; group columns are the tiebreak. + let severity_order = + config::meta::alerts::aggregation_level::severity_order_sql(agg, "alert_agg_value") + .map_err(|e| anyhow::anyhow!("Invalid aggregation threshold: {e}"))?; + return Ok(format!( + "SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} ORDER BY {severity_order}, {cols}" + )); + } + Ok(format!( + "SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} HAVING {having_expr}" + )) +} + +/// Quoted, escaped identifiers; `Err` carries the first entry that is not a column of `schema`. +fn quoted_group_by(schema: &Schema, group_by: &[String]) -> Result, String> { + group_by + .iter() + .map(|col| match schema.field_with_name(col) { + Ok(_) => Ok(quote_identifier(col)), + Err(_) => Err(col.clone()), + }) + .collect() } /// `HAVING` clause filtering the `alert_agg_value` aggregate. @@ -1707,11 +1701,11 @@ fn build_expr( field_alias: &str, field_type: &DataType, ) -> Result { - let field_alias = if !field_alias.is_empty() { - field_alias - } else { + let field_alias = quote_identifier(if field_alias.is_empty() { cond.column.as_str() - }; + } else { + field_alias + }); // Null/empty checks take no value. On non-string columns "empty" can only // mean null, so the empty checks degrade to the null checks there. let is_string_type = matches!( @@ -1719,20 +1713,20 @@ fn build_expr( DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View ); match cond.operator { - Operator::IsNull => return Ok(format!("\"{field_alias}\" IS NULL")), - Operator::IsNotNull => return Ok(format!("\"{field_alias}\" IS NOT NULL")), + Operator::IsNull => return Ok(format!("{field_alias} IS NULL")), + Operator::IsNotNull => return Ok(format!("{field_alias} IS NOT NULL")), Operator::IsEmpty => { return Ok(if is_string_type { - format!("(\"{field_alias}\" IS NULL OR \"{field_alias}\" = '')") + format!("({field_alias} IS NULL OR {field_alias} = '')") } else { - format!("\"{field_alias}\" IS NULL") + format!("{field_alias} IS NULL") }); } Operator::IsNotEmpty => { return Ok(if is_string_type { - format!("(\"{field_alias}\" IS NOT NULL AND \"{field_alias}\" != '')") + format!("({field_alias} IS NOT NULL AND {field_alias} != '')") } else { - format!("\"{field_alias}\" IS NOT NULL") + format!("{field_alias} IS NOT NULL") }); } _ => {} @@ -1748,17 +1742,17 @@ fn build_expr( // terminate the SQL string literal early. let val = val.replace('\'', "''"); match cond.operator { - Operator::EqualTo => format!("\"{field_alias}\" = '{val}'"), - Operator::NotEqualTo => format!("\"{field_alias}\" != '{val}'"), - Operator::GreaterThan => format!("\"{field_alias}\" > '{val}'"), + Operator::EqualTo => format!("{field_alias} = '{val}'"), + Operator::NotEqualTo => format!("{field_alias} != '{val}'"), + Operator::GreaterThan => format!("{field_alias} > '{val}'"), Operator::GreaterThanEquals => { - format!("\"{field_alias}\" >= '{val}'") + format!("{field_alias} >= '{val}'") } - Operator::LessThan => format!("\"{field_alias}\" < '{val}'"), - Operator::LessThanEquals => format!("\"{field_alias}\" <= '{val}'"), - Operator::Contains => format!("str_match(\"{field_alias}\", '{val}')"), + Operator::LessThan => format!("{field_alias} < '{val}'"), + Operator::LessThanEquals => format!("{field_alias} <= '{val}'"), + Operator::Contains => format!("str_match({field_alias}, '{val}')"), Operator::NotContains => { - format!("\"{field_alias}\" NOT LIKE '%{val}%'") + format!("{field_alias} NOT LIKE '%{val}%'") } // Null checks returned above; keep the dead arm panic-free. _ => { @@ -1807,15 +1801,15 @@ fn build_expr( .to_string() }; match cond.operator { - Operator::EqualTo => format!("\"{field_alias}\" = {val}"), - Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"), - Operator::GreaterThan => format!("\"{field_alias}\" > {val}"), + Operator::EqualTo => format!("{field_alias} = {val}"), + Operator::NotEqualTo => format!("{field_alias} != {val}"), + Operator::GreaterThan => format!("{field_alias} > {val}"), Operator::GreaterThanEquals => { - format!("\"{field_alias}\" >= {val}") + format!("{field_alias} >= {val}") } - Operator::LessThan => format!("\"{field_alias}\" < {val}"), + Operator::LessThan => format!("{field_alias} < {val}"), Operator::LessThanEquals => { - format!("\"{field_alias}\" <= {val}") + format!("{field_alias} <= {val}") } _ => { return Err(anyhow::anyhow!( @@ -1843,15 +1837,15 @@ fn build_expr( })? }; match cond.operator { - Operator::EqualTo => format!("\"{field_alias}\" = {val}"), - Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"), - Operator::GreaterThan => format!("\"{field_alias}\" > {val}"), + Operator::EqualTo => format!("{field_alias} = {val}"), + Operator::NotEqualTo => format!("{field_alias} != {val}"), + Operator::GreaterThan => format!("{field_alias} > {val}"), Operator::GreaterThanEquals => { - format!("\"{field_alias}\" >= {val}") + format!("{field_alias} >= {val}") } - Operator::LessThan => format!("\"{field_alias}\" < {val}"), + Operator::LessThan => format!("{field_alias} < {val}"), Operator::LessThanEquals => { - format!("\"{field_alias}\" <= {val}") + format!("{field_alias} <= {val}") } _ => { return Err(anyhow::anyhow!( @@ -1879,8 +1873,8 @@ fn build_expr( })? }; match cond.operator { - Operator::EqualTo => format!("\"{field_alias}\" = {val}"), - Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"), + Operator::EqualTo => format!("{field_alias} = {val}"), + Operator::NotEqualTo => format!("{field_alias} != {val}"), _ => { return Err(anyhow::anyhow!( "Column {} has data_type [{field_type}] and it does not supported operator [{:?}]", @@ -2973,6 +2967,129 @@ mod tests { assert_eq!(expr, "\"code\" < 1.5"); } + // ── Aggregation SQL: group_by identifiers ─────────────────────────────── + + fn agg_schema() -> Schema { + Schema::new(vec![ + Field::new("host", DataType::Utf8, false), + Field::new("region", DataType::Utf8, false), + Field::new("latency", DataType::Int64, false), + ]) + } + + fn group_agg(group_by: &[&str], multi_alert: bool) -> Aggregation { + Aggregation { + group_by: Some(group_by.iter().map(|s| s.to_string()).collect()), + function: AggFunction::Avg, + having: make_cond( + "latency", + Operator::GreaterThan, + Value::Number(serde_json::Number::from(100)), + ), + warning_value: None, + multi_alert, + } + } + + #[test] + fn test_build_aggregation_sql_quotes_group_by_columns() { + let agg = group_agg(&["host", "region"], false); + let sql = build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap(); + assert_eq!( + sql, + "SELECT \"host\", \"region\", AVG(\"latency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" GROUP BY \"host\", \"region\" HAVING \"alert_agg_value\" > 100" + ); + } + + #[test] + fn test_build_aggregation_sql_multi_alert_quotes_group_by_in_order_by() { + let agg = group_agg(&["host"], true); + let sql = build_aggregation_sql("svc", &agg_schema(), &agg, " WHERE 1 = 1").unwrap(); + assert_eq!( + sql, + "SELECT \"host\", AVG(\"latency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" WHERE 1 = 1 GROUP BY \"host\" ORDER BY CASE WHEN \"alert_agg_value\" > 100 THEN 2 ELSE 0 END DESC, \"host\"" + ); + } + + #[test] + fn test_build_aggregation_sql_rejects_group_by_entry_not_in_schema() { + for entry in [ + "nonexistent", + "host, (SELECT secret FROM \"other\" LIMIT 1)", + "host\" FROM \"other\" --", + "", + ] { + let agg = group_agg(&["host", entry], false); + let err = build_aggregation_sql("svc", &agg_schema(), &agg, "") + .unwrap_err() + .to_string(); + assert_eq!( + err, + format!("Group by column {entry} not found on stream svc"), + "entry {entry:?} must be rejected" + ); + } + } + + #[test] + fn test_build_aggregation_sql_escapes_quotes_inside_schema_backed_identifiers() { + let tricky = "host\" FROM \"other\" --"; + let schema = Schema::new(vec![ + Field::new(tricky, DataType::Utf8, false), + Field::new("lat\"ency", DataType::Int64, false), + ]); + let mut agg = group_agg(&[tricky], false); + agg.having.column = "lat\"ency".to_string(); + let sql = build_aggregation_sql("svc", &schema, &agg, "").unwrap(); + assert_eq!( + sql, + "SELECT \"host\"\" FROM \"\"other\"\" --\", AVG(\"lat\"\"ency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" GROUP BY \"host\"\" FROM \"\"other\"\" --\" HAVING \"alert_agg_value\" > 100" + ); + } + + #[test] + fn test_build_expr_escapes_quotes_inside_column_name() { + let cond = make_cond( + "lev\"el", + Operator::EqualTo, + Value::String("error".to_string()), + ); + let expr = build_expr(&cond, "", &DataType::Utf8).unwrap(); + assert_eq!(expr, "\"lev\"\"el\" = 'error'"); + let expr = build_expr(&cond, "al\"ias", &DataType::Utf8).unwrap(); + assert_eq!(expr, "\"al\"\"ias\" = 'error'"); + } + + #[test] + fn test_build_aggregation_sql_empty_group_by_collapses_to_single_row() { + let mut agg = group_agg(&[], false); + let sql = build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap(); + assert!(!sql.contains("GROUP BY"), "{sql}"); + agg.group_by = None; + assert_eq!( + build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap(), + sql + ); + } + + #[test] + fn test_quoted_group_by_reports_first_unknown_column() { + let schema = agg_schema(); + assert_eq!( + quoted_group_by(&schema, &["host".into(), "region".into()]), + Ok(vec!["\"host\"".to_string(), "\"region\"".to_string()]) + ); + assert_eq!( + quoted_group_by(&schema, &["host".into(), "a".into(), "b".into()]), + Err("a".to_string()) + ); + let quoted = Schema::new(vec![Field::new("a\"b", DataType::Utf8, false)]); + assert_eq!( + quoted_group_by("ed, &["a\"b".into()]), + Ok(vec!["\"a\"\"b\"".to_string()]) + ); + } + // ── HAVING clause type resolution ─────────────────────────────────────── fn make_agg(function: AggFunction, operator: Operator, value: Value) -> Aggregation { diff --git a/src/core/src/http.rs b/src/core/src/http.rs index 979bb61865..614ed5a314 100644 --- a/src/core/src/http.rs +++ b/src/core/src/http.rs @@ -105,7 +105,8 @@ impl From for Response { | AlertError::RecoveryWithNotificationGrouping | AlertError::RecoveryNeedsJsonPayload { .. } | AlertError::RecoveryNeedsRoutingKey { .. } - | AlertError::MultiAlertGroupingError(_) => MetaHttpResponse::bad_request(value), + | AlertError::MultiAlertGroupingError(_) + | AlertError::GroupByColumnNotFound { .. } => MetaHttpResponse::bad_request(value), // S-16 PR 4. A conflict, not a bad request: the alert being sent is // fine on its own terms — it is the SLOs that already exist and // measure from it that the request collides with. Both messages