From 21b6e1dc083552997c214db3fe9dd28bbc7a3321 Mon Sep 17 00:00:00 2001 From: "Aode (lion)" Date: Tue, 1 Mar 2022 11:23:15 -0600 Subject: [PATCH] Use actix-web's Range header --- src/error.rs | 6 +- src/main.rs | 27 +++---- src/range.rs | 206 +++++++++++---------------------------------------- 3 files changed, 59 insertions(+), 180 deletions(-) diff --git a/src/error.rs b/src/error.rs index 69abbe7..deec103 100644 --- a/src/error.rs +++ b/src/error.rs @@ -57,9 +57,6 @@ pub(crate) enum UploadError { #[error("Error parsing string, {0}")] ParseString(#[from] std::string::FromUtf8Error), - #[error("Error parsing request, {0}")] - ParseReq(String), - #[error("Error interacting with filesystem, {0}")] Io(#[from] std::io::Error), @@ -155,8 +152,7 @@ impl ResponseError for Error { UploadError::DuplicateAlias | UploadError::Limit(_) | UploadError::NoFiles - | UploadError::Upload(_) - | UploadError::ParseReq(_) => StatusCode::BAD_REQUEST, + | UploadError::Upload(_) => StatusCode::BAD_REQUEST, UploadError::MissingAlias | UploadError::MissingFilename => StatusCode::NOT_FOUND, UploadError::InvalidToken => StatusCode::FORBIDDEN, UploadError::Range => StatusCode::RANGE_NOT_SATISFIABLE, diff --git a/src/main.rs b/src/main.rs index e65f958..6de2bb5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use actix_form_data::{Field, Form, Value}; use actix_web::{ guard, - http::header::{CacheControl, CacheDirective, LastModified, ACCEPT_RANGES}, + http::header::{CacheControl, CacheDirective, LastModified, Range, ACCEPT_RANGES}, web, App, HttpResponse, HttpResponseBuilder, HttpServer, }; use awc::Client; @@ -341,7 +341,7 @@ where /// Process files #[instrument(name = "Serving processed image", skip(manager, filters))] async fn process( - range: Option, + range: Option>, query: web::Query, ext: web::Path, manager: web::Data, @@ -440,12 +440,14 @@ where let (details, bytes) = CancelSafeProcessor::new(thumbnail_path.clone(), process_fut).await?; - let (builder, stream) = if let Some(range_header) = range { - if let Some(range) = range_header.single_bytes_range() { - if let Some(content_range) = range.to_content_range(bytes.len() as u64) { + let (builder, stream) = if let Some(web::Header(range_header)) = range { + if let Some(range) = range::single_bytes_range(&range_header) { + let len = bytes.len() as u64; + + if let Some(content_range) = range::to_content_range(range, len) { let mut builder = HttpResponse::PartialContent(); builder.insert_header(content_range); - let stream = range.chop_bytes(bytes); + let stream = range::chop_bytes(range, bytes, len)?; (builder, Either::left(Either::left(stream))) } else { @@ -502,7 +504,7 @@ where /// Serve files #[instrument(name = "Serving file", skip(manager))] async fn serve( - range: Option, + range: Option>, alias: web::Path, manager: web::Data, store: web::Data, @@ -532,25 +534,24 @@ where async fn ranged_file_resp( store: &S, identifier: S::Identifier, - range: Option, + range: Option>, details: Details, ) -> Result where Error: From, { - let (builder, stream) = if let Some(range_header) = range { + let (builder, stream) = if let Some(web::Header(range_header)) = range { //Range header exists - return as ranged - if let Some(range) = range_header.single_bytes_range() { + if let Some(range) = range::single_bytes_range(&range_header) { let len = store.len(&identifier).await?; - if let Some(content_range) = range.to_content_range(len) { + if let Some(content_range) = range::to_content_range(range, len) { let mut builder = HttpResponse::PartialContent(); builder.insert_header(content_range); - ( builder, Either::left(Either::left(map_error::map_crate_error( - range.chop_store(store, identifier).await?, + range::chop_store(range, store, &identifier, len).await?, ))), ) } else { diff --git a/src/range.rs b/src/range.rs index 43c9902..e5d9efa 100644 --- a/src/range.rs +++ b/src/range.rs @@ -3,180 +3,62 @@ use crate::{ store::Store, }; use actix_web::{ - dev::Payload, - http::header::{ContentRange, ContentRangeSpec, HeaderValue}, + http::header::{ByteRangeSpec, ContentRange, ContentRangeSpec, Range}, web::Bytes, - FromRequest, HttpRequest, }; use futures_util::stream::{once, Stream}; use std::future::ready; -#[derive(Debug)] -pub(crate) enum Range { - Start(u64), - SuffixLength(u64), - Segment(u64, u64), +pub(crate) fn chop_bytes( + byte_range: &ByteRangeSpec, + bytes: Bytes, + length: u64, +) -> Result>, Error> { + if let Some((start, end)) = byte_range.to_satisfiable_range(length) { + return Ok(once(ready(Ok(bytes.slice(start as usize..end as usize))))); + } + + Err(UploadError::Range.into()) } -#[derive(Debug)] -pub(crate) struct RangeHeader { - unit: String, - ranges: Vec, +pub(crate) async fn chop_store( + byte_range: &ByteRangeSpec, + store: &S, + identifier: &S::Identifier, + length: u64, +) -> Result>, Error> +where + Error: From, +{ + if let Some((start, end)) = byte_range.to_satisfiable_range(length) { + return Ok(store + .to_stream(identifier, Some(start), Some(end.saturating_sub(start))) + .await?); + } + + Err(UploadError::Range.into()) } -impl Range { - pub(crate) fn to_content_range(&self, instance_length: u64) -> Option { - match self { - Range::Start(start) => { - if *start >= instance_length { - return None; - } - - Some(ContentRange(ContentRangeSpec::Bytes { - range: Some((*start, instance_length - *start)), - instance_length: Some(instance_length), - })) - } - Range::SuffixLength(from_start) => { - if *from_start > instance_length { - return None; - } - - Some(ContentRange(ContentRangeSpec::Bytes { - range: Some((0, *from_start)), - instance_length: Some(instance_length), - })) - } - Range::Segment(start, end) => { - if *start >= instance_length || *end > instance_length { - return None; - } - - Some(ContentRange(ContentRangeSpec::Bytes { - range: Some((*start, *end)), - instance_length: Some(instance_length), - })) - } +pub(crate) fn single_bytes_range(range: &Range) -> Option<&ByteRangeSpec> { + if let Range::Bytes(ranges) = range { + if ranges.len() == 1 { + return ranges.get(0); } } - pub(crate) fn chop_bytes(&self, bytes: Bytes) -> impl Stream> { - match self { - Range::Start(start) => once(ready(Ok(bytes.slice(*start as usize..)))), - Range::SuffixLength(from_start) => once(ready(Ok(bytes.slice(..*from_start as usize)))), - Range::Segment(start, end) => { - once(ready(Ok(bytes.slice(*start as usize..*end as usize)))) - } - } - } - - pub(crate) async fn chop_store( - &self, - store: &S, - identifier: S::Identifier, - ) -> Result>, Error> - where - Error: From, - { - match self { - Range::Start(start) => Ok(store.to_stream(&identifier, Some(*start), None).await?), - Range::SuffixLength(from_start) => Ok(store - .to_stream(&identifier, None, Some(*from_start)) - .await?), - Range::Segment(start, end) => Ok(store - .to_stream(&identifier, Some(*start), Some(end.saturating_sub(*start))) - .await?), - } - } + None } -impl RangeHeader { - pub(crate) fn single_bytes_range(&self) -> Option<&'_ Range> { - if self.ranges.len() == 1 && self.unit == "bytes" { - self.ranges.get(0) - } else { - None - } - } -} - -impl FromRequest for RangeHeader { - type Error = Error; - type Future = std::future::Ready>; - - fn from_request(req: &HttpRequest, _: &mut Payload) -> Self::Future { - if let Some(range_head) = req.headers().get("Range") { - ready(parse_range_header(range_head).map_err(|e| { - tracing::warn!("Failed to parse range header: {}", e); - e - })) - } else { - ready(Err(UploadError::ParseReq( - "Range header missing".to_string(), - ) - .into())) - } - } -} - -fn parse_range_header(range_head: &HeaderValue) -> Result { - let range_head_str = range_head.to_str().map_err(|_| { - UploadError::ParseReq("Range header contains non-utf8 characters".to_string()) - })?; - - let eq_pos = range_head_str - .find('=') - .ok_or_else(|| UploadError::ParseReq("Malformed Range Header".to_string()))?; - - let (unit, ranges) = range_head_str.split_at(eq_pos); - let ranges = ranges.trim_start_matches('='); - - let ranges = ranges - .split(',') - .map(parse_range) - .collect::, Error>>()?; - - Ok(RangeHeader { - unit: unit.to_owned(), - ranges, - }) -} - -fn parse_range(s: &str) -> Result { - let dash_pos = s - .find('-') - .ok_or_else(|| UploadError::ParseReq("Mailformed Range Bound".to_string()))?; - - let (start, end) = s.split_at(dash_pos); - let start = start.trim(); - let end = end.trim_start_matches('-').trim(); - - if start.is_empty() && end.is_empty() { - Err(UploadError::ParseReq("Malformed content range".to_string()).into()) - } else if start.is_empty() { - let suffix_length = end.parse().map_err(|_| { - UploadError::ParseReq("Cannot parse suffix length for range header".to_string()) - })?; - - Ok(Range::SuffixLength(suffix_length)) - } else if end.is_empty() { - let range_start = start.parse().map_err(|_| { - UploadError::ParseReq("Cannot parse range start for range header".to_string()) - })?; - - Ok(Range::Start(range_start)) - } else { - let range_start = start.parse().map_err(|_| { - UploadError::ParseReq("Cannot parse range start for range header".to_string()) - })?; - let range_end = end.parse().map_err(|_| { - UploadError::ParseReq("Cannot parse range end for range header".to_string()) - })?; - - if range_start > range_end { - return Err(UploadError::Range.into()); - } - - Ok(Range::Segment(range_start, range_end)) - } +pub(crate) fn to_content_range( + byte_range: &ByteRangeSpec, + instance_length: u64, +) -> Option { + byte_range + .to_satisfiable_range(instance_length) + .map(|range| { + ContentRange(ContentRangeSpec::Bytes { + range: Some(range), + instance_length: Some(instance_length), + }) + }) }