From 339ac8845362d0e4ea57059fdcd3ed706663a47e Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 09:58:54 +0100 Subject: [PATCH 1/6] fix(server): Consume request streams after reading them partially --- server/src/body/store_body.rs | 35 ++++++++++++++++++++++------------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/server/src/body/store_body.rs b/server/src/body/store_body.rs index 801eeedd8f3..cc3c21a40e0 100644 --- a/server/src/body/store_body.rs +++ b/server/src/body/store_body.rs @@ -148,24 +148,33 @@ impl Future for StoreBody { } let limit = self.limit; + let body = Some(BytesMut::with_capacity(8192)); + let future = self .stream .take() .expect("Can not be used second time") - .from_err() - .fold(BytesMut::with_capacity(8192), move |mut body, chunk| { - if (body.len() + chunk.len()) > limit { - Err(StorePayloadError::Overflow) - } else { - body.extend_from_slice(&chunk); - Ok(body) - } + .map_err(StorePayloadError::from) + .fold(body, move |body_opt, chunk| { + // Ensure that the stream is always fully consumed. Erroring here would leave a + // broken TCP stream that cannot be used with keep-alive connections. + Ok::<_, StorePayloadError>(body_opt.and_then(|mut body| { + if (body.len() + chunk.len()) > limit { + None + } else { + body.extend_from_slice(&chunk); + Some(body) + } + })) }) - .and_then(|body| { - metric!(time_raw("event.size_bytes.raw") = body.len() as u64); - let decoded = decode_bytes(body.freeze())?; - metric!(time_raw("event.size_bytes.uncompressed") = decoded.len() as u64); - Ok(decoded) + .and_then(|body_opt| match body_opt { + Some(body) => { + metric!(time_raw("event.size_bytes.raw") = body.len() as u64); + let decoded = decode_bytes(body.freeze())?; + metric!(time_raw("event.size_bytes.uncompressed") = decoded.len() as u64); + Ok(decoded) + } + None => Err(StorePayloadError::Overflow), }); self.fut = Some(Box::new(future)); From 1d01c8f8afc7e5ca6eef67b920542b0a9fa14126 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 10:10:55 +0100 Subject: [PATCH 2/6] fix(server): Drain the stream on content-length overflow --- server/src/body/store_body.rs | 55 +++++++++++-------------- server/src/endpoints/security_report.rs | 3 +- server/src/endpoints/store.rs | 3 +- 3 files changed, 25 insertions(+), 36 deletions(-) diff --git a/server/src/body/store_body.rs b/server/src/body/store_body.rs index cc3c21a40e0..5c20938eb29 100644 --- a/server/src/body/store_body.rs +++ b/server/src/body/store_body.rs @@ -75,10 +75,8 @@ impl From for StorePayloadError { /// Future that resolves to a complete store endpoint body. pub struct StoreBody { limit: usize, - length: Option, - // These states are mutually exclusive, and only separate options due to borrowing - // problems: + // These states are mutually exclusive: result: Option>, fut: Option>, stream: Option<::Stream>, @@ -86,44 +84,43 @@ pub struct StoreBody { impl StoreBody { /// Create `StoreBody` for request. - pub fn new(req: &HttpRequest) -> Self { + pub fn new(req: &HttpRequest, limit: usize) -> Self { if let Some(body) = data_from_querystring(req) { - StoreBody { - limit: 262_144, - length: Some(body.len()), + return StoreBody { + limit, stream: None, result: Some(decode_bytes(body.as_bytes())), fut: None, - } - } else { - let length = match get_content_length(req) { - Ok(x) => x, - Err(e) => return StoreBody::err(e), }; + } - StoreBody { - limit: 262_144, - length, - result: None, - stream: Some(req.payload()), - fut: None, + let length_opt = match get_content_length(req) { + Ok(length_opt) => length_opt, + Err(e) => return Self::err(e), + }; + + // Check the content length first. If we detect an overflow from the content length header, + // keep the payload in the request. The `ReadRequestMiddleware` should drain the payload. + if let Some(length) = length_opt { + if length > limit { + return Self::err(StorePayloadError::Overflow); } } - } - /// Change max size of payload. By default max size is 256Kb - pub fn limit(mut self, limit: usize) -> Self { - self.limit = limit; - self + StoreBody { + limit, + result: None, + fut: None, + stream: Some(req.payload()), + } } fn err(e: StorePayloadError) -> Self { StoreBody { - stream: None, - limit: 262_144, + limit: 0, result: Some(Err(e)), fut: None, - length: None, + stream: None, } } } @@ -141,12 +138,6 @@ impl Future for StoreBody { return fut.poll(); } - if let Some(len) = self.length.take() { - if len > self.limit { - return Err(StorePayloadError::Overflow); - } - } - let limit = self.limit; let body = Some(BytesMut::with_capacity(8192)); diff --git a/server/src/endpoints/security_report.rs b/server/src/endpoints/security_report.rs index a6ef0bb35c1..8ecf76f5f28 100644 --- a/server/src/endpoints/security_report.rs +++ b/server/src/endpoints/security_report.rs @@ -26,8 +26,7 @@ fn extract_envelope( max_event_payload_size: usize, params: SecurityReportParams, ) -> ResponseFuture { - let future = StoreBody::new(&request) - .limit(max_event_payload_size) + let future = StoreBody::new(&request, max_event_payload_size) .map_err(BadStoreRequest::PayloadError) .and_then(move |data| { if data.is_empty() { diff --git a/server/src/endpoints/store.rs b/server/src/endpoints/store.rs index f8bc0c83d66..fe221776f71 100644 --- a/server/src/endpoints/store.rs +++ b/server/src/endpoints/store.rs @@ -33,8 +33,7 @@ fn extract_envelope( max_event_payload_size: usize, content_type: String, ) -> ResponseFuture { - let future = StoreBody::new(&request) - .limit(max_event_payload_size) + let future = StoreBody::new(&request, max_event_payload_size) .map_err(BadStoreRequest::PayloadError) .and_then(move |mut data| { if data.is_empty() { From 8185c0cd2866cf14b2bed33d407b277acb735979 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 10:20:19 +0100 Subject: [PATCH 3/6] ref(server): Move content_length utility to utils --- server/src/body/store_body.rs | 29 ++++------------------------- server/src/utils/mod.rs | 2 ++ server/src/utils/request.rs | 9 +++++++++ 3 files changed, 15 insertions(+), 25 deletions(-) create mode 100644 server/src/utils/request.rs diff --git a/server/src/body/store_body.rs b/server/src/body/store_body.rs index 5c20938eb29..d2a19e77786 100644 --- a/server/src/body/store_body.rs +++ b/server/src/body/store_body.rs @@ -2,9 +2,8 @@ use std::borrow::Cow; use std::io::{self, Read}; use actix::ResponseFuture; -use actix_web::http::{header, StatusCode}; -use actix_web::HttpRequest; -use actix_web::{error::PayloadError, HttpMessage, HttpResponse, ResponseError}; +use actix_web::http::StatusCode; +use actix_web::{error::PayloadError, HttpMessage, HttpRequest, HttpResponse, ResponseError}; use base64::DecodeError; use bytes::{Bytes, BytesMut}; use failure::Fail; @@ -15,6 +14,7 @@ use url::form_urlencoded; use semaphore_common::metric; use crate::actors::outcome::DiscardReason; +use crate::utils; /// A set of errors that can occur during parsing json payloads #[derive(Fail, Debug)] @@ -94,14 +94,9 @@ impl StoreBody { }; } - let length_opt = match get_content_length(req) { - Ok(length_opt) => length_opt, - Err(e) => return Self::err(e), - }; - // Check the content length first. If we detect an overflow from the content length header, // keep the payload in the request. The `ReadRequestMiddleware` should drain the payload. - if let Some(length) = length_opt { + if let Some(length) = utils::get_content_length(req) { if length > limit { return Self::err(StorePayloadError::Overflow); } @@ -201,19 +196,3 @@ fn decode_bytes + AsRef<[u8]>>(body: B) -> Result(req: &HttpRequest) -> Result, StorePayloadError> { - if let Some(l) = req.headers().get(header::CONTENT_LENGTH) { - if let Ok(s) = l.to_str() { - if let Ok(l) = s.parse::() { - Ok(Some(l)) - } else { - Err(StorePayloadError::UnknownLength) - } - } else { - Err(StorePayloadError::UnknownLength) - } - } else { - Ok(None) - } -} diff --git a/server/src/utils/mod.rs b/server/src/utils/mod.rs index 27d24a9d701..481430be977 100644 --- a/server/src/utils/mod.rs +++ b/server/src/utils/mod.rs @@ -3,6 +3,7 @@ mod api; mod error_boundary; mod multipart; mod param_parser; +mod request; mod shutdown; mod timer; @@ -14,6 +15,7 @@ pub use self::api::*; pub use self::error_boundary::*; pub use self::multipart::*; pub use self::param_parser::*; +pub use self::request::*; pub use self::shutdown::*; pub use self::timer::*; diff --git a/server/src/utils/request.rs b/server/src/utils/request.rs new file mode 100644 index 00000000000..46df8d11d55 --- /dev/null +++ b/server/src/utils/request.rs @@ -0,0 +1,9 @@ +use actix_web::{http::header, HttpRequest}; + +// Resolve the content length from HTTP request headers. +pub fn get_content_length(req: &HttpRequest) -> Option { + req.headers() + .get(header::CONTENT_LENGTH) + .and_then(|h| h.to_str().ok()) + .and_then(|s| s.parse().ok()) +} From 0ebdad5c60ec025957cd894e5ff6671d5cd147e8 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 10:30:51 +0100 Subject: [PATCH 4/6] fix: Consume forward body streams correctly --- server/src/body/forward_body.rs | 70 ++++++++++++++------------------- server/src/body/store_body.rs | 2 +- server/src/endpoints/forward.rs | 3 +- server/src/utils/request.rs | 7 +++- 4 files changed, 36 insertions(+), 46 deletions(-) diff --git a/server/src/body/forward_body.rs b/server/src/body/forward_body.rs index f4d092e4284..7ec3ecbbbad 100644 --- a/server/src/body/forward_body.rs +++ b/server/src/body/forward_body.rs @@ -1,10 +1,11 @@ -use ::actix::ResponseFuture; -use actix_web::http::{header, StatusCode}; -use actix_web::{error::PayloadError, HttpMessage, HttpResponse, ResponseError}; +use actix::ResponseFuture; +use actix_web::{error::PayloadError, http::StatusCode, HttpMessage, HttpResponse, ResponseError}; use bytes::{Bytes, BytesMut}; use failure::Fail; use futures::prelude::*; +use crate::utils; + /// A set of errors that can occur during parsing json payloads #[derive(Fail, Debug)] pub enum ForwardPayloadError { @@ -43,7 +44,6 @@ impl From for ForwardPayloadError { /// Future that resolves to a complete store endpoint body. pub struct ForwardBody { limit: usize, - length: Option, stream: Option, err: Option, fut: Option>, @@ -51,42 +51,29 @@ pub struct ForwardBody { impl ForwardBody { /// Create `ForwardBody` for request. - pub fn new(req: &T) -> ForwardBody { - let mut len = None; - if let Some(l) = req.headers().get(header::CONTENT_LENGTH) { - if let Ok(s) = l.to_str() { - if let Ok(l) = s.parse::() { - len = Some(l) - } else { - return Self::err(ForwardPayloadError::UnknownLength); - } - } else { - return Self::err(ForwardPayloadError::UnknownLength); + pub fn new(req: &T, limit: usize) -> ForwardBody { + // Check the content length first. If we detect an overflow from the content length header, + // keep the payload in the request to drain it correctly in the `ReadRequestMiddleware`. + if let Some(length) = utils::get_content_length(req) { + if length > limit { + return Self::err(ForwardPayloadError::Overflow); } } ForwardBody { - limit: 262_144, - length: len, + limit, stream: Some(req.payload()), - fut: None, err: None, + fut: None, } } - /// Change max size of payload. By default max size is 256Kb - pub fn limit(mut self, limit: usize) -> Self { - self.limit = limit; - self - } - fn err(e: ForwardPayloadError) -> Self { ForwardBody { + limit: 0, stream: None, - limit: 262_144, fut: None, err: Some(e), - length: None, } } } @@ -107,27 +94,28 @@ where return Err(err); } - if let Some(len) = self.length.take() { - if len > self.limit { - return Err(ForwardPayloadError::Overflow); - } - } - let limit = self.limit; + let body = Some(BytesMut::with_capacity(8192)); + let future = self .stream .take() .expect("Can not be used second time") - .from_err() - .fold(BytesMut::with_capacity(8192), move |mut body, chunk| { - if (body.len() + chunk.len()) > limit { - Err(ForwardPayloadError::Overflow) - } else { - body.extend_from_slice(&chunk); - Ok(body) - } + .map_err(ForwardPayloadError::from) + .fold(body, move |body_opt, chunk| { + Ok::<_, ForwardPayloadError>(body_opt.and_then(|mut body| { + if (body.len() + chunk.len()) > limit { + None + } else { + body.extend_from_slice(&chunk); + Some(body) + } + })) }) - .map(BytesMut::freeze); + .and_then(|bytes_opt| match bytes_opt { + Some(bytes) => Ok(bytes.freeze()), + None => Err(ForwardPayloadError::Overflow), + }); self.fut = Some(Box::new(future)); diff --git a/server/src/body/store_body.rs b/server/src/body/store_body.rs index d2a19e77786..b7c0d94d95d 100644 --- a/server/src/body/store_body.rs +++ b/server/src/body/store_body.rs @@ -95,7 +95,7 @@ impl StoreBody { } // Check the content length first. If we detect an overflow from the content length header, - // keep the payload in the request. The `ReadRequestMiddleware` should drain the payload. + // keep the payload in the request to drain it correctly in the `ReadRequestMiddleware`. if let Some(length) = utils::get_content_length(req) { if length > limit { return Self::err(StorePayloadError::Overflow); diff --git a/server/src/endpoints/forward.rs b/server/src/endpoints/forward.rs index 6dc14011eb7..2790b4713e5 100644 --- a/server/src/endpoints/forward.rs +++ b/server/src/endpoints/forward.rs @@ -97,8 +97,7 @@ fn forward_upstream(request: &HttpRequest) -> ResponseFuture(req: &HttpRequest) -> Option { +pub fn get_content_length(req: &T) -> Option +where + T: HttpMessage, +{ req.headers() .get(header::CONTENT_LENGTH) .and_then(|h| h.to_str().ok()) From 0856bf57e1fdec5f2480db27ea1f2f4a26de6930 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 11:06:34 +0100 Subject: [PATCH 5/6] fix(server): Fully consume multipart bodies --- server/src/body/store_body.rs | 14 +++---- server/src/utils/multipart.rs | 77 +++++++++++++++++++---------------- 2 files changed, 48 insertions(+), 43 deletions(-) diff --git a/server/src/body/store_body.rs b/server/src/body/store_body.rs index b7c0d94d95d..3b05c1ddeb6 100644 --- a/server/src/body/store_body.rs +++ b/server/src/body/store_body.rs @@ -153,14 +153,12 @@ impl Future for StoreBody { } })) }) - .and_then(|body_opt| match body_opt { - Some(body) => { - metric!(time_raw("event.size_bytes.raw") = body.len() as u64); - let decoded = decode_bytes(body.freeze())?; - metric!(time_raw("event.size_bytes.uncompressed") = decoded.len() as u64); - Ok(decoded) - } - None => Err(StorePayloadError::Overflow), + .and_then(|body_opt| { + let body = body_opt.ok_or(StorePayloadError::Overflow)?; + metric!(time_raw("event.size_bytes.raw") = body.len() as u64); + let decoded = decode_bytes(body.freeze())?; + metric!(time_raw("event.size_bytes.uncompressed") = decoded.len() as u64); + Ok(decoded) }); self.fut = Some(Box::new(future)); diff --git a/server/src/utils/multipart.rs b/server/src/utils/multipart.rs index 0f5663b07b0..f0f239cf6f8 100644 --- a/server/src/utils/multipart.rs +++ b/server/src/utils/multipart.rs @@ -1,7 +1,7 @@ use actix::prelude::*; use actix_web::{dev::Payload, multipart, HttpMessage, HttpRequest}; use failure::Fail; -use futures::prelude::*; +use futures::{future, Future, Stream}; use serde::{Deserialize, Serialize}; use semaphore_common::LogError; @@ -91,21 +91,46 @@ impl<'a> Iterator for FormDataIter<'a> { } } -fn handle_multipart_item( +/// Reads data from a multipart field (coming from a HttpRequest) +fn consume_field( + field: multipart::Field, + max_size: usize, +) -> ResponseFuture>, MultipartError> { + let future = field.map_err(MultipartError::InvalidMultipart).fold( + Some(Vec::with_capacity(512)), + move |body_opt, chunk| { + Ok(body_opt.and_then(|mut body| { + if (body.len() + chunk.len()) > max_size { + None + } else { + body.extend_from_slice(&chunk); + Some(body) + } + })) + }, + ); + + Box::new(future) +} + +fn consume_item( mut content: MultipartEnvelope, item: multipart::MultipartItem, -) -> ResponseFuture { +) -> ResponseFuture, MultipartError> { let field = match item { + multipart::MultipartItem::Nested(nested) => return consume_stream(content, nested), multipart::MultipartItem::Field(field) => field, - multipart::MultipartItem::Nested(nested) => { - return handle_multipart_stream(content, nested); - } }; let content_type = field.content_type().to_string(); let content_disposition = field.content_disposition(); - let future = read_multipart_data(field, content.remaining_size).and_then(move |data| { + let future = consume_field(field, content.remaining_size).and_then(move |data_opt| { + let data = match data_opt { + Some(data) => data, + None => return Ok(None), + }; + content.remaining_size -= data.len(); let field_name = content_disposition.as_ref().and_then(|d| d.get_name()); @@ -129,45 +154,26 @@ fn handle_multipart_item( log::trace!("multipart content without name or file_name"); } - Ok(content) + Ok(Some(content)) }); Box::new(future) } -fn handle_multipart_stream( +fn consume_stream( content: MultipartEnvelope, stream: multipart::Multipart, -) -> ResponseFuture { - let future = stream - .map_err(MultipartError::InvalidMultipart) - .fold(content, move |content, item| { - handle_multipart_item(content, item) - }); - - Box::new(future) -} - -/// Reads data from a multipart field (coming from a HttpRequest) -fn read_multipart_data( - field: multipart::Field, - max_size: usize, -) -> ResponseFuture, MultipartError> { - let future = field.map_err(MultipartError::InvalidMultipart).fold( - Vec::with_capacity(512), - move |mut body, chunk| { - if (body.len() + chunk.len()) > max_size { - Err(MultipartError::Overflow) - } else { - body.extend_from_slice(&chunk); - Ok(body) - } +) -> ResponseFuture, MultipartError> { + let future = stream.map_err(MultipartError::InvalidMultipart).fold( + Some(content), + move |content_opt, item| match content_opt { + Some(content) => consume_item(content, item), + None => Box::new(future::ok(None)), }, ); Box::new(future) } - /// Internal structure used when constructing Envelopes to maintain a cap on the content size pub struct MultipartEnvelope { // the envelope under construction @@ -191,7 +197,8 @@ impl MultipartEnvelope { self, request: &HttpRequest, ) -> ResponseFuture { - let future = handle_multipart_stream(self, request.multipart()).and_then(|multipart| { + let future = consume_stream(self, request.multipart()).and_then(|multipart_opt| { + let multipart = multipart_opt.ok_or(MultipartError::Overflow)?; let mut envelope = multipart.envelope; let form_data = multipart.form_data.into_item(); From bfbdd2f51d0cacf24d74fb721b866e539d07cbd7 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Wed, 4 Dec 2019 11:22:39 +0100 Subject: [PATCH 6/6] ref(server): Code cleanup --- server/src/endpoints/minidump.rs | 5 ++--- server/src/utils/multipart.rs | 14 +++++++------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/server/src/endpoints/minidump.rs b/server/src/endpoints/minidump.rs index c6b9a526ff0..462cebb303b 100644 --- a/server/src/endpoints/minidump.rs +++ b/server/src/endpoints/minidump.rs @@ -3,6 +3,7 @@ use futures::Future; use semaphore_general::protocol::EventId; +use crate::body::ForwardBody; use crate::endpoints::common::{handle_store_like_request, BadStoreRequest}; use crate::envelope::{AttachmentType, ContentType, Envelope, Item, ItemType}; use crate::extractors::{EventMeta, StartTime}; @@ -62,9 +63,7 @@ where { // minidump can either be transmitted as request body, or as `upload_file_minidump` in a // multipart formdata request. if MINIDUMP_RAW_CONTENT_TYPES.contains(&request.content_type()) { - let future = request - .body() - .limit(max_payload_size) + let future = ForwardBody::new(request, max_payload_size) .map_err(|_| BadStoreRequest::InvalidMinidump) .and_then(move |data| { validate_minidump(&data)?; diff --git a/server/src/utils/multipart.rs b/server/src/utils/multipart.rs index f0f239cf6f8..c3151c0b2f6 100644 --- a/server/src/utils/multipart.rs +++ b/server/src/utils/multipart.rs @@ -125,12 +125,8 @@ fn consume_item( let content_type = field.content_type().to_string(); let content_disposition = field.content_disposition(); - let future = consume_field(field, content.remaining_size).and_then(move |data_opt| { - let data = match data_opt { - Some(data) => data, - None => return Ok(None), - }; - + let future = consume_field(field, content.remaining_size).map(move |data_opt| { + let data = data_opt?; content.remaining_size -= data.len(); let field_name = content_disposition.as_ref().and_then(|d| d.get_name()); @@ -154,7 +150,7 @@ fn consume_item( log::trace!("multipart content without name or file_name"); } - Ok(Some(content)) + Some(content) }); Box::new(future) @@ -164,6 +160,9 @@ fn consume_stream( content: MultipartEnvelope, stream: multipart::Multipart, ) -> ResponseFuture, MultipartError> { + // Ensure that we consume the entire stream here. If we overflow at a certain point, + // `consume_item` will return `None`. We need to continue folding, however, to ensure that we + // consume the entire request payload. let future = stream.map_err(MultipartError::InvalidMultipart).fold( Some(content), move |content_opt, item| match content_opt { @@ -174,6 +173,7 @@ fn consume_stream( Box::new(future) } + /// Internal structure used when constructing Envelopes to maintain a cap on the content size pub struct MultipartEnvelope { // the envelope under construction