From 3f593f4c1d22d0262b7585c3716ff582106726e0 Mon Sep 17 00:00:00 2001 From: eastagiletracker <310448263+eastagiletracker@users.noreply.github.com> Date: Tue, 18 Aug 2026 09:23:19 +0000 Subject: [PATCH] Reject malformed record queries instead of silently ignoring them The topic iteration route extracted its query string with Option>, so any parameter that failed to parse discarded the whole query string and fell back to defaults. A request carrying a partition filter alongside an unparseable parameter (order=descending, say) returned records from every partition with a 200, giving no sign that the filter had been dropped. PartitionSelector deserialization had a similar fallback: a "regex:" selector that would not compile was treated as an exact partition name, so the query matched nothing and read as an empty topic. Treat a missing query string as an empty one so all-default requests still parse, report serde_qs failures through ErrorReply as the 400 the rest of the API returns, take the query string as a required extractor on the iteration route, and report uncompilable regex selectors when they are deserialized. PartitionSelector::from keeps its infallible fallback, so in-process callers are unaffected. --- server/src/axum_util/query.rs | 16 +++-- server/src/http.rs | 12 +++- server/src/http/error.rs | 4 ++ server/tests/server.rs | 124 ++++++++++++++++++++++++++++++++++ transport/src/lib.rs | 88 ++++++++++++++++++++---- 5 files changed, 223 insertions(+), 21 deletions(-) diff --git a/server/src/axum_util/query.rs b/server/src/axum_util/query.rs index 3e542f8..58dd3b6 100644 --- a/server/src/axum_util/query.rs +++ b/server/src/axum_util/query.rs @@ -3,13 +3,15 @@ use axum::http; use axum::response; use serde::de; +use crate::http::ErrorReply; + #[derive(Debug)] pub struct Query(pub T); #[derive(Debug)] #[non_exhaustive] pub enum QueryRejection { - FailedToDeserializeQueryString, + FailedToDeserializeQueryString(String), } #[axum::async_trait] @@ -24,20 +26,20 @@ where parts: &mut http::request::Parts, _state: &S, ) -> Result { - let query = parts - .uri - .query() - .ok_or(QueryRejection::FailedToDeserializeQueryString)?; + // A request without a query string carries the same information as one + // with an empty query string: every parameter takes its default. + let query = parts.uri.query().unwrap_or_default(); let config = serde_qs::Config::new(2, false); config .deserialize_str(query) .map(Query) - .map_err(|_| QueryRejection::FailedToDeserializeQueryString) + .map_err(|e| QueryRejection::FailedToDeserializeQueryString(e.to_string())) } } impl response::IntoResponse for QueryRejection { fn into_response(self) -> response::Response { - http::StatusCode::NOT_ACCEPTABLE.into_response() + let Self::FailedToDeserializeQueryString(detail) = self; + ErrorReply::InvalidQueryString(detail).into_response() } } diff --git a/server/src/http.rs b/server/src/http.rs index 713fa50..1af710a 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -311,12 +311,20 @@ async fn topic_get_info( async fn topic_iterate_route( State(AppState(catalog, config)): State, Path(topic_name): Path, - query: Option>, + query: Query, headers: HeaderMap, position: Option>, ) -> Result { let max_page = config.http.max_page; - topic_iterate(topic_name, query, headers, position, catalog, max_page).await + topic_iterate( + topic_name, + Some(query), + headers, + position, + catalog, + max_page, + ) + .await } pub async fn topic_iterate( diff --git a/server/src/http/error.rs b/server/src/http/error.rs index a7c3e86..67c9eb5 100644 --- a/server/src/http/error.rs +++ b/server/src/http/error.rs @@ -11,6 +11,7 @@ pub enum ErrorReply { EmptyBody, WriterBusy, InvalidQuery, + InvalidQueryString(String), InvalidSchema, NullTypes, BadEncoding, @@ -30,6 +31,9 @@ impl axum::response::IntoResponse for ErrorReply { Self::Chunk(e) => (StatusCode::BAD_REQUEST, format!("chunk error: {e}")), Self::Path(e) => (StatusCode::BAD_REQUEST, format!("invalid path: {e}")), Self::InvalidQuery => (StatusCode::BAD_REQUEST, "invalid query".to_string()), + Self::InvalidQueryString(detail) => { + (StatusCode::BAD_REQUEST, format!("invalid query: {detail}")) + } Self::InvalidSchema => (StatusCode::BAD_REQUEST, "invalid schema".to_string()), Self::NullTypes => ( StatusCode::BAD_REQUEST, diff --git a/server/tests/server.rs b/server/tests/server.rs index f580422..8e7a8a2 100644 --- a/server/tests/server.rs +++ b/server/tests/server.rs @@ -771,6 +771,130 @@ async fn topic_time_query() -> Result<()> { Ok(()) } +/// Fill two partitions of `topic_name` with `count` records each. +async fn append_two_partitions( + client: &Client, + server: &TestServer, + topic_name: &str, + count: usize, +) { + for partition in ["alpha", "beta"] { + repeat_append( + client, + append_url(server, topic_name, partition).as_str(), + TEST_MESSAGE, + count, + ) + .await; + } + server.catalog.checkpoint().await; +} + +async fn assert_error_reply(response: Response, expected: reqwest::StatusCode) { + assert_eq!(response.status(), expected); + let body: json::Value = response.json().await.unwrap(); + let body = body.as_object().expect("expected an error object"); + assert_eq!( + body.get("code").and_then(json::Value::as_u64), + Some(expected.as_u16() as u64) + ); + assert!(body.contains_key("message"), "expected a message: {body:?}"); +} + +#[test_log::test(tokio::test)] +async fn topic_iterate_partition_filter() -> Result<()> { + let (client, topic_name, server) = setup().await; + append_two_partitions(&client, &server, &topic_name, 5).await; + let url = topic_records_url(&server, &topic_name); + + // no query string at all: every partition is iterated + let response = client.post(&url).json(&json::json!({})).send().await?; + assert_response_length(response.error_for_status()?, 10).await; + + // an exact partition name, and the equivalent regex, select one partition + for filter in ["alpha", "regex:^alpha$"] { + let response = client + .post(&url) + .query(&[("partition_filter[]", filter)]) + .json(&json::json!({})) + .send() + .await?; + assert_response_length(response.error_for_status()?, 5).await; + } + + Ok(()) +} + +#[test_log::test(tokio::test)] +async fn topic_iterate_rejects_malformed_query() -> Result<()> { + let (client, topic_name, server) = setup().await; + append_two_partitions(&client, &server, &topic_name, 5).await; + let url = topic_records_url(&server, &topic_name); + + // `descending` is not a valid order. Rejecting the request keeps the + // partition filter that came with it from being silently discarded, which + // would widen the query to every partition. + let response = client + .post(&url) + .query(&[("partition_filter[]", "alpha"), ("order", "descending")]) + .json(&json::json!({})) + .send() + .await?; + assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await; + + // the same request with a valid order still reads just that partition + let response = client + .post(&url) + .query(&[("partition_filter[]", "alpha"), ("order", "desc")]) + .json(&json::json!({})) + .send() + .await?; + assert_response_length(response.error_for_status()?, 5).await; + + Ok(()) +} + +#[test_log::test(tokio::test)] +async fn topic_iterate_rejects_uncompilable_partition_regex() -> Result<()> { + let (client, topic_name, server) = setup().await; + append_two_partitions(&client, &server, &topic_name, 5).await; + let url = topic_records_url(&server, &topic_name); + + // look-around is not supported by the regex crate, so this selector cannot + // compile: report it rather than matching it as a literal partition name, + // which reads as an empty topic. + for filter in ["regex:(?=alpha)", "regex:alpha("] { + let response = client + .post(&url) + .query(&[("partition_filter[]", filter)]) + .json(&json::json!({})) + .send() + .await?; + assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await; + } + + Ok(()) +} + +#[test_log::test(tokio::test)] +async fn partition_records_rejects_malformed_query() -> Result<()> { + let (client, topic_name, server) = setup().await; + append_two_partitions(&client, &server, &topic_name, 5).await; + let url = partition_records_url(&server, &topic_name, "alpha"); + + let response = client + .get(&url) + .query(&[("start", "0"), ("page_size", "many")]) + .send() + .await?; + assert_error_reply(response, reqwest::StatusCode::BAD_REQUEST).await; + + let response = client.get(&url).query(&[("start", "0")]).send().await?; + assert_response_length(response.error_for_status()?, 5).await; + + Ok(()) +} + #[test_log::test(tokio::test)] async fn topic_iterate_pandas_records() -> Result<()> { let (client, topic_name, server) = setup().await; diff --git a/transport/src/lib.rs b/transport/src/lib.rs index 6020cad..f5f3d91 100644 --- a/transport/src/lib.rs +++ b/transport/src/lib.rs @@ -336,8 +336,8 @@ pub struct TopicIterationQuery { /// begin with "regex:" to signify that any partition matching the following string can be converted. pub type PartitionFilter = Option>; -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, utoipa::ToSchema)] -#[serde(from = "String", into = "String")] +#[derive(Clone, Debug, serde::Serialize, utoipa::ToSchema)] +#[serde(into = "String")] pub enum PartitionSelector { /// The exact name of a partition. String(String), @@ -345,6 +345,11 @@ pub enum PartitionSelector { Regex(Regex), } +/// Returned when a `regex:` prefixed [`PartitionSelector`] does not compile. +#[derive(Clone, Debug, Error)] +#[error("invalid partition selector: {0}")] +pub struct PartitionSelectorError(String); + #[cfg(feature = "rweb")] impl Entity for PartitionSelector { fn type_name() -> std::borrow::Cow<'static, str> { @@ -377,25 +382,46 @@ impl From for String { } } +/// Note that this conversion cannot fail: a `regex:` selector that does not +/// compile falls back to the exact partition name. Use +/// [`PartitionSelector::parse`] to reject such a selector instead. impl From for PartitionSelector where T: AsRef, { fn from(text: T) -> Self { - let build_regex = |pattern| { - regex::RegexBuilder::new(pattern) - .size_limit(2 << 12) - .build() - .ok() - }; - text.as_ref() - .strip_prefix("regex:") - .and_then(build_regex) - .map_or_else(|| Self::String(text.as_ref().to_string()), Self::Regex) + let text = text.as_ref(); + Self::parse(text).unwrap_or_else(|_| Self::String(text.to_string())) + } +} + +impl<'de> Deserialize<'de> for PartitionSelector { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let text = String::deserialize(deserializer)?; + Self::parse(&text).map_err(serde::de::Error::custom) } } impl PartitionSelector { + /// Compiled size limit for a `regex:` selector. + const REGEX_SIZE_LIMIT: usize = 2 << 12; + + /// Build a selector from its string form, reporting a `regex:` pattern that + /// does not compile rather than silently treating it as a partition name. + pub fn parse(text: &str) -> Result { + match text.strip_prefix("regex:") { + Some(pattern) => regex::RegexBuilder::new(pattern) + .size_limit(Self::REGEX_SIZE_LIMIT) + .build() + .map(Self::Regex) + .map_err(|e| PartitionSelectorError(e.to_string())), + None => Ok(Self::String(text.to_string())), + } + } + pub fn matches(&self, name: &str) -> bool { match self { Self::String(s) => s == name, @@ -1078,6 +1104,8 @@ impl fmt::Display for PartitionId { mod tests { use super::*; use arrow_array::{Int32Array, Int64Array}; + use serde::de::value::{Error as ValueError, StrDeserializer}; + use serde::de::IntoDeserializer; fn nested_chunk() -> ( SchemaChunk, @@ -1252,6 +1280,42 @@ mod tests { assert_eq!(ps.to_string(), r"regex:\p{Greek}"); } + #[test] + fn partition_selector_reports_uncompilable_regex() { + // look-around is unsupported, and this pattern is far past the size limit + for pattern in [ + "regex:(?=alpha)", + "regex:alpha(", + &format!("regex:{}", "a{100}".repeat(100)), + ] { + assert!( + PartitionSelector::parse(pattern).is_err(), + "expected {pattern} to be rejected" + ); + + let deserializer: StrDeserializer<'_, ValueError> = pattern.into_deserializer(); + assert!( + PartitionSelector::deserialize(deserializer).is_err(), + "expected {pattern} to fail deserialization" + ); + + // the infallible conversion keeps its documented fallback + assert!(matches!( + PartitionSelector::from(pattern), + PartitionSelector::String(_) + )); + } + } + + #[test] + fn partition_selector_deserializes_valid_selectors() { + for (pattern, name) in [("alpha", "alpha"), ("regex:^alpha$", "alpha")] { + let deserializer: StrDeserializer<'_, ValueError> = pattern.into_deserializer(); + let selector = PartitionSelector::deserialize(deserializer).unwrap(); + assert!(selector.matches(name)); + } + } + #[test] fn test_focus() { let (test, (time, _child_struct, _grandchild_struct, _index)) = nested_chunk();