sift_query_sqlx/web/
search.rs

1use std::time::Duration;
2
3use sift_query_core::{ParamError, QueryLimits, RawParams};
4
5use super::{PublicError, WebError};
6
7/// GET query / POST form parameters, preserving ordered repeated values.
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct SearchParams(pub RawParams);
10
11/// Framework request configuration. Use the same limits as the query policy.
12/// Install as an axum extension, actix app data, or poem endpoint data.
13#[derive(Debug, Clone)]
14pub struct SearchParamConfig {
15    pub limits: QueryLimits,
16    pub body_timeout: Duration,
17}
18
19impl Default for SearchParamConfig {
20    fn default() -> Self {
21        Self {
22            limits: QueryLimits::default(),
23            body_timeout: Duration::from_secs(5),
24        }
25    }
26}
27
28impl SearchParams {
29    #[must_use]
30    pub fn raw(&self) -> &RawParams {
31        &self.0
32    }
33
34    /// Validates transport metadata before reading the body. POST query
35    /// parameters are rejected rather than merged with form parameters.
36    pub fn validate_request(
37        method: &str,
38        query: Option<&str>,
39        content_type: Option<&str>,
40    ) -> Result<(), WebError> {
41        match method {
42            "GET" => Ok(()),
43            "POST" => {
44                if query.is_some_and(|query| !query.is_empty()) {
45                    return Err(error(400, "conflicting_query_body"));
46                }
47                let mut parts = content_type.unwrap_or_default().split(';');
48                if !parts
49                    .next()
50                    .unwrap_or_default()
51                    .trim()
52                    .eq_ignore_ascii_case("application/x-www-form-urlencoded")
53                    || parts.any(|part| {
54                        let Some((name, value)) = part.trim().split_once('=') else {
55                            return true;
56                        };
57                        !name.trim().eq_ignore_ascii_case("charset")
58                            || !value.trim().trim_matches('"').eq_ignore_ascii_case("utf-8")
59                    })
60                {
61                    return Err(error(415, "unsupported_query_media_type"));
62                }
63                Ok(())
64            }
65            _ => Err(error(405, "query_method_not_allowed")),
66        }
67    }
68
69    /// Parses an already bounded request body using the same core protocol
70    /// and budgets as GET. Does not bypass entity or query validation.
71    pub fn parse_request(
72        method: &str,
73        query: Option<&str>,
74        content_type: Option<&str>,
75        body: &[u8],
76        limits: &QueryLimits,
77    ) -> Result<Self, WebError> {
78        Self::validate_request(method, query, content_type)?;
79        let input = if method == "GET" {
80            query.unwrap_or_default()
81        } else {
82            if body.len() > limits.max_raw_query_bytes {
83                return Err(error(413, "query_body_too_large"));
84            }
85            std::str::from_utf8(body).map_err(|_| WebError::from(ParamError::InvalidEncoding))?
86        };
87        RawParams::parse_with_limits(input, limits)
88            .map(Self)
89            .map_err(WebError::from)
90    }
91}
92
93pub(super) const fn error(status: u16, code: &'static str) -> WebError {
94    WebError(PublicError { status, code })
95}
96
97#[cfg(any(feature = "axum", feature = "actix-web", feature = "poem", test))]
98pub(super) async fn read_body<S, B, E>(
99    stream: S,
100    config: &SearchParamConfig,
101) -> Result<Vec<u8>, WebError>
102where
103    S: futures_core::Stream<Item = Result<B, E>>,
104    B: AsRef<[u8]>,
105{
106    use futures_util::StreamExt;
107    tokio::time::timeout(config.body_timeout, async {
108        futures_util::pin_mut!(stream);
109        let mut bytes = Vec::new();
110        while let Some(chunk) = stream.next().await {
111            let chunk = chunk.map_err(|_| error(400, "invalid_query_body"))?;
112            let chunk = chunk.as_ref();
113            if chunk.len()
114                > config
115                    .limits
116                    .max_raw_query_bytes
117                    .saturating_sub(bytes.len())
118            {
119                return Err(error(413, "query_body_too_large"));
120            }
121            bytes.extend_from_slice(chunk);
122        }
123        Ok(bytes)
124    })
125    .await
126    .map_err(|_| error(408, "query_body_timeout"))?
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[tokio::test]
134    async fn chunked_bodies_are_bounded_and_pending_readers_expire() {
135        let config = SearchParamConfig {
136            limits: QueryLimits {
137                max_raw_query_bytes: 3,
138                ..Default::default()
139            },
140            body_timeout: Duration::from_millis(5),
141        };
142        let chunks =
143            futures_util::stream::iter([Ok::<_, ()>(b"id".as_slice()), Ok(b"=1".as_slice())]);
144        assert_eq!(
145            read_body(chunks, &config).await.unwrap_err().0.code,
146            "query_body_too_large"
147        );
148        let pending = futures_util::stream::pending::<Result<Vec<u8>, ()>>();
149        assert_eq!(
150            read_body(pending, &config).await.unwrap_err().0.code,
151            "query_body_timeout"
152        );
153    }
154}
sift_query_sqlx/web/search.rs

sift_query_sqlx/web/
search.rs
