diff --git a/src/core/src/traces/service_graph/api.rs b/src/core/src/traces/service_graph/api.rs index c1b696201c..ac48029b9a 100644 --- a/src/core/src/traces/service_graph/api.rs +++ b/src/core/src/traces/service_graph/api.rs @@ -20,6 +20,8 @@ use axum::response::Response as HttpResponse; use common::meta::http::HttpResponse as MetaHttpResponse; +#[cfg(feature = "enterprise")] +use config::utils::sql::{quote_identifier, quote_sql_string}; use serde::Deserialize; /// Query parameters for service graph API @@ -140,10 +142,10 @@ async fn topology_v1( // byte-identical to the pre-B4 query (backward compatible). The // `_o2_service_graph` stream has NO `agent_version` column, so we never // reference version here. - let agent_pred: Option = query.agent_env.as_deref().map(|env| { - let escaped = env.replace('\'', "''"); - format!("agent_env = '{escaped}'") - }); + let agent_pred: Option = query + .agent_env + .as_deref() + .map(|env| format!("agent_env = {}", quote_sql_string(env))); // 1. Query current window let edges = match query_edges_from_stream_internal( @@ -311,22 +313,23 @@ fn build_edges_sql( let agent_clause = agent_pred .map(|p| format!("\n AND {p}")) .unwrap_or_default(); + let table = quote_identifier(stream_name); + let org = quote_sql_string(org_id); if let Some(stream) = stream_filter { + let stream = quote_sql_string(stream); format!( - "SELECT * FROM \"{}\" - WHERE _timestamp >= {} AND _timestamp < {} - AND org_id = '{}' - AND trace_stream_name = '{}'{} - LIMIT 10000", - stream_name, start_time, end_time, org_id, stream, agent_clause + "SELECT * FROM {table} + WHERE _timestamp >= {start_time} AND _timestamp < {end_time} + AND org_id = {org} + AND trace_stream_name = {stream}{agent_clause} + LIMIT 10000" ) } else { format!( - "SELECT * FROM \"{}\" - WHERE _timestamp >= {} AND _timestamp < {} - AND org_id = '{}'{} - LIMIT 10000", - stream_name, start_time, end_time, org_id, agent_clause + "SELECT * FROM {table} + WHERE _timestamp >= {start_time} AND _timestamp < {end_time} + AND org_id = {org}{agent_clause} + LIMIT 10000" ) } } @@ -489,24 +492,36 @@ pub async fn get_edge_history( }; let mut filters = format!( - "_timestamp >= {} AND _timestamp < {} AND org_id = '{}'", - start_time, end_time, org_id + "_timestamp >= {} AND _timestamp < {} AND org_id = {}", + start_time, + end_time, + quote_sql_string(&org_id) ); if let Some(ref client) = query.client_service { - filters.push_str(&format!(" AND client_service = '{}'", client)); + filters.push_str(&format!( + " AND client_service = {}", + quote_sql_string(client) + )); } if let Some(ref server) = query.server_service { - filters.push_str(&format!(" AND server_service = '{}'", server)); + filters.push_str(&format!( + " AND server_service = {}", + quote_sql_string(server) + )); } if let Some(ref stream) = query.stream_name { - filters.push_str(&format!(" AND trace_stream_name = '{}'", stream)); + filters.push_str(&format!( + " AND trace_stream_name = {}", + quote_sql_string(stream) + )); } let sql = format!( "SELECT _timestamp, p50_latency_ns, p95_latency_ns, p99_latency_ns, \ total_requests, failed_requests \ - FROM \"{}\" WHERE {} ORDER BY _timestamp ASC LIMIT 10000", - stream_name, filters + FROM {} WHERE {} ORDER BY _timestamp ASC LIMIT 10000", + quote_identifier(stream_name), + filters ); let req = config::meta::search::Request { @@ -768,6 +783,18 @@ mod tests { assert!(sql.contains("AND agent_env = 'prod'")); } + #[test] + fn test_build_edges_sql_escapes_org_and_stream_filter() { + let sql = build_edges_sql("_o2_service_graph", Some("a'b"), 1, 2, "o'rg", None); + assert!(sql.contains("org_id = 'o''rg'")); + assert!(sql.contains("trace_stream_name = 'a''b'")); + assert!(!sql.contains("'o'rg'")); + // org_id is interpolated in the unfiltered branch too; a partial fix would miss it. + let sql = build_edges_sql("_o2_service_graph", None, 1, 2, "o'rg", None); + assert!(sql.contains("org_id = 'o''rg'")); + assert_eq!(sql.matches('\'').count() % 2, 0); + } + #[test] fn test_aggregate_baselines_null_client_service() { let records = vec![make_record(None, "svc-b", 50, 100, 150, 5)];