use crate::Request;
use crate::service::web::extract::FromRequest;
use crate::utils::macros::{composite_http_rejection, define_http_rejection};
use ahash::HashMap;
use rama_core::bytes::Bytes;
use rama_core::extensions::{Extension, ExtensionsRef};
use rama_core::futures::{Stream, TryStream};
use rama_http_types::{HeaderMap, StatusCode, header};
use rama_utils::macros::generate_set_and_with;
use std::borrow::Cow;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{Context, Poll};
#[derive(Debug, Clone, Default, Extension)]
#[extension(tags(http))]
pub struct MultipartConfig {
default_field_limit: Option<u64>,
field_limits: HashMap<Cow<'static, str>, u64>,
}
impl MultipartConfig {
#[must_use]
pub fn new() -> Self {
Self::default()
}
generate_set_and_with! {
pub fn default_field_limit(mut self, limit: Option<u64>) -> Self {
self.default_field_limit = limit;
self
}
}
#[must_use]
pub fn with_field_limit(mut self, name: impl Into<Cow<'static, str>>, limit: u64) -> Self {
self.field_limits.insert(name.into(), limit);
self
}
pub fn set_field_limit(&mut self, name: impl Into<Cow<'static, str>>, limit: u64) -> &mut Self {
self.field_limits.insert(name.into(), limit);
self
}
pub fn merge_floor(&mut self, other: &Self) {
if let Some(other_default) = other.default_field_limit {
self.default_field_limit = Some(match self.default_field_limit {
Some(cur) => cur.min(other_default),
None => other_default,
});
}
for (name, &other_limit) in &other.field_limits {
self.field_limits
.entry(name.clone())
.and_modify(|cur| *cur = (*cur).min(other_limit))
.or_insert(other_limit);
}
}
fn to_constraints(&self) -> multer::Constraints {
let mut limit = multer::SizeLimit::new();
if let Some(default) = self.default_field_limit {
limit = limit.per_field(default);
}
for (name, &n) in &self.field_limits {
limit = limit.for_field(name.as_ref().to_owned(), n);
}
multer::Constraints::new().size_limit(limit)
}
}
#[derive(Debug)]
pub struct Multipart {
inner: multer::Multipart<'static>,
}
impl Multipart {
pub fn from_body(body: crate::Body, boundary: impl Into<String>) -> Self {
let stream = body.into_data_stream();
let inner = multer::Multipart::new(stream, boundary);
Self { inner }
}
pub fn from_body_with_config(
body: crate::Body,
boundary: impl Into<String>,
config: Option<&MultipartConfig>,
) -> Self {
let stream = body.into_data_stream();
let inner = match config {
Some(c) => multer::Multipart::with_constraints(stream, boundary, c.to_constraints()),
None => multer::Multipart::new(stream, boundary),
};
Self { inner }
}
pub fn from_stream_with_config<S, O, E, B>(
stream: S,
boundary: B,
config: Option<&MultipartConfig>,
) -> Self
where
S: Stream<Item = Result<O, E>> + Send + 'static,
O: Into<Bytes> + 'static,
E: Into<Box<dyn std::error::Error + Send + Sync>> + 'static,
B: Into<String>,
{
let inner = match config {
Some(c) => multer::Multipart::with_constraints(stream, boundary, c.to_constraints()),
None => multer::Multipart::new(stream, boundary),
};
Self { inner }
}
pub async fn next_field(&mut self) -> Result<Option<Field<'_>>, MultipartError> {
match self.inner.next_field().await {
Ok(Some(inner)) => Ok(Some(Field {
inner,
_marker: PhantomData,
})),
Ok(None) => Ok(None),
Err(err) => Err(MultipartError::from(err)),
}
}
}
#[derive(Debug)]
pub struct Field<'a> {
inner: multer::Field<'static>,
_marker: PhantomData<&'a mut Multipart>,
}
impl Field<'_> {
#[must_use]
pub fn name(&self) -> Option<&str> {
self.inner.name()
}
#[must_use]
pub fn file_name(&self) -> Option<&str> {
self.inner.file_name()
}
#[must_use]
pub fn content_type(&self) -> Option<&crate::mime::Mime> {
self.inner.content_type()
}
pub fn headers(&self) -> Result<HeaderMap, rama_http_types::Error> {
rama_http_hyperium::TryIntoRamaHttp::try_into_rama_http(self.inner.headers())
}
#[must_use]
pub fn index(&self) -> usize {
self.inner.index()
}
pub async fn bytes(self) -> Result<Bytes, MultipartError> {
self.inner.bytes().await.map_err(MultipartError::from)
}
pub async fn text(self) -> Result<String, MultipartError> {
self.inner.text().await.map_err(MultipartError::from)
}
pub async fn chunk(&mut self) -> Result<Option<Bytes>, MultipartError> {
self.inner.chunk().await.map_err(MultipartError::from)
}
}
impl Stream for Field<'_> {
type Item = Result<Bytes, MultipartError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
let inner = Pin::new(&mut this.inner);
inner
.try_poll_next(cx)
.map(|opt| opt.map(|res| res.map_err(MultipartError::from)))
}
}
#[derive(Debug)]
pub struct MultipartError {
source: multer::Error,
}
impl MultipartError {
#[must_use]
pub fn status(&self) -> StatusCode {
match &self.source {
multer::Error::FieldSizeExceeded { .. } | multer::Error::StreamSizeExceeded { .. } => {
StatusCode::PAYLOAD_TOO_LARGE
}
multer::Error::StreamReadFailed(_) | multer::Error::LockFailure => {
StatusCode::INTERNAL_SERVER_ERROR
}
_ => StatusCode::BAD_REQUEST,
}
}
#[must_use]
pub fn body_text(&self) -> String {
format!(
"Failed to parse `multipart/form-data` request: {}",
self.source
)
}
}
impl From<multer::Error> for MultipartError {
fn from(source: multer::Error) -> Self {
Self { source }
}
}
impl std::fmt::Display for MultipartError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Error parsing `multipart/form-data` request")
}
}
impl std::error::Error for MultipartError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(&self.source)
}
}
impl crate::service::web::endpoint::IntoResponse for MultipartError {
fn into_response(self) -> crate::Response {
crate::utils::macros::log_http_rejection!(
rejection_type = MultipartError,
body_text = self.body_text(),
status = self.status(),
);
(self.status(), self.body_text()).into_response()
}
}
define_http_rejection! {
#[status = BAD_REQUEST]
#[body = "Multipart request is missing or has an invalid `Content-Type` boundary"]
pub struct InvalidMultipartBoundary;
}
define_http_rejection! {
#[status = UNSUPPORTED_MEDIA_TYPE]
#[body = "Multipart requests must have `Content-Type: multipart/form-data`"]
pub struct InvalidMultipartContentType;
}
composite_http_rejection! {
pub enum MultipartRejection {
InvalidMultipartContentType,
InvalidMultipartBoundary,
MultipartError,
}
}
impl FromRequest for Multipart {
type Rejection = MultipartRejection;
async fn from_request(req: Request) -> Result<Self, Self::Rejection> {
let content_type = req
.headers()
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.ok_or(InvalidMultipartContentType)?;
if !is_multipart_form_data(content_type) {
return Err(InvalidMultipartContentType.into());
}
let boundary =
multer::parse_boundary(content_type).map_err(|_e| InvalidMultipartBoundary)?;
let config = req.extensions().get_arc::<MultipartConfig>();
Ok(Self::from_body_with_config(
req.into_body(),
boundary,
config.as_deref(),
))
}
}
fn is_multipart_form_data(content_type: &str) -> bool {
content_type
.parse::<crate::mime::Mime>()
.ok()
.is_some_and(|m| {
m.type_() == crate::mime::MULTIPART && m.subtype() == crate::mime::FORM_DATA
})
}
#[cfg(test)]
mod test {
use super::*;
use crate::StatusCode;
use crate::service::web::WebService;
use rama_core::Service;
use rama_utils::octets::kib_u64;
const BOUNDARY: &str = "X-RAMA-TEST-BOUNDARY";
fn body_with(parts: &[(&str, Option<&str>, Option<&str>, &[u8])]) -> Vec<u8> {
let mut out = Vec::new();
for (name, file_name, content_type, data) in parts {
out.extend_from_slice(b"--");
out.extend_from_slice(BOUNDARY.as_bytes());
out.extend_from_slice(b"\r\nContent-Disposition: form-data; name=\"");
out.extend_from_slice(name.as_bytes());
out.extend_from_slice(b"\"");
if let Some(fname) = file_name {
out.extend_from_slice(b"; filename=\"");
out.extend_from_slice(fname.as_bytes());
out.extend_from_slice(b"\"");
}
out.extend_from_slice(b"\r\n");
if let Some(ct) = content_type {
out.extend_from_slice(b"Content-Type: ");
out.extend_from_slice(ct.as_bytes());
out.extend_from_slice(b"\r\n");
}
out.extend_from_slice(b"\r\n");
out.extend_from_slice(data);
out.extend_from_slice(b"\r\n");
}
out.extend_from_slice(b"--");
out.extend_from_slice(BOUNDARY.as_bytes());
out.extend_from_slice(b"--\r\n");
out
}
fn ct() -> String {
format!("multipart/form-data; boundary={BOUNDARY}")
}
#[tokio::test]
async fn test_multipart_text_and_file() {
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f1 = mp.next_field().await.unwrap().unwrap();
assert_eq!(f1.name(), Some("name"));
assert_eq!(f1.file_name(), None);
assert_eq!(f1.text().await.unwrap(), "glen");
let f2 = mp.next_field().await.unwrap().unwrap();
assert_eq!(f2.name(), Some("avatar"));
assert_eq!(f2.file_name(), Some("a.bin"));
assert_eq!(
f2.content_type().map(|m| m.essence_str()),
Some("application/octet-stream")
);
assert_eq!(f2.bytes().await.unwrap().as_ref(), b"\x00\x01\x02");
assert!(mp.next_field().await.unwrap().is_none());
StatusCode::OK
});
let body = body_with(&[
("name", None, None, b"glen"),
(
"avatar",
Some("a.bin"),
Some("application/octet-stream"),
&[0, 1, 2],
),
]);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_quoted_boundary_in_content_type() {
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.text().await.unwrap(), "v");
StatusCode::OK
});
let body = body_with(&[("k", None, None, b"v")]);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(
rama_http_types::header::CONTENT_TYPE,
format!("multipart/form-data; boundary=\"{BOUNDARY}\""),
)
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_with_preamble_and_epilogue() {
let mut body = Vec::new();
body.extend_from_slice(b"This is a preamble that must be ignored.\r\n");
body.extend_from_slice(&body_with(&[("k", None, None, b"v")]));
body.extend_from_slice(b"trailing epilogue ignored\r\n");
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.name(), Some("k"));
assert_eq!(f.text().await.unwrap(), "v");
StatusCode::OK
});
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_with_transport_padding() {
let mut body = Vec::new();
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b" \r\n"); body.extend_from_slice(b"Content-Disposition: form-data; name=\"k\"\r\n\r\n");
body.extend_from_slice(b"v\r\n");
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"--\t \r\n");
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.text().await.unwrap(), "v");
StatusCode::OK
});
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_filename_with_rfc5987_ext_value() {
let mut body = Vec::new();
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"\r\n");
body.extend_from_slice(
b"Content-Disposition: form-data; name=\"file\"; filename*=UTF-8''r%C3%A9sum%C3%A9.txt\r\n",
);
body.extend_from_slice(b"\r\n");
body.extend_from_slice(b"hello\r\n");
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"--\r\n");
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.name(), Some("file"));
assert_eq!(f.file_name(), None);
assert_eq!(f.text().await.unwrap(), "hello");
StatusCode::OK
});
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_filename_utf8_passthrough() {
let mut body = Vec::new();
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"\r\n");
body.extend_from_slice(
"Content-Disposition: form-data; name=\"file\"; filename=\"résumé.txt\"\r\n".as_bytes(),
);
body.extend_from_slice(b"\r\n");
body.extend_from_slice(b"hello\r\n");
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"--\r\n");
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.name(), Some("file"));
assert_eq!(f.file_name(), Some("résumé.txt"));
assert_eq!(f.text().await.unwrap(), "hello");
StatusCode::OK
});
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_filename_with_escaped_quote() {
let mut body = Vec::new();
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"\r\n");
body.extend_from_slice(
br#"Content-Disposition: form-data; name="file"; filename="we\"ird.txt""#,
);
body.extend_from_slice(b"\r\n\r\nhello\r\n");
body.extend_from_slice(b"--");
body.extend_from_slice(BOUNDARY.as_bytes());
body.extend_from_slice(b"--\r\n");
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f = mp.next_field().await.unwrap().unwrap();
assert_eq!(f.file_name(), Some("we\"ird.txt"));
assert_eq!(f.text().await.unwrap(), "hello");
StatusCode::OK
});
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_multipart_invalid_boundary() {
let service = WebService::default().with_post("/", async |_: Multipart| StatusCode::OK);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, "multipart/form-data")
.body("ignored".into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
}
#[tokio::test]
async fn test_multipart_rejects_non_form_data_subtype() {
let service = WebService::default().with_post("/", async |_: Multipart| StatusCode::OK);
let body = body_with(&[("k", None, None, b"v")]);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(
rama_http_types::header::CONTENT_TYPE,
format!("multipart/mixed; boundary={BOUNDARY}"),
)
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
}
#[tokio::test]
async fn test_multipart_rejects_missing_content_type() {
let service = WebService::default().with_post("/", async |_: Multipart| StatusCode::OK);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.body("ignored".into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
}
#[tokio::test]
async fn test_multipart_field_limit_exceeded() {
let service = WebService::default().with_post(
"/",
async |mut mp: Multipart| -> Result<StatusCode, MultipartRejection> {
let f = mp.next_field().await?.unwrap();
_ = f.bytes().await?;
Ok(StatusCode::OK)
},
);
let body = body_with(&[("blob", None, None, &[42u8; 64])]);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
req.extensions()
.insert(MultipartConfig::new().with_default_field_limit(8));
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
}
#[tokio::test]
async fn test_multipart_field_as_stream_chunks() {
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let mut field = mp.next_field().await.unwrap().unwrap();
let mut total = Vec::new();
while let Some(chunk) = field.chunk().await.unwrap() {
total.extend_from_slice(&chunk);
}
assert_eq!(total, b"hello world");
StatusCode::OK
});
let body = body_with(&[("payload", None, None, b"hello world")]);
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, ct())
.body(body.into())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[tokio::test]
async fn test_roundtrip_with_client_form() {
use crate::service::client::multipart as client;
let service =
WebService::default().with_post("/", async |mut mp: Multipart| -> StatusCode {
let f1 = mp.next_field().await.unwrap().unwrap();
assert_eq!(f1.name(), Some("greeting"));
assert_eq!(f1.text().await.unwrap(), "hello world");
let f2 = mp.next_field().await.unwrap().unwrap();
assert_eq!(f2.name(), Some("payload"));
assert_eq!(f2.file_name(), Some("blob.bin"));
assert_eq!(
f2.content_type().map(|m| m.essence_str()),
Some("application/octet-stream")
);
assert_eq!(f2.bytes().await.unwrap().as_ref(), &[7u8, 8, 9, 10]);
assert!(mp.next_field().await.unwrap().is_none());
StatusCode::OK
});
let form = client::Form::new().text("greeting", "hello world").part(
"payload",
client::Part::bytes([7u8, 8, 9, 10].as_slice())
.with_file_name("blob.bin")
.with_mime(crate::mime::APPLICATION_OCTET_STREAM),
);
let content_type = form.content_type();
let req = rama_http_types::Request::builder()
.method(rama_http_types::Method::POST)
.header(rama_http_types::header::CONTENT_TYPE, content_type)
.body(form.into_body())
.unwrap();
let resp = service.serve(req).await.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
}
#[test]
fn test_config_merge_floor_keeps_min() {
let mut a = MultipartConfig::new()
.with_default_field_limit(1024)
.with_field_limit("avatar", kib_u64(8));
let b = MultipartConfig::new()
.with_default_field_limit(2048)
.with_field_limit("avatar", kib_u64(4))
.with_field_limit("doc", 100);
a.merge_floor(&b);
assert_eq!(a.default_field_limit, Some(1024));
assert_eq!(a.field_limits.get("avatar").copied(), Some(kib_u64(4)));
assert_eq!(a.field_limits.get("doc").copied(), Some(100));
}
}