taler-rust

extract.rs (7079B)

---

 /*
   This file is part of TALER
   Copyright (C) 2025, 2026 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU Affero General Public License as published by the Free Software
   Foundation; either version 3, or (at your option) any later version.
 
   TALER is distributed in the hope that it will be useful, but WITHOUT ANY
   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
   A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public License along with
   TALER; see the file COPYING.  If not, see <http://www.gnu.org/licenses/>
 */
 
 use axum::{
     body::{Body, Bytes, HttpBody},
     extract::{FromRequest, FromRequestParts, Request},
     http::{
         HeaderMap, StatusCode,
         header::{self, CONTENT_TYPE},
         request::Parts,
     },
 };
 use http_body_util::BodyExt as _;
 use serde::de::DeserializeOwned;
 use taler_common::error_code::ErrorCode;
 use tracing::trace;
 use url::form_urlencoded;
 use zlib_rs::{InflateConfig, ReturnCode};
 
 use crate::{
     constants::MAX_BODY_LENGTH,
     error::{ApiError, ApiResult, failure, failure_status},
 };
 
 pub async fn decompressed_strict_body(headers: &HeaderMap, body: Body) -> ApiResult<Bytes> {
     // Check content type
     match headers.get(header::CONTENT_TYPE) {
         Some(header) => {
             if header.as_bytes() != b"application/json" {
                 return Err(failure_status(
                     ErrorCode::GENERIC_HTTP_HEADERS_MALFORMED,
                     "Bad Content-Type header",
                     StatusCode::UNSUPPORTED_MEDIA_TYPE,
                 ));
             }
         }
         None => {
             return Err(failure_status(
                 ErrorCode::GENERIC_HTTP_HEADERS_MALFORMED,
                 "Missing Content-Type header",
                 StatusCode::UNSUPPORTED_MEDIA_TYPE,
             ));
         }
     }
 
     // Check content length if present and well formed
     if let Some(length) = headers
         .get(header::CONTENT_LENGTH)
         .and_then(|it| it.to_str().ok())
         .and_then(|it| it.parse::<usize>().ok())
         && length > MAX_BODY_LENGTH
     {
         return Err(failure(
             ErrorCode::GENERIC_UPLOAD_EXCEEDS_LIMIT,
             format!("Body is suspiciously big > {MAX_BODY_LENGTH}B"),
         ));
     }
 
     // Check compression
     let compressed = if let Some(encoding) = headers.get(header::CONTENT_ENCODING) {
         if encoding == "deflate" {
             true
         } else {
             return Err(failure_status(
                 ErrorCode::GENERIC_HTTP_HEADERS_MALFORMED,
                 format!(
                     "Unsupported encoding '{}'",
                     String::from_utf8_lossy(encoding.as_bytes())
                 ),
                 StatusCode::UNSUPPORTED_MEDIA_TYPE,
             ));
         }
     } else {
         false
     };
 
     // Buffer body
     let body = http_body_util::Limited::new(body, MAX_BODY_LENGTH);
     let bytes = match body.collect().await {
         Ok(chunks) => chunks.to_bytes(),
         Err(it) => match it.downcast::<http_body_util::LengthLimitError>() {
             Ok(_) => {
                 return Err(failure(
                     ErrorCode::GENERIC_UPLOAD_EXCEEDS_LIMIT,
                     format!("Body is suspiciously big > {MAX_BODY_LENGTH}B"),
                 ));
             }
             Err(err) => {
                 return Err(failure(
                     ErrorCode::GENERIC_UNEXPECTED_REQUEST_ERROR,
                     format!("Failed to read body: {err}"),
                 ));
             }
         },
     };
 
     let bytes = if compressed {
         let mut buf = [0; MAX_BODY_LENGTH];
         let (decompressed, code) =
             zlib_rs::decompress_slice(&mut buf, &bytes, InflateConfig::default());
         match code {
             ReturnCode::Ok => Bytes::copy_from_slice(decompressed),
             ReturnCode::BufError => {
                 return Err(failure(
                     ErrorCode::GENERIC_UPLOAD_EXCEEDS_LIMIT,
                     format!("Decompressed body is suspiciously big > {MAX_BODY_LENGTH}B"),
                 ));
             }
             _ => {
                 return Err(failure(
                     ErrorCode::GENERIC_COMPRESSION_INVALID,
                     "Failed to decompress body: invalid compression",
                 ));
             }
         }
     } else {
         bytes
     };
     trace!(target: "api", "req {}", String::from_utf8_lossy(&bytes));
     Ok(bytes)
 }
 
 #[derive(Debug, Clone, Copy, Default)]
 #[must_use]
 pub struct Req<T>(pub T);
 
 impl<T, S> FromRequest<S> for Req<T>
 where
     T: DeserializeOwned,
     S: Send + Sync,
 {
     type Rejection = ApiError;
 
     async fn from_request(req: Request, _state: &S) -> Result<Self, Self::Rejection> {
         let (parts, body) = req.into_parts();
         let bytes = decompressed_strict_body(&parts.headers, body).await?;
 
         Self::try_from(&bytes)
     }
 }
 
 impl<T: DeserializeOwned> TryFrom<&Bytes> for Req<T> {
     type Error = ApiError;
 
     fn try_from(value: &Bytes) -> Result<Self, Self::Error> {
         let mut de = serde_json::de::Deserializer::from_slice(value);
         let parsed = serde_path_to_error::deserialize(&mut de)?;
         de.end()
             .map_err(|err| failure(ErrorCode::GENERIC_JSON_INVALID, err))?;
         Ok(Req(parsed))
     }
 }
 
 #[derive(Debug, Clone, Copy, Default)]
 #[must_use]
 pub struct OptReq<T>(pub Option<T>);
 
 impl<T, S> FromRequest<S> for OptReq<T>
 where
     T: DeserializeOwned,
     S: Send + Sync,
 {
     type Rejection = ApiError;
 
     async fn from_request(req: Request, _state: &S) -> Result<Self, Self::Rejection> {
         let (parts, body) = req.into_parts();
         if !parts.headers.contains_key(CONTENT_TYPE) && body.size_hint().exact() == Some(0) {
             Ok(Self(None))
         } else {
             let bytes = decompressed_strict_body(&parts.headers, body).await?;
             let req = Req::try_from(&bytes)?;
             Ok(Self(Some(req.0)))
         }
     }
 }
 
 #[derive(Debug, Clone, Copy, Default)]
 pub struct Path<T: DeserializeOwned + Send>(pub T);
 
 impl<T: serde::de::DeserializeOwned + Send, S: Sync + Send> FromRequestParts<S> for Path<T> {
     type Rejection = ApiError;
 
     async fn from_request_parts(parts: &mut Parts, _: &S) -> Result<Self, Self::Rejection> {
         Ok(Self(
             axum::extract::Path::from_request_parts(parts, &()).await?.0,
         ))
     }
 }
 
 #[derive(Debug, Clone, Copy, Default)]
 pub struct Query<T>(pub T);
 
 impl<T: serde::de::DeserializeOwned, S: Sync + Send> FromRequestParts<S> for Query<T> {
     type Rejection = ApiError;
 
     async fn from_request_parts(parts: &mut Parts, _: &S) -> Result<Self, Self::Rejection> {
         let query = parts.uri.query().unwrap_or_default();
         let deserializer =
             serde_urlencoded::Deserializer::new(form_urlencoded::parse(query.as_bytes()));
         let params = serde_path_to_error::deserialize(deserializer)?;
         Ok(Query(params))
     }
 }