From 7fcb4ae1fa62edbc8d0656e75146fef8465fc462 Mon Sep 17 00:00:00 2001 From: Amey Pawar <138877912+ameyypawar@users.noreply.github.com> Date: Tue, 6 Oct 2026 07:06:50 +0000 Subject: [PATCH] fix(search): refuse malformed multi-stream requests instead of panicking (#15030) --- src/api/search/src/search/multi_streams.rs | 71 ++++++++++++++++++- .../test_multi_stream_malformed_requests.py | 39 ++++++++++ 2 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 tests/api-testing/tests/search/test_multi_stream_malformed_requests.py diff --git a/src/api/search/src/search/multi_streams.rs b/src/api/search/src/search/multi_streams.rs index a814f66756..aaffd85055 100644 --- a/src/api/search/src/search/multi_streams.rs +++ b/src/api/search/src/search/multi_streams.rs @@ -219,6 +219,10 @@ pub async fn search_multi( let user_id = &user_email.user_id; let mut queries = multi_req.to_query_req(); + // The count divides cached_ratio below; the router answers an empty sql with this message. + if queries.is_empty() { + return MetaHttpResponse::bad_request("Failed to parse multi search request"); + } let mut multi_res = search::Response::new(multi_req.from, multi_req.size); let per_query_resp = multi_req.per_query_response; @@ -866,6 +870,7 @@ pub async fn _search_partition_multi( "size": 10, "scan_size": 28943 })), + (status = 400, description = "Failure", content_type = "application/json", body = ()), (status = 500, description = "Failure", content_type = "application/json", body = ()), ) )] @@ -905,6 +910,13 @@ pub async fn around_multi( .collect::>(); if let Some(v) = query.get("sql") { let sqls = v.split(',').collect::>(); + if sqls.len() > around_sqls.len() { + return MetaHttpResponse::bad_request(format!( + "sql has more entries ({}) than streams ({})", + sqls.len(), + around_sqls.len() + )); + } for (i, sql) in sqls.into_iter().enumerate() { if let Ok(sql) = base64::decode_url(sql) { around_sqls[i] = sql; @@ -1589,8 +1601,20 @@ pub async fn report_to_audit( #[cfg(test)] mod tests { + use axum::{ + extract::{Path, Query}, + http::{HeaderMap, StatusCode}, + }; use chrono::Utc; - use config::meta::search::{MultiSearchPartitionRequest, MultiStreamRequest}; + use config::{ + meta::search::{MultiSearchPartitionRequest, MultiStreamRequest}, + utils::base64, + }; + use hashbrown::HashMap; + use openobserve_api_common::extractors::Headers; + use openobserve_core::auth::UserEmail; + + use super::around_multi; #[test] fn test_multi_stream_request_structure() { @@ -2198,4 +2222,49 @@ mod tests { "per_query_response=false should always set query_fn on requests" ); } + + // The cloud build checks the org's trial period first, which can answer 403 before this guard. + #[cfg(not(feature = "cloud"))] + #[tokio::test] + async fn search_multi_refuses_an_empty_sql_list_instead_of_panicking() { + use axum::Json; + use config::utils::json; + + use super::search_multi; + + for body in [ + r#"{"sql": [], "start_time": 1, "end_time": 2}"#, + r#"{"start_time": 1, "end_time": 2}"#, + ] { + let req: MultiStreamRequest = json::from_str(body).unwrap(); + let resp = search_multi( + Path("default".to_string()), + Query(HashMap::new()), + Headers(UserEmail { + user_id: "user@example.com".to_string(), + }), + HeaderMap::new(), + Json(req), + ) + .await; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "{body}"); + } + } + + #[tokio::test] + async fn around_multi_refuses_more_sql_entries_than_streams_instead_of_panicking() { + let sql = base64::encode_url(r#"SELECT * FROM "default""#); + let mut query = HashMap::new(); + query.insert("sql".to_string(), format!("{sql},{sql}")); + let resp = around_multi( + Path(("default".to_string(), base64::encode_url("default"))), + HeaderMap::new(), + Query(query), + Headers(UserEmail { + user_id: "user@example.com".to_string(), + }), + ) + .await; + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + } } diff --git a/tests/api-testing/tests/search/test_multi_stream_malformed_requests.py b/tests/api-testing/tests/search/test_multi_stream_malformed_requests.py new file mode 100644 index 0000000000..992d090ac9 --- /dev/null +++ b/tests/api-testing/tests/search/test_multi_stream_malformed_requests.py @@ -0,0 +1,39 @@ +"""Malformed multi-stream search requests are answered with 400 (#14915). + +`_search_multi` divides by the number of queries, and `_around_multi` writes each `sql` entry into +its per-stream list by position. Both requests below must be refused before those lines, or the +handler panics and the client gets no response at all. +""" + +from __future__ import annotations + +import pytest + +from support.client import OpenObserveClient + +# URL-safe base64, as config::utils::base64::encode_url produces it +DEFAULT_STREAM = "ZGVmYXVsdA.." # default +SELECT_DEFAULT = "U0VMRUNUICogRlJPTSAiZGVmYXVsdCI." # SELECT * FROM "default" + + +@pytest.mark.parametrize( + "body", + [ + {"sql": [], "start_time": 1, "end_time": 2}, + {"start_time": 1, "end_time": 2}, + ], + ids=["empty-sql-list", "no-sql-field"], +) +def test_search_multi_without_queries_is_a_bad_request(client: OpenObserveClient, body): + resp = client.post("_search_multi", json=body) + assert resp.status_code == 400, resp.text + assert resp.json()["message"] == "Failed to parse multi search request" + + +def test_around_multi_with_more_sql_than_streams_is_a_bad_request(client: OpenObserveClient): + resp = client.get( + f"{DEFAULT_STREAM}/_around_multi", + params={"key": 1, "sql": f"{SELECT_DEFAULT},{SELECT_DEFAULT}"}, + ) + assert resp.status_code == 400, resp.text + assert resp.json()["message"] == "sql has more entries (2) than streams (1)"