1use std::{
4 borrow::Cow,
5 future::Future,
6 pin::Pin,
7 str,
8 task::{Context, Poll},
9};
10
11use actix_http::error::PayloadError;
12use actix_utils::future::{ready, Either, Ready};
13use bytes::{Bytes, BytesMut};
14use encoding_rs::{Encoding, UTF_8};
15use futures_core::{ready, stream::Stream};
16use mime::Mime;
17
18use crate::{
19 body, dev, error::ErrorBadRequest, http::header, web, Error, FromRequest, HttpMessage,
20 HttpRequest,
21};
22
23pub struct Payload(dev::Payload);
47
48impl Payload {
49 #[inline]
51 pub fn into_inner(self) -> dev::Payload {
52 self.0
53 }
54
55 pub async fn to_bytes_limited(
83 self,
84 limit: usize,
85 ) -> Result<crate::Result<Bytes>, body::BodyLimitExceeded> {
86 let stream = body::BodyStream::new(self.0);
87
88 match body::to_bytes_limited(stream, limit).await {
89 Ok(Ok(body)) => Ok(Ok(body)),
90 Ok(Err(err)) => Ok(Err(err.into())),
91 Err(err) => Err(err),
92 }
93 }
94
95 pub async fn to_bytes(self) -> crate::Result<Bytes> {
117 let stream = body::BodyStream::new(self.0);
118 Ok(body::to_bytes(stream).await?)
119 }
120}
121
122impl Stream for Payload {
123 type Item = Result<Bytes, PayloadError>;
124
125 #[inline]
126 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
127 Pin::new(&mut self.0).poll_next(cx)
128 }
129}
130
131impl FromRequest for Payload {
133 type Error = Error;
134 type Future = Ready<Result<Self, Self::Error>>;
135
136 #[inline]
137 fn from_request(_: &HttpRequest, payload: &mut dev::Payload) -> Self::Future {
138 ready(Ok(Payload(payload.take())))
139 }
140}
141
142impl FromRequest for Bytes {
159 type Error = Error;
160 type Future = Either<BytesExtractFut, Ready<Result<Bytes, Error>>>;
161
162 #[inline]
163 fn from_request(req: &HttpRequest, payload: &mut dev::Payload) -> Self::Future {
164 let cfg = PayloadConfig::from_req(req);
166
167 if let Err(err) = cfg.check_mimetype(req) {
168 return Either::right(ready(Err(err)));
169 }
170
171 Either::left(BytesExtractFut {
172 body_fut: HttpMessageBody::new(req, payload).limit(cfg.limit),
173 })
174 }
175}
176
177pub struct BytesExtractFut {
179 body_fut: HttpMessageBody,
180}
181
182impl Future for BytesExtractFut {
183 type Output = Result<Bytes, Error>;
184
185 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
186 Pin::new(&mut self.body_fut).poll(cx).map_err(Into::into)
187 }
188}
189
190impl FromRequest for String {
206 type Error = Error;
207 type Future = Either<StringExtractFut, Ready<Result<String, Error>>>;
208
209 #[inline]
210 fn from_request(req: &HttpRequest, payload: &mut dev::Payload) -> Self::Future {
211 let cfg = PayloadConfig::from_req(req);
212
213 if let Err(err) = cfg.check_mimetype(req) {
215 return Either::right(ready(Err(err)));
216 }
217
218 let encoding = match req.encoding() {
220 Ok(enc) => enc,
221 Err(err) => return Either::right(ready(Err(err.into()))),
222 };
223 let limit = cfg.limit;
224 let body_fut = HttpMessageBody::new(req, payload).limit(limit);
225
226 Either::left(StringExtractFut { body_fut, encoding })
227 }
228}
229
230pub struct StringExtractFut {
232 body_fut: HttpMessageBody,
233 encoding: &'static Encoding,
234}
235
236impl Future for StringExtractFut {
237 type Output = Result<String, Error>;
238
239 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
240 let encoding = self.encoding;
241
242 Pin::new(&mut self.body_fut).poll(cx).map(|out| {
243 let body = out?;
244 bytes_to_string(body, encoding)
245 })
246 }
247}
248
249fn bytes_to_string(body: Bytes, encoding: &'static Encoding) -> Result<String, Error> {
250 if encoding == UTF_8 {
251 Ok(str::from_utf8(body.as_ref())
252 .map_err(|_| ErrorBadRequest("Can not decode body"))?
253 .to_owned())
254 } else {
255 Ok(encoding
256 .decode_without_bom_handling_and_without_replacement(&body)
257 .map(Cow::into_owned)
258 .ok_or_else(|| ErrorBadRequest("Can not decode body"))?)
259 }
260}
261
262#[derive(Clone)]
274pub struct PayloadConfig {
275 limit: usize,
276 mimetype: Option<Mime>,
277}
278
279impl PayloadConfig {
280 pub fn new(limit: usize) -> Self {
282 Self {
283 limit,
284 ..Default::default()
285 }
286 }
287
288 pub fn limit(mut self, limit: usize) -> Self {
290 self.limit = limit;
291 self
292 }
293
294 pub fn mimetype(mut self, mt: Mime) -> Self {
296 self.mimetype = Some(mt);
297 self
298 }
299
300 fn check_mimetype(&self, req: &HttpRequest) -> Result<(), Error> {
301 if let Some(ref mt) = self.mimetype {
303 match req.mime_type() {
304 Ok(Some(ref req_mt)) => {
305 if mt != req_mt {
306 return Err(ErrorBadRequest("Unexpected Content-Type"));
307 }
308 }
309 Ok(None) => {
310 return Err(ErrorBadRequest("Content-Type is expected"));
311 }
312 Err(err) => {
313 return Err(err.into());
314 }
315 }
316 }
317
318 Ok(())
319 }
320
321 fn from_req(req: &HttpRequest) -> &Self {
324 req.app_data::<Self>()
325 .or_else(|| req.app_data::<web::Data<Self>>().map(|d| d.as_ref()))
326 .unwrap_or(&DEFAULT_CONFIG)
327 }
328}
329
330const DEFAULT_CONFIG_LIMIT: usize = 262_144; const DEFAULT_CONFIG: PayloadConfig = PayloadConfig {
334 limit: DEFAULT_CONFIG_LIMIT,
335 mimetype: None,
336};
337
338impl Default for PayloadConfig {
339 fn default() -> Self {
340 DEFAULT_CONFIG
341 }
342}
343
344pub struct HttpMessageBody {
349 limit: usize,
350 length: Option<usize>,
351 #[cfg(feature = "__compress")]
352 stream: dev::Decompress<dev::Payload>,
353 #[cfg(not(feature = "__compress"))]
354 stream: dev::Payload,
355 buf: BytesMut,
356 err: Option<PayloadError>,
357}
358
359impl HttpMessageBody {
360 #[allow(clippy::borrow_interior_mutable_const)]
362 pub fn new(req: &HttpRequest, payload: &mut dev::Payload) -> HttpMessageBody {
363 let mut length = None;
364 let mut err = None;
365
366 if let Some(l) = req.headers().get(&header::CONTENT_LENGTH) {
367 match l.to_str() {
368 Ok(s) => match s.parse::<usize>() {
369 Ok(l) => {
370 if l > DEFAULT_CONFIG_LIMIT {
371 err = Some(PayloadError::Overflow);
372 }
373 length = Some(l)
374 }
375 Err(_) => err = Some(PayloadError::UnknownLength),
376 },
377 Err(_) => err = Some(PayloadError::UnknownLength),
378 }
379 }
380
381 let stream = {
382 cfg_if::cfg_if! {
383 if #[cfg(feature = "__compress")] {
384 dev::Decompress::from_headers(payload.take(), req.headers())
385 } else {
386 payload.take()
387 }
388 }
389 };
390
391 HttpMessageBody {
392 stream,
393 limit: DEFAULT_CONFIG_LIMIT,
394 length,
395 buf: BytesMut::with_capacity(8192),
396 err,
397 }
398 }
399
400 pub fn limit(mut self, limit: usize) -> Self {
402 if let Some(l) = self.length {
403 self.err = if l > limit {
404 Some(PayloadError::Overflow)
405 } else {
406 None
407 };
408 }
409 self.limit = limit;
410 self
411 }
412}
413
414impl Future for HttpMessageBody {
415 type Output = Result<Bytes, PayloadError>;
416
417 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
418 let this = self.get_mut();
419
420 if let Some(err) = this.err.take() {
421 return Poll::Ready(Err(err));
422 }
423
424 loop {
425 let res = ready!(Pin::new(&mut this.stream).poll_next(cx));
426 match res {
427 Some(chunk) => {
428 let chunk = chunk?;
429 if this.buf.len() + chunk.len() > this.limit {
430 return Poll::Ready(Err(PayloadError::Overflow));
431 } else {
432 this.buf.extend_from_slice(&chunk);
433 }
434 }
435 None => return Poll::Ready(Ok(this.buf.split().freeze())),
436 }
437 }
438 }
439}
440
441#[cfg(test)]
442mod tests {
443 use super::*;
444 use crate::{
445 http::StatusCode,
446 test::{call_service, init_service, read_body, TestRequest},
447 App, Responder,
448 };
449
450 #[actix_rt::test]
451 async fn payload_to_bytes() {
452 async fn payload_handler(pl: Payload) -> crate::Result<impl Responder> {
453 pl.to_bytes().await
454 }
455
456 async fn limited_payload_handler(pl: Payload) -> crate::Result<impl Responder> {
457 match pl.to_bytes_limited(5).await {
458 Ok(res) => res,
459 Err(_limited) => Err(ErrorBadRequest("too big")),
460 }
461 }
462
463 let srv = init_service(
464 App::new()
465 .route("/all", web::to(payload_handler))
466 .route("limited", web::to(limited_payload_handler)),
467 )
468 .await;
469
470 let req = TestRequest::with_uri("/all")
471 .set_payload("1234567890")
472 .to_request();
473 let res = call_service(&srv, req).await;
474 assert_eq!(res.status(), StatusCode::OK);
475 let body = read_body(res).await;
476 assert_eq!(body, "1234567890");
477
478 let req = TestRequest::with_uri("/limited")
479 .set_payload("1234567890")
480 .to_request();
481 let res = call_service(&srv, req).await;
482 assert_eq!(res.status(), StatusCode::BAD_REQUEST);
483
484 let req = TestRequest::with_uri("/limited")
485 .set_payload("12345")
486 .to_request();
487 let res = call_service(&srv, req).await;
488 assert_eq!(res.status(), StatusCode::OK);
489 let body = read_body(res).await;
490 assert_eq!(body, "12345");
491 }
492
493 #[actix_rt::test]
494 async fn test_payload_config() {
495 let req = TestRequest::default().to_http_request();
496 let cfg = PayloadConfig::default().mimetype(mime::APPLICATION_JSON);
497 assert!(cfg.check_mimetype(&req).is_err());
498
499 let req = TestRequest::default()
500 .insert_header((header::CONTENT_TYPE, "application/x-www-form-urlencoded"))
501 .to_http_request();
502 assert!(cfg.check_mimetype(&req).is_err());
503
504 let req = TestRequest::default()
505 .insert_header((header::CONTENT_TYPE, "application/json"))
506 .to_http_request();
507 assert!(cfg.check_mimetype(&req).is_ok());
508 }
509
510 #[allow(deprecated)]
512 #[actix_rt::test]
513 async fn test_config_recall_locations() {
514 async fn bytes_handler(_: Bytes) -> impl Responder {
515 "payload is probably json bytes"
516 }
517
518 async fn string_handler(_: String) -> impl Responder {
519 "payload is probably json string"
520 }
521
522 let srv = init_service(
523 App::new()
524 .service(
525 web::resource("/bytes-app-data")
526 .app_data(PayloadConfig::default().mimetype(mime::APPLICATION_JSON))
527 .route(web::get().to(bytes_handler)),
528 )
529 .service(
530 web::resource("/bytes-data")
531 .data(PayloadConfig::default().mimetype(mime::APPLICATION_JSON))
532 .route(web::get().to(bytes_handler)),
533 )
534 .service(
535 web::resource("/string-app-data")
536 .app_data(PayloadConfig::default().mimetype(mime::APPLICATION_JSON))
537 .route(web::get().to(string_handler)),
538 )
539 .service(
540 web::resource("/string-data")
541 .data(PayloadConfig::default().mimetype(mime::APPLICATION_JSON))
542 .route(web::get().to(string_handler)),
543 ),
544 )
545 .await;
546
547 let req = TestRequest::with_uri("/bytes-app-data").to_request();
548 let resp = call_service(&srv, req).await;
549 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
550
551 let req = TestRequest::with_uri("/bytes-data").to_request();
552 let resp = call_service(&srv, req).await;
553 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
554
555 let req = TestRequest::with_uri("/string-app-data").to_request();
556 let resp = call_service(&srv, req).await;
557 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
558
559 let req = TestRequest::with_uri("/string-data").to_request();
560 let resp = call_service(&srv, req).await;
561 assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
562
563 let req = TestRequest::with_uri("/bytes-app-data")
564 .insert_header(header::ContentType(mime::APPLICATION_JSON))
565 .to_request();
566 let resp = call_service(&srv, req).await;
567 assert_eq!(resp.status(), StatusCode::OK);
568
569 let req = TestRequest::with_uri("/bytes-data")
570 .insert_header(header::ContentType(mime::APPLICATION_JSON))
571 .to_request();
572 let resp = call_service(&srv, req).await;
573 assert_eq!(resp.status(), StatusCode::OK);
574
575 let req = TestRequest::with_uri("/string-app-data")
576 .insert_header(header::ContentType(mime::APPLICATION_JSON))
577 .to_request();
578 let resp = call_service(&srv, req).await;
579 assert_eq!(resp.status(), StatusCode::OK);
580
581 let req = TestRequest::with_uri("/string-data")
582 .insert_header(header::ContentType(mime::APPLICATION_JSON))
583 .to_request();
584 let resp = call_service(&srv, req).await;
585 assert_eq!(resp.status(), StatusCode::OK);
586 }
587
588 #[actix_rt::test]
589 async fn test_bytes() {
590 let (req, mut pl) = TestRequest::default()
591 .insert_header((header::CONTENT_LENGTH, "11"))
592 .set_payload(Bytes::from_static(b"hello=world"))
593 .to_http_parts();
594
595 let s = Bytes::from_request(&req, &mut pl).await.unwrap();
596 assert_eq!(s, Bytes::from_static(b"hello=world"));
597 }
598
599 #[actix_rt::test]
600 async fn test_string() {
601 let (req, mut pl) = TestRequest::default()
602 .insert_header((header::CONTENT_LENGTH, "11"))
603 .set_payload(Bytes::from_static(b"hello=world"))
604 .to_http_parts();
605
606 let s = String::from_request(&req, &mut pl).await.unwrap();
607 assert_eq!(s, "hello=world");
608 }
609
610 #[actix_rt::test]
611 async fn test_message_body() {
612 let (req, mut pl) = TestRequest::default()
613 .insert_header((header::CONTENT_LENGTH, "xxxx"))
614 .to_srv_request()
615 .into_parts();
616 let res = HttpMessageBody::new(&req, &mut pl).await;
617 match res.err().unwrap() {
618 PayloadError::UnknownLength => {}
619 _ => unreachable!("error"),
620 }
621
622 let (req, mut pl) = TestRequest::default()
623 .insert_header((header::CONTENT_LENGTH, "1000000"))
624 .to_srv_request()
625 .into_parts();
626 let res = HttpMessageBody::new(&req, &mut pl).await;
627 match res.err().unwrap() {
628 PayloadError::Overflow => {}
629 _ => unreachable!("error"),
630 }
631
632 let (req, mut pl) = TestRequest::default()
633 .set_payload(Bytes::from_static(b"test"))
634 .to_http_parts();
635 let res = HttpMessageBody::new(&req, &mut pl).await;
636 assert_eq!(res.ok().unwrap(), Bytes::from_static(b"test"));
637
638 let (req, mut pl) = TestRequest::default()
639 .set_payload(Bytes::from_static(b"11111111111111"))
640 .to_http_parts();
641 let res = HttpMessageBody::new(&req, &mut pl).limit(5).await;
642 match res.err().unwrap() {
643 PayloadError::Overflow => {}
644 _ => unreachable!("error"),
645 }
646 }
647}