tako_rs_core/graphql/
request.rs

1//! `GraphQL` HTTP request extraction: single and batch extractors, body-size
2//! limits, content-type classification, and the `receive_*` helpers.
3
4use async_graphql::BatchRequest as GqlBatchRequest;
5use async_graphql::http::MultipartOptions;
6use http::StatusCode;
7use http_body_util::BodyExt;
8
9use crate::extractors::FromRequest;
10use crate::responder::Responder;
11use crate::types::Request;
12use crate::types::Response;
13
14/// Single `GraphQL` request extractor.
15pub struct GraphQLRequest(pub async_graphql::Request);
16
17impl GraphQLRequest {
18  pub fn into_inner(self) -> async_graphql::Request {
19    self.0
20  }
21}
22
23/// Batch `GraphQL` request extractor.
24pub struct GraphQLBatchRequest(pub GqlBatchRequest);
25
26impl GraphQLBatchRequest {
27  pub fn into_inner(self) -> GqlBatchRequest {
28    self.0
29  }
30}
31
32/// Cap on the raw POST body GraphQL extractors will buffer.
33///
34/// Async-graphql parses the entire request body into memory before it can
35/// validate the query, so without an upstream limit a single unauthenticated
36/// POST could buffer many GB and OOM the process. `4 MiB` matches the default
37/// body limits of comparable frameworks (Apollo, Hasura, federation gateways)
38/// and is large enough for any realistic GraphQL document plus variables.
39pub const MAX_GRAPHQL_BODY_SIZE: usize = 4 * 1024 * 1024;
40
41/// Errors that can occur while parsing `GraphQL` HTTP requests.
42#[derive(Debug)]
43pub enum GraphQLError {
44  MissingQuery,
45  BodyRead(String),
46  /// Request body exceeds [`MAX_GRAPHQL_BODY_SIZE`] — either by
47  /// advertised `Content-Length` or by actual streamed bytes.
48  BodyTooLarge,
49  InvalidJson(String),
50  Parse(String),
51  UnsupportedMediaType(String),
52}
53
54/// Per-request or global options for `GraphQL` extraction.
55#[derive(Clone, Default)]
56pub struct GraphQLOptions {
57  pub multipart: MultipartOptions,
58}
59
60impl Responder for GraphQLError {
61  fn into_response(self) -> Response {
62    match self {
63      GraphQLError::MissingQuery => {
64        (StatusCode::BAD_REQUEST, "Missing GraphQL query").into_response()
65      }
66      GraphQLError::BodyRead(e) => {
67        (StatusCode::BAD_REQUEST, format!("Failed to read body: {e}")).into_response()
68      }
69      GraphQLError::BodyTooLarge => (
70        StatusCode::PAYLOAD_TOO_LARGE,
71        format!("GraphQL body exceeds {MAX_GRAPHQL_BODY_SIZE} bytes"),
72      )
73        .into_response(),
74      GraphQLError::InvalidJson(e) => {
75        (StatusCode::BAD_REQUEST, format!("Invalid JSON: {e}")).into_response()
76      }
77      GraphQLError::Parse(e) => {
78        (StatusCode::BAD_REQUEST, format!("Invalid request: {e}")).into_response()
79      }
80      GraphQLError::UnsupportedMediaType(ct) => (
81        StatusCode::UNSUPPORTED_MEDIA_TYPE,
82        format!("Unsupported GraphQL content-type: {ct}"),
83      )
84        .into_response(),
85    }
86  }
87}
88
89/// Returns the `GraphQL` POST body media-type bucket if the request's
90/// `Content-Type` header advertises one async-graphql understands, or
91/// `Err(UnsupportedMediaType)` otherwise. Used to fail fast before buffering
92/// a body that the parser would reject anyway with a confusing message.
93fn classify_graphql_content_type(ct: Option<&str>) -> Result<GraphQLBodyKind, GraphQLError> {
94  let raw = ct.unwrap_or("").trim();
95  if raw.is_empty() {
96    return Err(GraphQLError::UnsupportedMediaType("<missing>".to_string()));
97  }
98  let essence = raw
99    .split(';')
100    .next()
101    .unwrap_or("")
102    .trim()
103    .to_ascii_lowercase();
104  match essence.as_str() {
105    "application/json" => Ok(GraphQLBodyKind::Json),
106    "application/graphql" | "application/graphql-response+json" => Ok(GraphQLBodyKind::Graphql),
107    "multipart/form-data" => Ok(GraphQLBodyKind::Multipart),
108    _ => Err(GraphQLError::UnsupportedMediaType(raw.to_string())),
109  }
110}
111
112#[derive(Copy, Clone, Debug, PartialEq, Eq)]
113enum GraphQLBodyKind {
114  Json,
115  Graphql,
116  Multipart,
117}
118
119#[inline]
120fn resolve_opts(req: &Request) -> MultipartOptions {
121  // Prefer per-request options in extensions
122  if let Some(opts) = req.extensions().get::<GraphQLOptions>() {
123    return opts.multipart;
124  }
125  // Fallback to global state
126  if let Some(global) = crate::state::get_state::<GraphQLOptions>() {
127    return global.as_ref().multipart;
128  }
129  MultipartOptions::default()
130}
131
132fn parse_get_request(req: &Request) -> Result<async_graphql::Request, GraphQLError> {
133  let qs = req.uri().query().unwrap_or("");
134  async_graphql::http::parse_query_string(qs).map_err(|e| GraphQLError::Parse(e.to_string()))
135}
136
137async fn read_body_bytes(req: &mut Request) -> Result<bytes::Bytes, GraphQLError> {
138  // Pre-check the advertised length: if the client says >MAX up front,
139  // refuse without touching the body at all. Defends against allocation
140  // pressure from header-only flooders.
141  if let Some(cl) = req.headers().get(http::header::CONTENT_LENGTH)
142    && let Some(n) = cl.to_str().ok().and_then(|s| s.parse::<usize>().ok())
143    && n > MAX_GRAPHQL_BODY_SIZE
144  {
145    return Err(GraphQLError::BodyTooLarge);
146  }
147
148  // Then wrap the body in `Limited` so a missing or lying Content-Length
149  // (chunked transfer, HTTP/2 without length) still cannot drag us past
150  // the cap. Same pattern as the idempotency / hmac / json-schema paths.
151  let body = std::mem::take(req.body_mut());
152  let limited = http_body_util::Limited::new(body, MAX_GRAPHQL_BODY_SIZE);
153  match limited.collect().await {
154    Ok(c) => Ok(c.to_bytes()),
155    Err(e) => {
156      // `Limited` surfaces `LengthLimitError` on cap overrun; otherwise
157      // it's a transport / body error. Use the type-name to disambiguate
158      // without depending on the private error path.
159      if e
160        .downcast_ref::<http_body_util::LengthLimitError>()
161        .is_some()
162      {
163        Err(GraphQLError::BodyTooLarge)
164      } else {
165        Err(GraphQLError::BodyRead(e.to_string()))
166      }
167    }
168  }
169}
170
171impl<'a> FromRequest<'a> for GraphQLRequest {
172  type Error = GraphQLError;
173
174  fn from_request(
175    req: &'a mut Request,
176  ) -> impl core::future::Future<Output = core::result::Result<Self, Self::Error>> + Send + 'a {
177    async move {
178      if req.method() == http::Method::GET {
179        return Ok(GraphQLRequest(parse_get_request(req)?));
180      }
181
182      // Resolve MultipartOptions: request extensions -> global state -> default
183      let opts = resolve_opts(req);
184
185      let content_type = req
186        .headers()
187        .get(http::header::CONTENT_TYPE)
188        .and_then(|v| v.to_str().ok())
189        .map(std::string::ToString::to_string);
190      classify_graphql_content_type(content_type.as_deref())?;
191
192      let body = read_body_bytes(req).await?;
193      if body.is_empty() {
194        return Err(GraphQLError::Parse("empty request body".to_string()));
195      }
196
197      let reader = futures_util::io::Cursor::new(body.to_vec());
198      let req = async_graphql::http::receive_body(content_type.as_deref(), reader, opts)
199        .await
200        .map_err(|e| GraphQLError::Parse(e.to_string()))?;
201      Ok(GraphQLRequest(req))
202    }
203  }
204}
205
206/// Helper to receive a single `GraphQL` request with custom `MultipartOptions`.
207/// Attach per-request `GraphQL` options into request extensions.
208pub fn attach_graphql_options(req: &mut Request, opts: GraphQLOptions) {
209  req.extensions_mut().insert(opts);
210}
211
212/// Set global `GraphQL` options via Tako's global state.
213pub fn set_global_graphql_options(opts: GraphQLOptions) {
214  crate::state::set_state::<GraphQLOptions>(opts);
215}
216
217pub async fn receive_graphql(
218  req: &mut Request,
219  opts: MultipartOptions,
220) -> Result<async_graphql::Request, GraphQLError> {
221  if req.method() == http::Method::GET {
222    return parse_get_request(req);
223  }
224  let body = read_body_bytes(req).await?;
225  let content_type = req
226    .headers()
227    .get(http::header::CONTENT_TYPE)
228    .and_then(|v| v.to_str().ok())
229    .map(std::string::ToString::to_string);
230  let reader = futures_util::io::Cursor::new(body.to_vec());
231  async_graphql::http::receive_body(content_type.as_deref(), reader, opts)
232    .await
233    .map_err(|e| GraphQLError::Parse(e.to_string()))
234}
235
236/// Helper to receive a batch `GraphQL` request with custom `MultipartOptions`.
237pub async fn receive_graphql_batch(
238  req: &mut Request,
239  opts: MultipartOptions,
240) -> Result<GqlBatchRequest, GraphQLError> {
241  if req.method() == http::Method::GET {
242    let single = parse_get_request(req)?;
243    return Ok(GqlBatchRequest::Single(single));
244  }
245  let content_type = req
246    .headers()
247    .get(http::header::CONTENT_TYPE)
248    .and_then(|v| v.to_str().ok())
249    .map(std::string::ToString::to_string);
250  classify_graphql_content_type(content_type.as_deref())?;
251  let body = read_body_bytes(req).await?;
252  if body.is_empty() {
253    return Err(GraphQLError::Parse("empty request body".to_string()));
254  }
255  let reader = futures_util::io::Cursor::new(body.to_vec());
256  async_graphql::http::receive_batch_body(content_type.as_deref(), reader, opts)
257    .await
258    .map_err(|e| GraphQLError::Parse(e.to_string()))
259}
260
261impl<'a> FromRequest<'a> for GraphQLBatchRequest {
262  type Error = GraphQLError;
263
264  fn from_request(
265    req: &'a mut Request,
266  ) -> impl core::future::Future<Output = core::result::Result<Self, Self::Error>> + Send + 'a {
267    async move {
268      if req.method() == http::Method::GET {
269        // Treat GET as single request
270        let single = parse_get_request(req)?;
271        return Ok(GraphQLBatchRequest(GqlBatchRequest::Single(single)));
272      }
273
274      // Resolve MultipartOptions: request extensions -> global state -> default
275      let opts = resolve_opts(req);
276
277      let content_type = req
278        .headers()
279        .get(http::header::CONTENT_TYPE)
280        .and_then(|v| v.to_str().ok())
281        .map(std::string::ToString::to_string);
282      classify_graphql_content_type(content_type.as_deref())?;
283      let body = read_body_bytes(req).await?;
284      if body.is_empty() {
285        return Err(GraphQLError::Parse("empty request body".to_string()));
286      }
287      let reader = futures_util::io::Cursor::new(body.to_vec());
288      let batch = async_graphql::http::receive_batch_body(content_type.as_deref(), reader, opts)
289        .await
290        .map_err(|e| GraphQLError::Parse(e.to_string()))?;
291      Ok(GraphQLBatchRequest(batch))
292    }
293  }
294}
tako_rs_core/graphql/request.rs

tako_rs_core/graphql/
request.rs
