From 85d261b5d0e4b68dbc79ca18b844fc53e2368ee9 Mon Sep 17 00:00:00 2001 From: Ashish Kolhe <35160958+oasisk@users.noreply.github.com> Date: Tue, 14 Apr 2026 12:52:11 +0530 Subject: [PATCH] fix: vulnerability scan fixes (#11249) | # | Severity | Category | File | | :--- | :--- | :--- | :--- | | 1 | **High** | SSRF bypass (IPv6) | `ssrf_guard.rs` | | 2 | **High** | SSRF bypass (retry) | `enrichment_table/mod.rs` | | 3 | **High** | SSRF defense-in-depth | `alert.rs` | | 4 | **Critical** | CORS + credentials | `router/mod.rs` | | 5 | **Critical** | SQL Injection | `postgres.rs`, `sqlite.rs` | | 6 | **High** | Privilege escalation (token) | `users.rs` | | 7 | **Medium** | Panic/DoS | `users.rs` | | 8 | **Low** | Integer overflow | `alerts/mod.rs` | | 9 | **Medium** | XSS | `alertSummaryGenerator.ts`, `colorizeQuery.ts` | | 10 | **Medium** | Credential exposure in logs | `validator.rs`, `organization.rs`, `users.rs` | --------- Co-authored-by: Claude Sonnet 4.6 (1M context) Co-authored-by: Shrinath Rao Co-authored-by: Prabhat Sharma --- .github/workflows/playwright.yml | 3 + src/common/utils/ssrf_guard.rs | 90 ++++- src/config/src/config.rs | 10 + src/handler/http/auth/validator.rs | 2 +- .../http/request/alerts/destinations.rs | 2 +- src/handler/http/request/alerts/mod.rs | 3 +- .../http/request/enrichment_table/mod.rs | 7 +- src/handler/http/router/mod.rs | 144 ++++++- src/infra/src/db/postgres.rs | 372 ++++++++++++++---- src/infra/src/db/sqlite.rs | 353 +++++++++++++---- src/service/alerts/alert.rs | 14 +- src/service/organization.rs | 4 +- src/service/users.rs | 41 +- tests/integration_test.rs | 55 ++- tests/ui-testing/pages/logsPages/logsPage.js | 9 +- .../Streams/multiselect-stream.spec.js | 10 +- web/src/components/O2AIChat.vue | 7 +- web/src/components/alerts/AlertSummary.vue | 3 +- .../components/alerts/IncidentRCAAnalysis.vue | 10 +- .../components/alerts/IncidentTimeline.vue | 22 +- .../dashboards/addPanel/StreamFieldSelect.vue | 2 +- .../settings/VariableCustomValueSelector.vue | 2 +- web/src/components/settings/License.vue | 14 +- web/src/plugins/logs/Index.vue | 4 +- web/src/plugins/logs/IndexList.vue | 2 +- web/src/utils/alerts/alertSummaryGenerator.ts | 14 +- web/src/utils/query/colorizeQuery.ts | 13 +- 27 files changed, 978 insertions(+), 234 deletions(-) diff --git a/.github/workflows/playwright.yml b/.github/workflows/playwright.yml index b1f586615f..7cbcb65da1 100644 --- a/.github/workflows/playwright.yml +++ b/.github/workflows/playwright.yml @@ -48,6 +48,9 @@ env: ZO_CREATE_ORG_THROUGH_INGESTION: true ZO_UTF8_VIEW_ENABLED: false ZO_ENABLE_CROSS_LINKING: true + # Allow alert destinations to target localhost so the test validation stream + # (http://localhost:5080/api/.../..._json) is not blocked by the SSRF guard. + ZO_SSRF_ALLOW_LOOPBACK: true UPLOAD_TO_TESTDINO: ${{ vars.UPLOAD_TO_TESTDINO || 'false' }} jobs: diff --git a/src/common/utils/ssrf_guard.rs b/src/common/utils/ssrf_guard.rs index 1339dfc2e7..647b7c7707 100644 --- a/src/common/utils/ssrf_guard.rs +++ b/src/common/utils/ssrf_guard.rs @@ -39,7 +39,21 @@ use std::net::IpAddr; pub struct SsrfGuard; impl SsrfGuard { - /// Check if a URL is safe to request (not an SSRF attempt) + /// Check if a URL is safe to request (not an SSRF attempt), reading the + /// `ZO_SSRF_ALLOW_LOOPBACK` config flag to decide whether loopback/localhost + /// destinations are permitted. + /// + /// Use this at all alert/destination call sites so that the allow-loopback + /// escape-hatch is honoured consistently. + pub fn validate_url_with_config(url: &str) -> Result<(), String> { + let allow_loopback = config::get_config().common.ssrf_allow_loopback; + Self::validate_url_inner(url, allow_loopback) + } + + /// Check if a URL is safe to request (not an SSRF attempt). + /// + /// Loopback/localhost is always blocked. For a config-aware version that + /// respects `ZO_SSRF_ALLOW_LOOPBACK`, use [`validate_url_with_config`]. /// /// # Arguments /// * `url` - The URL to validate @@ -48,6 +62,10 @@ impl SsrfGuard { /// * `Ok(())` if the URL is safe /// * `Err(String)` with error message if the URL is potentially dangerous pub fn validate_url(url: &str) -> Result<(), String> { + Self::validate_url_inner(url, false) + } + + fn validate_url_inner(url: &str, allow_loopback: bool) -> Result<(), String> { // Parse the URL let parsed = url::Url::parse(url).map_err(|e| format!("Invalid URL: {}", e))?; @@ -64,10 +82,18 @@ impl SsrfGuard { .host_str() .ok_or_else(|| "URL must have a valid host".to_string())?; - // Check if host is an IP address - if let Ok(ip_addr) = host.parse::() { + // Check if host is an IP address. + // The url crate (some versions) returns bracketed IPv6 like "[::1]" from host_str(). + // Strip brackets so the IP can be parsed by Rust's IpAddr. + let unbracketed_host = if host.starts_with('[') && host.ends_with(']') { + &host[1..host.len() - 1] + } else { + host + }; + + if let Ok(ip_addr) = unbracketed_host.parse::() { // Block private IP ranges to prevent SSRF - if Self::is_private_ip(&ip_addr) { + if Self::is_private_ip_inner(&ip_addr, allow_loopback) { return Err(format!( "Access to private IP address {} is not allowed for security reasons. \ This prevents Server-Side Request Forgery (SSRF) attacks.", @@ -78,11 +104,11 @@ impl SsrfGuard { // It's a hostname - check for localhost and internal domains let lower_host = host.to_lowercase(); - // Block localhost variations - if lower_host == "localhost" - || lower_host.starts_with("localhost.") - || lower_host == "127.0.0.1" - || lower_host == "[::1]" + // Block localhost variations (unless loopback is explicitly allowed) + if !allow_loopback + && (lower_host == "localhost" + || lower_host.starts_with("localhost.") + || lower_host == "127.0.0.1") { return Err("Access to localhost is not allowed for security reasons".to_string()); } @@ -117,12 +143,24 @@ impl SsrfGuard { Ok(()) } - /// Check if an IP address is in a private/reserved range + /// Check if an IP address is in a private/reserved range (loopback always blocked). + /// Used by unit tests; production call sites use `is_private_ip_inner`. + #[cfg(test)] fn is_private_ip(ip: &IpAddr) -> bool { + Self::is_private_ip_inner(ip, false) + } + + /// Core private-IP check. When `allow_loopback` is true, the 127.0.0.0/8 + /// and ::1 ranges are not treated as blocked so that single-node/test + /// deployments can use loopback destinations. + fn is_private_ip_inner(ip: &IpAddr, allow_loopback: bool) -> bool { match ip { IpAddr::V4(ipv4) => { let octets = ipv4.octets(); + // Loopback (127.0.0.0/8) — skip when explicitly allowed + (!allow_loopback && octets[0] == 127) || + // Private networks (RFC 1918) // 10.0.0.0/8 (octets[0] == 10) || @@ -131,9 +169,6 @@ impl SsrfGuard { // 192.168.0.0/16 (octets[0] == 192 && octets[1] == 168) || - // Loopback (127.0.0.0/8) - already covered but explicit is better - (octets[0] == 127) || - // Link-local (169.254.0.0/16) (octets[0] == 169 && octets[1] == 254) || @@ -155,10 +190,15 @@ impl SsrfGuard { (octets[0] >= 240 && !(octets[0] == 255 && octets[1] == 255 && octets[2] == 255 && octets[3] == 255)) } IpAddr::V6(ipv6) => { + // IPv4-mapped IPv6 (e.g. ::ffff:127.0.0.1) — re-check as IPv4 + if let Some(v4) = ipv6.to_ipv4_mapped() { + return Self::is_private_ip_inner(&IpAddr::V4(v4), allow_loopback); + } + let segments = ipv6.segments(); - // Loopback (::1) - (ipv6.is_loopback()) || + // Loopback (::1) — skip when explicitly allowed + (!allow_loopback && ipv6.is_loopback()) || // Link-local (fe80::/10) (segments[0] & 0xffc0) == 0xfe80 || @@ -242,8 +282,28 @@ mod tests { assert!(SsrfGuard::is_private_ip(&"169.254.1.1".parse().unwrap())); assert!(SsrfGuard::is_private_ip(&"::1".parse().unwrap())); + // IPv4-mapped IPv6 addresses must also be blocked (SSRF bypass vector) + assert!(SsrfGuard::is_private_ip( + &"::ffff:127.0.0.1".parse().unwrap() + )); + assert!(SsrfGuard::is_private_ip( + &"::ffff:10.0.0.1".parse().unwrap() + )); + assert!(SsrfGuard::is_private_ip( + &"::ffff:192.168.1.1".parse().unwrap() + )); + // Public IPs should not be private assert!(!SsrfGuard::is_private_ip(&"8.8.8.8".parse().unwrap())); assert!(!SsrfGuard::is_private_ip(&"1.1.1.1".parse().unwrap())); } + + #[test] + fn test_block_ipv4_mapped_ipv6() { + // IPv4-mapped IPv6 must be blocked to prevent SSRF bypass (NEW-001) + assert!(SsrfGuard::validate_url("http://[::ffff:127.0.0.1]/webhook").is_err()); + assert!(SsrfGuard::validate_url("http://[::ffff:10.0.0.1]/webhook").is_err()); + assert!(SsrfGuard::validate_url("http://[::ffff:192.168.1.1]/webhook").is_err()); + assert!(SsrfGuard::validate_url("http://[::ffff:169.254.169.254]/webhook").is_err()); + } } diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 8ce92c1863..71158b9262 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -870,6 +870,16 @@ pub struct Common { pub ingestion_url: String, #[env_config(name = "ZO_WEB_URL", default = "http://localhost:5080")] pub web_url: String, + /// Comma-separated list of extra origins allowed for CORS in addition to `web_url`. + /// Example: `http://localhost:8081,https://staging.example.com` + #[env_config(name = "ZO_CORS_ALLOWED_ORIGINS", default = "")] + pub cors_allowed_origins: String, + /// Allow alert destinations to target loopback/localhost addresses. + /// Disabled by default (SSRF protection). Enable only in trusted environments + /// such as CI/CD pipelines or self-hosted single-node setups where the + /// server legitimately needs to send notifications to itself. + #[env_config(name = "ZO_SSRF_ALLOW_LOOPBACK", default = false)] + pub ssrf_allow_loopback: bool, #[env_config(name = "ZO_BASE_URI", default = "")] // /abc pub base_uri: String, #[env_config(name = "ZO_DATA_DIR", default = "./data/openobserve/")] diff --git a/src/handler/http/auth/validator.rs b/src/handler/http/auth/validator.rs index ee2697356e..7c06bf9075 100644 --- a/src/handler/http/auth/validator.rs +++ b/src/handler/http/auth/validator.rs @@ -926,7 +926,7 @@ async fn oo_validator_internal( Some(value) => value, None => return Err(AuthError::Unauthorized("Unauthorized Access".to_string())), }; - log::info!("Auth ext token found: validating: {username} {password}"); + log::info!("Auth ext token found: validating: {username}"); validator(req_data, &username, &password, auth_info, path_prefix).await } } else { diff --git a/src/handler/http/request/alerts/destinations.rs b/src/handler/http/request/alerts/destinations.rs index 4d9dce97ba..1ac1f28ee5 100644 --- a/src/handler/http/request/alerts/destinations.rs +++ b/src/handler/http/request/alerts/destinations.rs @@ -169,7 +169,7 @@ async fn test_http_destination(test_req: &TestDestinationRequest) -> Response { } // SSRF protection: Validate URL before making request - if let Err(error_msg) = SsrfGuard::validate_url(url) { + if let Err(error_msg) = SsrfGuard::validate_url_with_config(url) { return MetaHttpResponse::json(TestDestinationResponse { success: false, status_code: None, diff --git a/src/handler/http/request/alerts/mod.rs b/src/handler/http/request/alerts/mod.rs index 91c25d1ee6..95f4780888 100644 --- a/src/handler/http/request/alerts/mod.rs +++ b/src/handler/http/request/alerts/mod.rs @@ -905,7 +905,8 @@ pub async fn list_alerts( // Apply pagination to the combined list (regular alerts + anomaly configs). #[cfg(feature = "enterprise")] let list = if let Some((page_size, page_idx)) = page_size_and_idx { - let start = (page_idx * page_size) as usize; + // Use saturating_mul to prevent u64 overflow before casting to usize. + let start = page_idx.saturating_mul(page_size) as usize; list.into_iter() .skip(start) .take(page_size as usize) diff --git a/src/handler/http/request/enrichment_table/mod.rs b/src/handler/http/request/enrichment_table/mod.rs index 9af4d59abc..2ed04aa42f 100644 --- a/src/handler/http/request/enrichment_table/mod.rs +++ b/src/handler/http/request/enrichment_table/mod.rs @@ -262,9 +262,10 @@ pub async fn save_enrichment_table_from_url( return MetaHttpResponse::bad_request("URL cannot be empty"); } - // URL validation: Skip if retry mode (we're just reprocessing existing URLs) - // Apply validation for normal updates and replace_failed mode - if !retry && let Err(err_msg) = validate_enrichment_url(&request_body.url) { + // Always validate the URL for SSRF protection, regardless of retry mode. + // The retry flag controls job scheduling, not URL sourcing — the body URL + // can differ from the original on a retry call, so we cannot skip validation. + if let Err(err_msg) = validate_enrichment_url(&request_body.url) { return MetaHttpResponse::bad_request(err_msg); } diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 715dbc10a0..11a52ceff9 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -93,11 +93,69 @@ pub fn cors_layer() -> CorsLayer { header::HeaderName::from_static("x-openobserve-sampled"), X_O2_ASSISTANT_SESSION_ID, ]) - .allow_origin(AllowOrigin::mirror_request()) + // Restrict CORS to the configured web_url origin, plus any extra origins in + // ZO_CORS_ALLOWED_ORIGINS (comma-separated). mirror_request() + allow_credentials(true) + // allows any origin to make credentialed requests — effectively disabling same-origin + // protection. Only reflect the Origin header back if it matches one of our allowed origins. + // We extract only the scheme+host+port from each URL (ignoring any path) because the Origin + // header per RFC 6454 never carries a path component. This prevents prefix-match bypasses + // like https://app.example.com.evil.com when the allowed URL is https://app.example.com. + .allow_origin(AllowOrigin::predicate(|origin, _parts| { + let cfg = config::get_config(); + let web_url = cfg.common.web_url.trim_end_matches('/'); + if is_origin_allowed(origin.as_bytes(), web_url) { + return true; + } + // Also check any extra origins in ZO_CORS_ALLOWED_ORIGINS. + let extra = cfg.common.cors_allowed_origins.trim(); + if !extra.is_empty() { + for url in extra.split(',') { + let url = url.trim().trim_end_matches('/'); + if is_origin_allowed(origin.as_bytes(), url) { + return true; + } + } + } + false + })) .allow_credentials(true) .max_age(Duration::from_secs(3600)) } +/// Returns `true` if `request_origin` should be permitted for CORS. +/// +/// Rules: +/// - If `web_url` is empty (dev / unconfigured), allow all origins and log a warning. +/// - Extract the `scheme://host[:port]` portion of `web_url`, ignoring any path component, because +/// the `Origin` header per RFC 6454 never carries a path. +/// - Perform a byte-exact comparison — no prefix-match that could be bypassed by `https://app.example.com.evil.com` +/// when `web_url=https://app.example.com`. +/// - If `web_url` is malformed (no `://`), deny the request. +pub(crate) fn is_origin_allowed(request_origin: &[u8], web_url: &str) -> bool { + if web_url.is_empty() { + // In dev/test when web_url is empty, fall back to permissive behaviour. + // Log a warning so misconfigured production deployments are visible. + log::warn!( + "ZO_WEB_URL is not configured — CORS is unrestricted. \ + Set ZO_WEB_URL to your frontend origin in production." + ); + return true; + } + // Extract the origin (scheme://host[:port]) from web_url so that a + // web_url with a path (e.g. https://app.example.com/o2) still matches + // the bare Origin header (https://app.example.com). + let allowed_origin = match web_url.find("://") { + Some(scheme_end) => { + let after_scheme = &web_url[scheme_end + 3..]; + let host_end = after_scheme.find('/').unwrap_or(after_scheme.len()); + &web_url[..scheme_end + 3 + host_end] + } + // Malformed web_url — deny rather than fall back to prefix match. + None => return false, + }; + request_origin == allowed_origin.as_bytes() +} + /// Authentication middleware for API routes pub async fn auth_middleware(request: Request, next: Next) -> Response { // Extract request data FIRST, before any async calls @@ -1095,4 +1153,88 @@ mod tests { .is_some_and(|keywords| !keywords.is_empty()) ); } + + // ── is_origin_allowed unit tests ────────────────────────────────────────── + + #[test] + fn test_cors_exact_match_allowed() { + assert!(is_origin_allowed( + b"https://app.example.com", + "https://app.example.com" + )); + } + + #[test] + fn test_cors_exact_match_with_port() { + assert!(is_origin_allowed( + b"https://app.example.com:8080", + "https://app.example.com:8080" + )); + } + + #[test] + fn test_cors_web_url_with_path_stripped() { + // web_url has a path component — origin must still match scheme+host only. + assert!(is_origin_allowed( + b"https://app.example.com", + "https://app.example.com/o2" + )); + } + + #[test] + fn test_cors_web_url_trailing_slash_stripped() { + // Caller trims trailing slashes on web_url before calling us. + assert!(is_origin_allowed( + b"https://app.example.com", + "https://app.example.com" + )); + } + + #[test] + fn test_cors_prefix_bypass_denied() { + // https://app.example.com.evil.com must NOT match https://app.example.com + assert!(!is_origin_allowed( + b"https://app.example.com.evil.com", + "https://app.example.com" + )); + } + + #[test] + fn test_cors_subdomain_denied() { + assert!(!is_origin_allowed( + b"https://sub.app.example.com", + "https://app.example.com" + )); + } + + #[test] + fn test_cors_different_scheme_denied() { + assert!(!is_origin_allowed( + b"http://app.example.com", + "https://app.example.com" + )); + } + + #[test] + fn test_cors_different_port_denied() { + assert!(!is_origin_allowed( + b"https://app.example.com:9000", + "https://app.example.com:8080" + )); + } + + #[test] + fn test_cors_malformed_web_url_denied() { + // web_url without "://" is malformed — should deny + assert!(!is_origin_allowed( + b"https://app.example.com", + "app.example.com" + )); + } + + #[test] + fn test_cors_empty_web_url_allows_all() { + // Empty web_url is permissive (dev mode) + assert!(is_origin_allowed(b"https://any.origin.com", "")); + } } diff --git a/src/infra/src/db/postgres.rs b/src/infra/src/db/postgres.rs index 2ecbe754de..d923418339 100644 --- a/src/infra/src/db/postgres.rs +++ b/src/infra/src/db/postgres.rs @@ -128,10 +128,14 @@ impl super::Db for PostgresDb { let (module, key1, key2) = super::parse_key(key); let pool = CLIENT_RO.clone(); DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc(); - let query = format!( - "SELECT value FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND key2 = '{key2}' ORDER BY start_dt DESC;" - ); - let value: String = match sqlx::query_scalar(&query).fetch_one(&pool).await { + let query = "SELECT value FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 ORDER BY start_dt DESC;"; + let value: String = match sqlx::query_scalar(query) + .bind(&module) + .bind(&key1) + .bind(&key2) + .fetch_one(&pool) + .await + { Ok(v) => v, Err(e) => { if let sqlx::Error::RowNotFound = e { @@ -236,9 +240,12 @@ impl super::Db for PostgresDb { } else { lock_id as i64 }; - let lock_sql = format!("SELECT pg_advisory_xact_lock({lock_id})"); DB_QUERY_NUMS.with_label_values(&["get_lock", ""]).inc(); - if let Err(e) = sqlx::query(&lock_sql).execute(&mut *tx).await { + if let Err(e) = sqlx::query("SELECT pg_advisory_xact_lock($1)") + .bind(lock_id) + .execute(&mut *tx) + .await + { if let Err(e) = tx.rollback().await { log::error!("[POSTGRES] rollback get_for_update error: {e}"); } @@ -415,56 +422,119 @@ impl super::Db for PostgresDb { } let (module, key1, key2) = super::parse_key(key); - // Escape ' (single quote) character with '' - let (key1, key2) = (key1.replace("'", "''"), key2.replace("'", "''")); - let sql = if with_prefix { - if key1.is_empty() { - format!(r#"DELETE FROM meta WHERE module = '{module}';"#) - } else if key2.is_empty() { - format!(r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}';"#) - } else { - format!( - r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND (key2 = '{key2}' OR key2 LIKE '{key2}/%');"# - ) - } - } else { - format!( - r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND key2 = '{key2}';"# - ) - }; - - let sql = if let Some(start_dt) = start_dt { - sql.replace(';', &format!(" AND start_dt = {start_dt};")) - } else { - sql - }; - let pool = CLIENT.clone(); DB_QUERY_NUMS.with_label_values(&["delete", "meta"]).inc(); - sqlx::query(&sql).execute(&pool).await?; + + if with_prefix { + if key1.is_empty() { + // module = $1 [, start_dt = $2] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND start_dt = $2") + .bind(&module) + .bind(dt) + .execute(&pool) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1") + .bind(&module) + .execute(&pool) + .await?; + } + } else if key2.is_empty() { + // module = $1, key1 = $2 [, start_dt = $3] + if let Some(dt) = start_dt { + sqlx::query( + "DELETE FROM meta WHERE module = $1 AND key1 = $2 AND start_dt = $3", + ) + .bind(&module) + .bind(&key1) + .bind(dt) + .execute(&pool) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2") + .bind(&module) + .bind(&key1) + .execute(&pool) + .await?; + } + } else { + // module = $1, key1 = $2, key2 = $3, key2/% = $4 [, start_dt = $5] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4) AND start_dt = $5") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(format!("{}/%", key2)) + .bind(dt) + .execute(&pool) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4)") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(format!("{}/%", key2)) + .execute(&pool) + .await?; + } + } + } else { + // module = $1, key1 = $2, key2 = $3 [, start_dt = $4] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 AND start_dt = $4") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(dt) + .execute(&pool) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3") + .bind(&module) + .bind(&key1) + .bind(&key2) + .execute(&pool) + .await?; + } + } Ok(()) } async fn list(&self, prefix: &str) -> Result> { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta".to_string(); + let mut sql = + "SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } - sql = format!("{sql} ORDER BY start_dt ASC"); + sql.push_str(" ORDER BY start_dt ASC"); let pool = CLIENT_RO.clone(); DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in params { + query = query.bind(p); + } + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| { @@ -478,22 +548,37 @@ impl super::Db for PostgresDb { async fn list_keys(&self, prefix: &str) -> Result> { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, '' AS value FROM meta ".to_string(); + let mut sql = + "SELECT id, module, key1, key2, start_dt, '' AS value FROM meta WHERE 1=1 ".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } - sql = format!("{sql} ORDER BY start_dt ASC"); + sql.push_str(" ORDER BY start_dt ASC"); + let pool = CLIENT_RO.clone(); DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in params { + query = query.bind(p); + } + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| format!("/{}/{}/{}", r.module, r.key1, r.key2)) @@ -521,25 +606,19 @@ impl super::Db for PostgresDb { } let (min_dt, max_dt) = start_dt.unwrap(); - let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta".to_string(); - if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); - } - if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); - } - if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); - } - sql = format!("{sql} AND start_dt >= {min_dt} AND start_dt <= {max_dt}"); - sql = format!("{sql} ORDER BY start_dt ASC"); + let (sql, str_params) = build_list_by_start_dt_sql(prefix, min_dt, max_dt); let pool = CLIENT_RO.clone(); DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in str_params { + query = query.bind(p); + } + // min_dt and max_dt are i64 — bind them directly so PostgreSQL receives + // them as BIGINT rather than TEXT, avoiding "operator does not exist: bigint >= text". + query = query.bind(min_dt); + query = query.bind(max_dt); + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| (r.start_dt, Bytes::from(r.value))) @@ -548,19 +627,34 @@ impl super::Db for PostgresDb { async fn count(&self, prefix: &str) -> Result { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT COUNT(*) AS num FROM meta".to_string(); + let mut sql = "SELECT COUNT(*) AS num FROM meta WHERE 1=1".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } let pool = CLIENT_RO.clone(); DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc(); - let count: i64 = sqlx::query_scalar(&sql).fetch_one(&pool).await?; + let mut query = sqlx::query_scalar::<_, i64>(&sql); + for p in params { + query = query.bind(p); + } + let count: i64 = query.fetch_one(&pool).await?; Ok(count) } @@ -778,6 +872,57 @@ pub async fn drop_column(table: &str, column: &str) -> Result<()> { Ok(()) } +/// Builds the SQL string and string-typed bind parameters for `list_values_by_start_dt`. +/// +/// Returns `(sql, str_params)` where `str_params` contains the string-typed positional +/// parameters (module / key1 / key2). The caller must then bind `min_dt` and `max_dt` +/// as `i64` **after** the string params so that PostgreSQL sees them as BIGINT, not TEXT. +/// +/// Keeping this logic in a standalone function makes the SQL generation unit-testable +/// without requiring a live database connection. +pub(crate) fn build_list_by_start_dt_sql( + prefix: &str, + min_dt: i64, + max_dt: i64, +) -> (String, Vec) { + let (module, key1, key2) = super::parse_key(prefix); + let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string(); + let mut params: Vec = Vec::new(); + + if !module.is_empty() { + sql.push_str(" AND module = $1"); + params.push(module); + } + if !key1.is_empty() { + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); + } + if !key2.is_empty() { + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{key2}/%"); + params.push(key2); + params.push(key2_prefix); + } + // min_dt / max_dt are intentionally NOT pushed into params here. + // The caller binds them as i64 after the string-params loop so that the + // PostgreSQL driver receives them as BIGINT, not TEXT. + let n = params.len(); + sql.push_str(&format!( + " AND start_dt >= ${} AND start_dt <= ${}", + n + 1, + n + 2 + )); + // Suppress unused-variable warning: min_dt / max_dt are consumed by the caller. + let _ = (min_dt, max_dt); + sql.push_str(" ORDER BY start_dt ASC"); + + (sql, params) +} + #[cfg(test)] mod tests { use super::*; @@ -1109,4 +1254,95 @@ mod tests { assert_eq!(k1, "key1"); assert!(k2.starts_with("key2")); } + + // ── build_list_by_start_dt_sql unit tests ───────────────────────────────── + // These tests verify that the SQL and parameter vector produced by the helper + // are correct for every combination of module / key1 / key2 presence, and that + // min_dt / max_dt placeholder indices are always one past the last string param. + + #[test] + fn test_build_list_by_start_dt_sql_full_prefix() { + let (sql, params) = build_list_by_start_dt_sql("/module/key1/key2", 100, 200); + // 3 string params: module=$1, key1=$2, key2=$3 / key2-prefix=$4 + assert_eq!(params.len(), 4); + assert_eq!(params[0], "module"); + assert_eq!(params[1], "key1"); + assert_eq!(params[2], "key2"); + assert_eq!(params[3], "key2/%"); + assert!( + sql.contains("AND module = $1"), + "module param missing: {sql}" + ); + assert!(sql.contains("AND key1 = $2"), "key1 param missing: {sql}"); + assert!( + sql.contains("AND (key2 = $3 OR key2 LIKE $4)"), + "key2 param missing: {sql}" + ); + // min_dt=$5, max_dt=$6 (one past the 4 string params) + assert!( + sql.contains("AND start_dt >= $5 AND start_dt <= $6"), + "start_dt bounds wrong: {sql}" + ); + assert!( + sql.ends_with("ORDER BY start_dt ASC"), + "missing ORDER BY: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_module_only() { + let (sql, params) = build_list_by_start_dt_sql("/module", 1, 999); + assert_eq!(params.len(), 1); + assert_eq!(params[0], "module"); + assert!(sql.contains("AND module = $1")); + // min_dt=$2, max_dt=$3 + assert!( + sql.contains("AND start_dt >= $2 AND start_dt <= $3"), + "start_dt bounds wrong: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_module_and_key1() { + let (sql, params) = build_list_by_start_dt_sql("/module/key1", 10, 20); + assert_eq!(params.len(), 2); + assert_eq!(params[0], "module"); + assert_eq!(params[1], "key1"); + assert!(sql.contains("AND module = $1")); + assert!(sql.contains("AND key1 = $2")); + // min_dt=$3, max_dt=$4 + assert!( + sql.contains("AND start_dt >= $3 AND start_dt <= $4"), + "start_dt bounds wrong: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_empty_prefix() { + let (sql, params) = build_list_by_start_dt_sql("", 0, 9999); + // No string params — min_dt=$1, max_dt=$2 + assert!(params.is_empty()); + assert!( + sql.contains("AND start_dt >= $1 AND start_dt <= $2"), + "start_dt bounds wrong for empty prefix: {sql}" + ); + assert!(sql.contains("WHERE 1=1")); + } + + #[test] + fn test_build_list_by_start_dt_sql_key2_prefix_pattern() { + let (sql, params) = build_list_by_start_dt_sql("/m/k1/k2", 0, 0); + // key2 LIKE pattern must end with /% + assert_eq!(params[3], "k2/%"); + assert!(sql.contains("OR key2 LIKE $4")); + } + + #[test] + fn test_build_list_by_start_dt_sql_starts_with_base_select() { + let (sql, _) = build_list_by_start_dt_sql("/m/k1/k2", 1, 2); + assert!( + sql.starts_with("SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1"), + "unexpected SQL start: {sql}" + ); + } } diff --git a/src/infra/src/db/sqlite.rs b/src/infra/src/db/sqlite.rs index e003957187..494a7f1bae 100644 --- a/src/infra/src/db/sqlite.rs +++ b/src/infra/src/db/sqlite.rs @@ -228,10 +228,14 @@ impl super::Db for SqliteDb { async fn get(&self, key: &str) -> Result { let (module, key1, key2) = super::parse_key(key); let pool = CLIENT_RO.clone(); - let query = format!( - "SELECT value FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND key2 = '{key2}' ORDER BY start_dt DESC;" - ); - let value: String = match sqlx::query_scalar(&query).fetch_one(&pool).await { + let query = "SELECT value FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 ORDER BY start_dt DESC;"; + let value: String = match sqlx::query_scalar(query) + .bind(&module) + .bind(&key1) + .bind(&key2) + .fetch_one(&pool) + .await + { Ok(v) => v, Err(e) => { if let sqlx::Error::RowNotFound = e { @@ -521,54 +525,117 @@ impl super::Db for SqliteDb { } let (module, key1, key2) = super::parse_key(key); - // Escape ' (single quote) character with '' - let (key1, key2) = (key1.replace("'", "''"), key2.replace("'", "''")); - let sql = if with_prefix { - if key1.is_empty() { - format!(r#"DELETE FROM meta WHERE module = '{module}';"#) - } else if key2.is_empty() { - format!(r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}';"#) - } else { - format!( - r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND (key2 = '{key2}' OR key2 LIKE '{key2}/%');"# - ) - } - } else { - format!( - r#"DELETE FROM meta WHERE module = '{module}' AND key1 = '{key1}' AND key2 = '{key2}';"# - ) - }; - - let sql = if let Some(start_dt) = start_dt { - sql.replace(';', &format!(" AND start_dt = {start_dt};")) - } else { - sql - }; - let client = CLIENT_RW.clone(); let client = client.lock().await; - sqlx::query(&sql).execute(&*client).await?; + + if with_prefix { + if key1.is_empty() { + // module = $1 [, start_dt = $2] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND start_dt = $2") + .bind(&module) + .bind(dt) + .execute(&*client) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1") + .bind(&module) + .execute(&*client) + .await?; + } + } else if key2.is_empty() { + // module = $1, key1 = $2 [, start_dt = $3] + if let Some(dt) = start_dt { + sqlx::query( + "DELETE FROM meta WHERE module = $1 AND key1 = $2 AND start_dt = $3", + ) + .bind(&module) + .bind(&key1) + .bind(dt) + .execute(&*client) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2") + .bind(&module) + .bind(&key1) + .execute(&*client) + .await?; + } + } else { + // module = $1, key1 = $2, key2 = $3, key2/% = $4 [, start_dt = $5] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4) AND start_dt = $5") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(format!("{}/%", key2)) + .bind(dt) + .execute(&*client) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4)") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(format!("{}/%", key2)) + .execute(&*client) + .await?; + } + } + } else { + // module = $1, key1 = $2, key2 = $3 [, start_dt = $4] + if let Some(dt) = start_dt { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 AND start_dt = $4") + .bind(&module) + .bind(&key1) + .bind(&key2) + .bind(dt) + .execute(&*client) + .await?; + } else { + sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3") + .bind(&module) + .bind(&key1) + .bind(&key2) + .execute(&*client) + .await?; + } + } Ok(()) } async fn list(&self, prefix: &str) -> Result> { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta".to_string(); + let mut sql = + "SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } - sql = format!("{sql} ORDER BY start_dt ASC"); + sql.push_str(" ORDER BY start_dt ASC"); let pool = CLIENT_RO.clone(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in params { + query = query.bind(p); + } + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| { @@ -582,22 +649,36 @@ impl super::Db for SqliteDb { async fn list_keys(&self, prefix: &str) -> Result> { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, '' AS value FROM meta".to_string(); + let mut sql = + "SELECT id, module, key1, key2, start_dt, '' AS value FROM meta WHERE 1=1".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } - sql = format!("{sql} ORDER BY start_dt ASC"); + sql.push_str(" ORDER BY start_dt ASC"); let pool = CLIENT_RO.clone(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in params { + query = query.bind(p); + } + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| format!("/{}/{}/{}", r.module, r.key1, r.key2)) @@ -625,24 +706,17 @@ impl super::Db for SqliteDb { } let (min_dt, max_dt) = start_dt.unwrap(); - let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta".to_string(); - if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); - } - if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); - } - if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); - } - sql = format!("{sql} AND start_dt >= {min_dt} AND start_dt <= {max_dt}"); - sql = format!("{sql} ORDER BY start_dt ASC"); + let (sql, str_params) = build_list_by_start_dt_sql(prefix, min_dt, max_dt); let pool = CLIENT_RO.clone(); - let ret = sqlx::query_as::<_, super::MetaRecord>(&sql) - .fetch_all(&pool) - .await?; + let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql); + for p in str_params { + query = query.bind(p); + } + // Bind min_dt / max_dt as i64 so SQLite receives them as INTEGER, not TEXT. + query = query.bind(min_dt); + query = query.bind(max_dt); + let ret = query.fetch_all(&pool).await?; Ok(ret .into_iter() .map(|r| (r.start_dt, Bytes::from(r.value))) @@ -651,19 +725,34 @@ impl super::Db for SqliteDb { async fn count(&self, prefix: &str) -> Result { let (module, key1, key2) = super::parse_key(prefix); - let mut sql = "SELECT COUNT(*) AS num FROM meta".to_string(); + let mut sql = "SELECT COUNT(*) AS num FROM meta WHERE 1=1".to_string(); + let mut params = Vec::new(); + if !module.is_empty() { - sql = format!("{sql} WHERE module = '{module}'"); + sql.push_str(" AND module = $1"); + params.push(module); } if !key1.is_empty() { - sql = format!("{sql} AND key1 = '{key1}'"); + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); } if !key2.is_empty() { - sql = format!("{sql} AND (key2 = '{key2}' OR key2 LIKE '{key2}/%')"); + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{}/%", key2); + params.push(key2); + params.push(key2_prefix); } let pool = CLIENT_RO.clone(); - let count: i64 = sqlx::query_scalar(&sql).fetch_one(&pool).await?; + let mut query = sqlx::query_scalar::<_, i64>(&sql); + for p in params { + query = query.bind(p); + } + let count: i64 = query.fetch_one(&pool).await?; Ok(count) } @@ -853,6 +942,56 @@ pub async fn drop_column(client: &Pool, table: &str, column: &str) -> Re Ok(()) } +/// Builds the SQL string and string-typed bind parameters for `list_values_by_start_dt`. +/// +/// Returns `(sql, str_params)` where `str_params` contains the string-typed positional +/// parameters (module / key1 / key2). The caller must then bind `min_dt` and `max_dt` +/// as `i64` **after** the string params so that the database receives them as INTEGER, +/// not TEXT. +/// +/// Keeping this logic in a standalone function makes the SQL generation unit-testable +/// without requiring a live database connection. +pub(crate) fn build_list_by_start_dt_sql( + prefix: &str, + min_dt: i64, + max_dt: i64, +) -> (String, Vec) { + let (module, key1, key2) = super::parse_key(prefix); + let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string(); + let mut params: Vec = Vec::new(); + + if !module.is_empty() { + sql.push_str(" AND module = $1"); + params.push(module); + } + if !key1.is_empty() { + sql.push_str(&format!(" AND key1 = ${}", params.len() + 1)); + params.push(key1); + } + if !key2.is_empty() { + sql.push_str(&format!( + " AND (key2 = ${} OR key2 LIKE ${})", + params.len() + 1, + params.len() + 2 + )); + let key2_prefix = format!("{key2}/%"); + params.push(key2); + params.push(key2_prefix); + } + // min_dt / max_dt are intentionally NOT pushed into params here. + // The caller binds them as i64 after the string-params loop. + let n = params.len(); + sql.push_str(&format!( + " AND start_dt >= ${} AND start_dt <= ${}", + n + 1, + n + 2 + )); + let _ = (min_dt, max_dt); + sql.push_str(" ORDER BY start_dt ASC"); + + (sql, params) +} + #[cfg(test)] mod tests { use super::*; @@ -1143,4 +1282,86 @@ mod tests { assert_eq!(k1, "key1"); assert!(k2.starts_with("key2")); } + + // ── build_list_by_start_dt_sql unit tests ───────────────────────────────── + + #[test] + fn test_build_list_by_start_dt_sql_full_prefix() { + let (sql, params) = build_list_by_start_dt_sql("/module/key1/key2", 100, 200); + assert_eq!(params.len(), 4); + assert_eq!(params[0], "module"); + assert_eq!(params[1], "key1"); + assert_eq!(params[2], "key2"); + assert_eq!(params[3], "key2/%"); + assert!( + sql.contains("AND module = $1"), + "module param missing: {sql}" + ); + assert!(sql.contains("AND key1 = $2"), "key1 param missing: {sql}"); + assert!( + sql.contains("AND (key2 = $3 OR key2 LIKE $4)"), + "key2 param missing: {sql}" + ); + assert!( + sql.contains("AND start_dt >= $5 AND start_dt <= $6"), + "start_dt bounds wrong: {sql}" + ); + assert!( + sql.ends_with("ORDER BY start_dt ASC"), + "missing ORDER BY: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_module_only() { + let (sql, params) = build_list_by_start_dt_sql("/module", 1, 999); + assert_eq!(params.len(), 1); + assert_eq!(params[0], "module"); + assert!(sql.contains("AND module = $1")); + assert!( + sql.contains("AND start_dt >= $2 AND start_dt <= $3"), + "start_dt bounds wrong: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_module_and_key1() { + let (sql, params) = build_list_by_start_dt_sql("/module/key1", 10, 20); + assert_eq!(params.len(), 2); + assert_eq!(params[0], "module"); + assert_eq!(params[1], "key1"); + assert!(sql.contains("AND module = $1")); + assert!(sql.contains("AND key1 = $2")); + assert!( + sql.contains("AND start_dt >= $3 AND start_dt <= $4"), + "start_dt bounds wrong: {sql}" + ); + } + + #[test] + fn test_build_list_by_start_dt_sql_empty_prefix() { + let (sql, params) = build_list_by_start_dt_sql("", 0, 9999); + assert!(params.is_empty()); + assert!( + sql.contains("AND start_dt >= $1 AND start_dt <= $2"), + "start_dt bounds wrong for empty prefix: {sql}" + ); + assert!(sql.contains("WHERE 1=1")); + } + + #[test] + fn test_build_list_by_start_dt_sql_key2_prefix_pattern() { + let (sql, params) = build_list_by_start_dt_sql("/m/k1/k2", 0, 0); + assert_eq!(params[3], "k2/%"); + assert!(sql.contains("OR key2 LIKE $4")); + } + + #[test] + fn test_build_list_by_start_dt_sql_starts_with_base_select() { + let (sql, _) = build_list_by_start_dt_sql("/m/k1/k2", 1, 2); + assert!( + sql.starts_with("SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1"), + "unexpected SQL start: {sql}" + ); + } } diff --git a/src/service/alerts/alert.rs b/src/service/alerts/alert.rs index 090f499eda..995e7536e7 100644 --- a/src/service/alerts/alert.rs +++ b/src/service/alerts/alert.rs @@ -70,7 +70,10 @@ use crate::{ common::{ infra::config::ORGANIZATIONS, meta::authz::Authz, - utils::auth::{is_ofga_unsupported, remove_ownership, set_ownership}, + utils::{ + auth::{is_ofga_unsupported, remove_ownership, set_ownership}, + ssrf_guard::SsrfGuard, + }, }, service::{ alerts::{QueryConditionExt, build_sql, destinations}, @@ -1253,6 +1256,15 @@ async fn send_http_notification(endpoint: &Endpoint, msg: String) -> Result Result<( let invite = org_invites::get_by_token_user(invite_token, user_email) .await .map_err(|e| { - log::info!("error getting token {invite_token} for email {user_email} : {e}"); + log::info!("error getting invite token for email {user_email} : {e}"); anyhow::anyhow!("Provided Token is not valid for this email id") })?; @@ -928,7 +928,7 @@ pub async fn decline_invitation( let invite = org_invites::get_by_token_user(token, user_email) .await .map_err(|e| { - log::info!("error getting token {token} for email {user_email} : {e}"); + log::info!("error getting invite token for email {user_email} : {e}"); anyhow::anyhow!("Provided Token is not valid for this email id") })?; diff --git a/src/service/users.rs b/src/service/users.rs index b671f65d4b..da4e81b6ab 100644 --- a/src/service/users.rs +++ b/src/service/users.rs @@ -315,10 +315,17 @@ pub async fn update_user( if is_root_user(initiator_id) { allow_password_update = true } else { - let initiating_user = db::user::get(Some(org_id), initiator_id) - .await - .unwrap() - .unwrap(); + let initiating_user = match db::user::get(Some(org_id), initiator_id).await + { + Ok(Some(u)) => u, + Ok(None) => { + return Ok(MetaHttpResponse::unauthorized("Not Allowed")); + } + Err(e) => { + log::error!("Error fetching initiating user {initiator_id}: {e}"); + return Ok(MetaHttpResponse::unauthorized("Not Allowed")); + } + }; if (local_user.role.eq(&UserRole::Root) && initiating_user.role.eq(&UserRole::Root)) || (!local_user.role.eq(&UserRole::Root) @@ -412,7 +419,10 @@ pub async fn update_user( is_org_updated = true; } } - if user.token.is_some() { + // Token replacement is a privileged operation — only allow if the + // initiator is updating their own token OR has password-update rights + // (i.e. Root or Admin updating a non-root user). + if user.token.is_some() && (update_mode.is_self_update() || allow_password_update) { new_user.token = user.token.unwrap(); is_org_updated = true; } @@ -616,9 +626,12 @@ pub async fn add_user_to_org( root_user.get("root").unwrap().clone() } else { match db::user::get(Some(org_id), initiator_id).await { - Ok(user) => user.unwrap(), + Ok(Some(user)) => user, + Ok(None) => { + return Ok(MetaHttpResponse::unauthorized("Not Allowed")); + } Err(e) => { - log::error!("Error fetching user: {e}"); + log::error!("Error fetching initiating user: {e}"); return Ok(MetaHttpResponse::not_found("User not found")); } } @@ -905,10 +918,16 @@ pub async fn remove_user_from_org( let initiating_user = if is_root_user(initiator_id) { ROOT_USER.get("root").unwrap().to_owned() } else { - db::user::get(Some(org_id), initiator_id) - .await - .unwrap() - .unwrap() + match db::user::get(Some(org_id), initiator_id).await { + Ok(Some(user)) => user, + Ok(None) => { + return Ok(MetaHttpResponse::unauthorized("Not Allowed")); + } + Err(e) => { + log::error!("Error fetching initiating user {initiator_id}: {e}"); + return Ok(MetaHttpResponse::unauthorized("Not Allowed")); + } + } }; let is_allowed = diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 4bb0dd2d8a..1c7f973bd7 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -2594,22 +2594,23 @@ mod tests { } async fn e2e_handle_derived_stream_evaluation_failure() { - let auth = setup(); - // Create a pipeline with invalid SQL to cause evaluation failure + // Create a pipeline that queries a non-existent stream to cause evaluation failure. + // evaluate_scheduled() returns Err when the stream is not found in schema cache, + // which should increment the trigger's retries counter. let pipeline_data = Pipeline { id: "test_derived_stream_pipeline_invalid".to_string(), version: 1, enabled: true, org: "e2e".to_string(), name: "test_derived_stream_invalid".to_string(), - description: "Test pipeline with invalid SQL".to_string(), + description: "Test pipeline with non-existent stream to cause evaluation failure" + .to_string(), source: PipelineSource::Scheduled(DerivedStream { org_id: "e2e".to_string(), stream_type: StreamType::Logs, query_condition: QueryCondition { query_type: config::meta::alerts::QueryType::SQL, - sql: Some("SELECT _timestamp, city FROM \"olympics_schema\"".to_string()), /* Invalid - * SQL */ + sql: Some("SELECT * FROM \"nonexistent_stream_xyz_abc_test\"".to_string()), ..Default::default() }, trigger_condition: TriggerCondition { @@ -2622,7 +2623,7 @@ mod tests { delay: None, }), nodes: vec![ - // Source node (query node for scheduled pipeline with invalid SQL) + // Source node (query node for scheduled pipeline with non-existent stream) config::meta::pipeline::components::Node::new( "source-node-2".to_string(), config::meta::pipeline::components::NodeData::Query(DerivedStream { @@ -2631,7 +2632,7 @@ mod tests { query_condition: QueryCondition { query_type: config::meta::alerts::QueryType::SQL, sql: Some( - "SELECT _timestamp, city FROM \"olympics_schema\"".to_string(), + "SELECT * FROM \"nonexistent_stream_xyz_abc_test\"".to_string(), ), ..Default::default() }, @@ -2669,28 +2670,26 @@ mod tests { )], }; - // Save pipeline using API call - e2e_post_pipeline(pipeline_data.clone()).await; - - // Check if pipeline was saved successfully by doing a list using API - let app = init_test_router(); - - // delete the city field from the stream - // Make a PUT call to `e2e/streams/olympics_schema/delete_fields` api - // with `{"fields":["method"]}` as the payload - let delete_fields_url = "/api/e2e/streams/olympics_schema/delete_fields"; - let delete_fields_payload = serde_json::json!({"fields": ["city"]}); - - let headers = auth_headers(auth); - let (status, _body) = make_request( - &app, - Method::PUT, - delete_fields_url, - Some(headers), - Some(serde_json::to_string(&delete_fields_payload).unwrap()), + // Save pipeline directly to DB (bypassing API validation) to simulate a pipeline + // with an invalid query that was saved before validation was added, or to test + // what happens at evaluation time when the stream does not exist. + openobserve::service::db::pipeline::set(&pipeline_data) + .await + .expect("Failed to set pipeline in DB"); + // Create the scheduler trigger directly with needs_validated=false so the + // non-existent stream does not fail the pre-save test run. + let derived_stream = match &pipeline_data.source { + PipelineSource::Scheduled(ds) => ds.clone(), + _ => panic!("Expected scheduled pipeline"), + }; + openobserve::service::alerts::derived_streams::save( + derived_stream, + &pipeline_data.name, + &pipeline_data.id, + false, ) - .await; - assert!(status.is_success(), "Failed to delete fields"); + .await + .expect("Failed to save derived stream trigger"); let pipeline = get_pipeline_from_api(pipeline_data.name.as_str()).await; diff --git a/tests/ui-testing/pages/logsPages/logsPage.js b/tests/ui-testing/pages/logsPages/logsPage.js index 877238955d..0d7f80639b 100644 --- a/tests/ui-testing/pages/logsPages/logsPage.js +++ b/tests/ui-testing/pages/logsPages/logsPage.js @@ -1128,6 +1128,9 @@ export class LogsPage { await this.page.waitForTimeout(200); // Click the q-item directly - it has @click="handleQuickMode" handler await this.page.locator(this.quickModeToggle).click(); + // Close the utilities menu - it stays open after toggle click + // (no v-close-popup on the quick mode item) + await this.page.keyboard.press('Escape'); } // Click on the Quick Mode text label (not the toggle switch) - for testing #10821 @@ -3381,12 +3384,10 @@ export class LogsPage { const isSwitchedOff = await toggleInner.evaluate(node => node.classList.contains('q-toggle__inner--falsy')).catch(() => false); if (isSwitchedOff) { await toggleInner.click(); - } else { - await this.page.keyboard.press('Escape'); } - } else { - await this.page.keyboard.press('Escape'); } + // Always close the utilities menu + await this.page.keyboard.press('Escape'); } async clickTimestampField() { diff --git a/tests/ui-testing/playwright-tests/Streams/multiselect-stream.spec.js b/tests/ui-testing/playwright-tests/Streams/multiselect-stream.spec.js index 406d118ad8..ab43f45a33 100644 --- a/tests/ui-testing/playwright-tests/Streams/multiselect-stream.spec.js +++ b/tests/ui-testing/playwright-tests/Streams/multiselect-stream.spec.js @@ -116,9 +116,10 @@ test.describe("Stream multiselect testcases", () => { } await page.goto(`${logData.logsUrl}?org_identifier=${process.env["ORGNAME"]}`); - await pm.logsPage.selectStream("e2e_automate"); + await pm.logsPage.selectStream("e2e_automate"); await applyQueryButton(page); - await pm.logsPage.clickQuickModeToggle(); + // Ensure quick mode is ON (idempotent) so the All Fields toggle is visible + await pm.logsPage.enableQuickModeIfDisabled(); await pm.logsPage.clickAllFieldsButton(); }); @@ -287,8 +288,9 @@ async function multistreamselect(page) { await pageManager.logsPage.toggleStreamSelection("e2e_stream1"); await page.waitForTimeout(4000); - // Click All Fields and enable function editor using POM - await pageManager.logsPage.clickQuickModeToggle(); + // Ensure quick mode is ON so the All Fields toggle is visible + // (quickMode default depends on whether localStorage has saved streams from prior navigation) + await pageManager.logsPage.enableQuickModeIfDisabled(); await pageManager.logsPage.clickAllFieldsButton(); await pageManager.logsPage.toggleQueryModeEditor(); await pageManager.logsPage.clickMonacoEditor(); diff --git a/web/src/components/O2AIChat.vue b/web/src/components/O2AIChat.vue index 2e03ca7757..176d0353cb 100644 --- a/web/src/components/O2AIChat.vue +++ b/web/src/components/O2AIChat.vue @@ -4280,7 +4280,7 @@ export default defineComponent({ const parsed = JSON.parse(content); const formatted = JSON.stringify(parsed, null, 2); // Apply syntax highlighting - return formatted + const highlighted = formatted .replace(/("(\\u[a-zA-Z0-9]{4}|\\[^u]|[^\\"])*"(\s*:)?|\b(true|false|null)\b|-?\d+(?:\.\d*)?(?:[eE][+\-]?\d+)?)/g, (match) => { let cls = 'json-number'; if (/^"/.test(match)) { @@ -4296,15 +4296,16 @@ export default defineComponent({ } return `${match}`; }); + return DOMPurify.sanitize(highlighted); } catch { // Not JSON, return plain text with HTML escaping - return content + return DOMPurify.sanitize(content .replace(/&/g, '&') .replace(//g, '>') .replace(/"/g, '"') .replace(/'/g, ''') - .replace(/\n/g, '
'); + .replace(/\n/g, '
')); } }; diff --git a/web/src/components/alerts/AlertSummary.vue b/web/src/components/alerts/AlertSummary.vue index f3cb3d3dc1..8ff1498891 100644 --- a/web/src/components/alerts/AlertSummary.vue +++ b/web/src/components/alerts/AlertSummary.vue @@ -17,7 +17,7 @@ along with this program. If not, see . @@ -164,7 +164,7 @@ along with this program. If not, see . @@ -181,7 +181,7 @@ along with this program. If not, see . @@ -342,6 +342,7 @@ import { useStore } from "vuex"; import { useQuasar } from "quasar"; import { date } from "quasar"; import incidentsService from "@/services/incidents"; +import DOMPurify from "dompurify"; interface Props { orgId: string; @@ -546,8 +547,15 @@ const getInlineEventText = (event: any): string => { const eventColor = getEventBadgeColor(event); const isDark = store.state.theme === 'dark'; const opacity = isDark ? '50' : '40'; - const bold = (text: string) => `${text}`; - const severityBadge = (severity: string) => `${severity}`; + // Escape user-controlled strings before embedding in HTML (XSS prevention) + const esc = (s: string) => String(s) + .replace(/&/g, '&') + .replace(//g, '>') + .replace(/"/g, '"') + .replace(/'/g, '''); + const bold = (text: string) => `${esc(text)}`; + const severityBadge = (severity: string) => `${esc(severity)}`; const isSystemEvent = getUserId(event) === 'System'; switch (event.type) { @@ -570,8 +578,8 @@ const getInlineEventText = (event: any): string => { case "SeverityUpgrade": return isSystemEvent - ? `Severity upgraded from ${severityBadge(data.from)} to ${severityBadge(data.to)}` + (data.reason ? ` - ${data.reason}` : '') - : `changed the severity from ${severityBadge(data.from)} to ${severityBadge(data.to)}` + (data.reason ? ` - ${data.reason}` : ''); + ? `Severity upgraded from ${severityBadge(data.from)} to ${severityBadge(data.to)}` + (data.reason ? ` - ${esc(data.reason)}` : '') + : `changed the severity from ${severityBadge(data.from)} to ${severityBadge(data.to)}` + (data.reason ? ` - ${esc(data.reason)}` : ''); case "SeverityOverride": return isSystemEvent diff --git a/web/src/components/dashboards/addPanel/StreamFieldSelect.vue b/web/src/components/dashboards/addPanel/StreamFieldSelect.vue index f2b04fa46b..0a2fa92479 100644 --- a/web/src/components/dashboards/addPanel/StreamFieldSelect.vue +++ b/web/src/components/dashboards/addPanel/StreamFieldSelect.vue @@ -36,7 +36,7 @@ :data-test="`stream-field-select-option-${child.name}`" > - + {{ child.name }} diff --git a/web/src/components/dashboards/settings/VariableCustomValueSelector.vue b/web/src/components/dashboards/settings/VariableCustomValueSelector.vue index d0e9203bc9..2cde5e2195 100644 --- a/web/src/components/dashboards/settings/VariableCustomValueSelector.vue +++ b/web/src/components/dashboards/settings/VariableCustomValueSelector.vue @@ -75,7 +75,7 @@ along with this program. If not, see . /> - + {{ opt.label }} diff --git a/web/src/components/settings/License.vue b/web/src/components/settings/License.vue index ef230b1586..a05b48256b 100644 --- a/web/src/components/settings/License.vue +++ b/web/src/components/settings/License.vue @@ -17,7 +17,7 @@
{{ t('about.installation_id') }}: {{ licenseData.installation_id || 'N/A' }}
-
+
- + @@ -231,18 +231,18 @@ /> - 30 ? 'text-negative' : 'text-warning', days: licenseData?.ingestion_exceeded, plural: licenseData?.ingestion_exceeded > 1 ? 's' : '' - })">. + }))">. {{ t('about.no_limit_exceedances', { max: limitBreachAllowedCount }) }} @@ -327,6 +327,7 @@ import { useQuasar } from "quasar"; import { useI18n } from "vue-i18n"; import licenseServer from "@/services/license_server"; import { useStore } from "vuex"; +import DOMPurify from 'dompurify'; import LicensePeriod from "@/enterprise/components/billings/LicensePeriod.vue"; const RenderDashboardCharts = defineAsyncComponent( @@ -803,6 +804,7 @@ export default defineComponent({ ingestionLimitGB, generateUsageDashboard, limitBreachAllowedCount, + DOMPurify, }; }, }); diff --git a/web/src/plugins/logs/Index.vue b/web/src/plugins/logs/Index.vue index f4ac3e4e5b..b884897b0a 100644 --- a/web/src/plugins/logs/Index.vue +++ b/web/src/plugins/logs/Index.vue @@ -122,8 +122,8 @@ along with this program. If not, see . />
+ style="white-space: pre-line" + >{{ searchObj.data.filterErrMsg }}
.