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 801eeedd8f3..3b05c1ddeb6 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)] @@ -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,38 @@ 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, + // 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(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,27 +133,28 @@ 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)); + 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| { + .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); @@ -201,19 +194,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/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 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() { 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/multipart.rs b/server/src/utils/multipart.rs index 0f5663b07b0..c3151c0b2f6 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,42 @@ 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).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()); @@ -129,39 +150,24 @@ fn handle_multipart_item( log::trace!("multipart content without name or file_name"); } - Ok(content) + 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> { + // 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 { + Some(content) => consume_item(content, item), + None => Box::new(future::ok(None)), }, ); @@ -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(); diff --git a/server/src/utils/request.rs b/server/src/utils/request.rs new file mode 100644 index 00000000000..9ed06969c1e --- /dev/null +++ b/server/src/utils/request.rs @@ -0,0 +1,12 @@ +use actix_web::{http::header, HttpMessage}; + +// Resolve the content length from HTTP request headers. +pub fn get_content_length(req: &T) -> Option +where + T: HttpMessage, +{ + req.headers() + .get(header::CONTENT_LENGTH) + .and_then(|h| h.to_str().ok()) + .and_then(|s| s.parse().ok()) +}