use crate::server::axum::response::ApiError;
use axum::body::Body;
use axum::http::Request;
use axum::http::header::CONTENT_LENGTH;
use axum::response::{IntoResponse, Response};
use futures::future::BoxFuture;
use std::task::{Context, Poll};
use tower::{Layer, Service};
#[doc(inline)]
pub use tower_http::limit::RequestBodyLimitLayer;
#[derive(Clone, Debug)]
pub struct BodyLimiterConfig {
pub body_max_size: usize,
}
#[derive(Clone)]
pub struct BodyLimiterLayer {
pub config: BodyLimiterConfig,
}
impl BodyLimiterLayer {
pub fn new(config: &BodyLimiterConfig) -> Self {
Self { config: config.clone() }
}
}
impl<S> Layer<S> for BodyLimiterLayer {
type Service = BodyLimiterMiddleware<S>;
fn layer(&self, inner: S) -> Self::Service {
BodyLimiterMiddleware {
inner,
config: self.config.clone(),
}
}
}
#[derive(Clone)]
pub struct BodyLimiterMiddleware<S> {
inner: S,
config: BodyLimiterConfig,
}
impl<S> Service<Request<Body>> for BodyLimiterMiddleware<S>
where
S: Service<Request<Body>, Response = Response> + Send + 'static,
S::Future: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, request: Request<Body>) -> Self::Future {
let content_length = request
.headers()
.get(CONTENT_LENGTH)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<usize>().ok());
if let Some(length) = content_length
&& length > self.config.body_max_size
{
return Box::pin(async move { Ok(ApiError::PayloadTooLarge.into_response()) });
}
Box::pin(self.inner.call(request))
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::StatusCode;
use std::convert::Infallible;
use tower::{ServiceBuilder, ServiceExt};
fn layer() -> BodyLimiterLayer {
BodyLimiterLayer::new(&BodyLimiterConfig { body_max_size: 8 })
}
fn ok_service() -> impl Service<Request<Body>, Response = Response, Error = Infallible> + Clone {
ServiceBuilder::new()
.layer(layer())
.service(tower::service_fn(|_req: Request<Body>| async {
Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::OK)
.body(Body::from("ok"))
.unwrap(),
)
}))
}
async fn read_body(response: Response) -> String {
let body = axum::body::to_bytes(response.into_body(), 4096).await.unwrap();
String::from_utf8(body.to_vec()).unwrap()
}
#[tokio::test]
async fn content_length_over_limit_is_rejected_with_413() {
let svc = ServiceBuilder::new()
.layer(layer())
.service(tower::service_fn(|_req: Request<Body>| async {
Ok::<_, Infallible>(
Response::builder()
.status(StatusCode::OK)
.body(Body::from("ok"))
.unwrap(),
)
}));
let response = svc
.oneshot(
Request::builder()
.uri("/")
.header(CONTENT_LENGTH, "9")
.body(Body::from("123456789"))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
let body = read_body(response).await;
assert!(body.contains("\"code\":413"), "body was: {body}");
}
#[tokio::test]
async fn content_length_at_limit_passes_through() {
let response = ok_service()
.oneshot(
Request::builder()
.uri("/")
.header(CONTENT_LENGTH, "8")
.body(Body::from("12345678"))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(read_body(response).await, "ok");
}
#[tokio::test]
async fn content_length_under_limit_passes_through() {
let response = ok_service()
.oneshot(
Request::builder()
.uri("/")
.header(CONTENT_LENGTH, "3")
.body(Body::from("abc"))
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(read_body(response).await, "ok");
}
#[tokio::test]
async fn missing_content_length_passes_through() {
let response = ok_service()
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(read_body(response).await, "ok");
}
}