use futures_util::{StreamExt, TryStreamExt};
use http_body_util::{combinators::UnsyncBoxBody, BodyExt, Full};
use hyper::body::{Bytes, Frame, Incoming};
use hyper::Request;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, SeekFrom};
use tokio::sync::Mutex;
use webx_core::auth::IClaims;
use webx_core::config::AppOptions;
use webx_core::error::{Error, Result};
use webx_core::form::{FormFileBuilder, MultipartForm};
use webx_core::http::{
content_disposition_value, ByteRange, FileBody, FileSource, IClaimsExt, IHttpContext,
IHttpRequest, IHttpResponse, ResponseBody, SeekableReader,
};
use crate::problem_response::{build_problem, problem_to_bytes};
pub type RespBody = UnsyncBoxBody<Bytes, std::io::Error>;
#[derive(Debug, Clone)]
pub struct BodyLimits {
pub body: usize,
pub multipart_body: u64,
pub file: u64,
pub field: usize,
pub memory_threshold: usize,
pub spool_dir: PathBuf,
}
impl Default for BodyLimits {
fn default() -> Self {
Self {
body: 10 * 1024 * 1024,
multipart_body: 256 * 1024 * 1024,
file: 128 * 1024 * 1024,
field: 1024 * 1024,
memory_threshold: 1024 * 1024,
spool_dir: webx_core::form::spool_root(),
}
}
}
impl BodyLimits {
pub fn from_options(options: &AppOptions) -> Self {
let spool_dir = match options.form.temp_dir.as_deref() {
Some(dir) if !dir.trim().is_empty() => {
let path = PathBuf::from(dir);
if path.is_absolute() {
path
} else {
webx_core::paths::app_base().join(path)
}
}
_ => webx_core::form::spool_root(),
};
Self {
body: options.app.max_body_size,
multipart_body: options.form.max_request_size as u64,
file: options.form.max_file_size,
field: options.form.max_field_size,
memory_threshold: options.form.memory_threshold,
spool_dir,
}
}
pub fn apply_spool_root(&self) {
webx_core::form::set_spool_root(self.spool_dir.clone());
}
}
pub struct HttpContext {
req: HttpRequest,
resp: HttpResponse,
claims: Option<Box<dyn IClaims>>,
}
impl HttpContext {
pub fn new(req: Request<Incoming>, limits: BodyLimits) -> Self {
let (parts, body) = req.into_parts();
let mut headers = HashMap::new();
for (name, value) in parts.headers.iter() {
if let Ok(value) = value.to_str() {
headers.insert(name.to_string(), value.to_string());
}
}
let (query_params, query_multi) = parts
.uri
.query()
.map(parse_query_string)
.unwrap_or_default();
let mut resp = HttpResponse::new(200);
let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
limits.multipart_body
} else {
limits.body as u64
};
if let Some(declared) = headers
.get("content-length")
.and_then(|value| value.parse::<u64>().ok())
{
if declared > declared_limit {
let problem = build_problem(
413,
format!("request body exceeds the {declared_limit}-byte limit"),
);
resp.set_status(413);
resp.set_header("content-type", "application/problem+json");
resp.body = Some(ResponseBody::Bytes(problem_to_bytes(&problem)));
}
}
let req = HttpRequest {
method: parts.method.to_string(),
path: parts.uri.path().to_string(),
headers,
query_params,
query_multi,
route_params: HashMap::new(),
route_pattern: None,
body: Mutex::new(RequestBody::Pending(Some(body))),
limits,
};
Self {
req,
resp,
claims: None,
}
}
pub async fn into_response(self) -> hyper::Response<RespBody> {
let HttpContext { req, resp, .. } = self;
finalize_response(&req, resp).await
}
}
impl IClaimsExt for HttpContext {
fn set_claims(&mut self, claims: Box<dyn IClaims>) {
self.claims = Some(claims);
}
fn claims(&self) -> Option<&dyn IClaims> {
self.claims.as_deref()
}
}
impl IHttpContext for HttpContext {
fn request(&self) -> &dyn IHttpRequest {
&self.req
}
fn request_mut(&mut self) -> &mut dyn IHttpRequest {
&mut self.req
}
fn response(&self) -> &dyn IHttpResponse {
&self.resp
}
fn response_mut(&mut self) -> &mut dyn IHttpResponse {
&mut self.resp
}
}
enum RequestBody {
Pending(Option<Incoming>),
Buffered(Vec<u8>),
Form(Arc<MultipartForm>),
}
pub struct HttpRequest {
method: String,
path: String,
headers: HashMap<String, String>,
query_params: HashMap<String, String>,
query_multi: HashMap<String, Vec<String>>,
route_params: HashMap<String, String>,
route_pattern: Option<String>,
body: Mutex<RequestBody>,
limits: BodyLimits,
}
#[async_trait::async_trait]
impl IHttpRequest for HttpRequest {
fn method(&self) -> &str {
&self.method
}
fn path(&self) -> &str {
&self.path
}
fn header(&self, name: &str) -> Option<&str> {
if let Some(value) = self.headers.get(name) {
return Some(value.as_str());
}
let lower = name.to_ascii_lowercase();
self.headers.get(&lower).map(|s| s.as_str())
}
fn query(&self) -> &HashMap<String, String> {
&self.query_params
}
fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
Some(&self.query_multi)
}
fn route_params(&self) -> &HashMap<String, String> {
&self.route_params
}
fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
&mut self.route_params
}
fn route_pattern(&self) -> Option<&str> {
self.route_pattern.as_deref()
}
fn route_pattern_mut(&mut self) -> &mut Option<String> {
&mut self.route_pattern
}
async fn body_bytes(&mut self) -> Result<Vec<u8>> {
let mut guard = self.body.lock().await;
let incoming = match &mut *guard {
RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
RequestBody::Form(_) => {
return Err(Error::Http(
"request body was already read as multipart/form-data".to_string(),
))
}
RequestBody::Pending(slot) => slot.take(),
};
let Some(incoming) = incoming else {
return Err(Error::Http(
"request body has already been consumed".to_string(),
));
};
let bytes = read_body(incoming, self.limits.body).await?;
*guard = RequestBody::Buffered(bytes.clone());
Ok(bytes)
}
async fn body_text(&mut self) -> Result<String> {
let bytes = self.body_bytes().await?;
String::from_utf8(bytes).map_err(|e| Error::Http(e.to_string()))
}
async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
let mut guard = self.body.lock().await;
match &*guard {
RequestBody::Form(form) => return Ok(Arc::clone(form)),
RequestBody::Buffered(_) => {
return Err(Error::UnsupportedMediaType(
"request body was already read as a non-multipart body".to_string(),
))
}
RequestBody::Pending(_) => {}
}
if !self.is_multipart() {
return Err(Error::UnsupportedMediaType(
"expected Content-Type: multipart/form-data".to_string(),
));
}
let raw_content_type = self.header("content-type").unwrap_or("");
let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
Error::Validation(
"multipart/form-data request is missing its boundary parameter".to_string(),
)
})?;
let incoming = match &mut *guard {
RequestBody::Pending(slot) => slot.take(),
_ => None,
};
let Some(body) = incoming else {
return Err(Error::Http(
"request body has already been consumed".to_string(),
));
};
let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
*guard = RequestBody::Form(Arc::clone(&form));
Ok(form)
}
}
async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
let mut out = Vec::new();
let mut total = 0usize;
let mut stream = incoming.into_data_stream();
while let Some(chunk) = stream.next().await {
let chunk =
chunk.map_err(|err| Error::Http(format!("failed to read request body: {err}")))?;
total = total.saturating_add(chunk.len());
if total > limit {
return Err(Error::PayloadTooLarge(format!(
"request body exceeds the {limit}-byte limit"
)));
}
out.extend_from_slice(&chunk);
}
Ok(out)
}
fn is_multipart_content_type(content_type: Option<&String>) -> bool {
content_type
.map(|value| {
value
.split(';')
.next()
.unwrap_or("")
.trim()
.eq_ignore_ascii_case("multipart/form-data")
})
.unwrap_or(false)
}
fn multipart_boundary(content_type: &str) -> Option<String> {
for part in content_type.split(';').skip(1) {
let part = part.trim();
let Some((key, value)) = part.split_once('=') else {
continue;
};
if key.trim().eq_ignore_ascii_case("boundary") {
let value = value.trim().trim_matches('"');
if !value.is_empty() {
return Some(value.to_string());
}
}
}
None
}
async fn parse_multipart(
incoming: Incoming,
boundary: &str,
limits: &BodyLimits,
) -> Result<MultipartForm> {
let stream = incoming.into_data_stream();
let mut multipart = multer::Multipart::new(stream, boundary);
let mut form = MultipartForm::new();
let mut total: u64 = 0;
while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
let field_name = field.name().unwrap_or_default().to_string();
if field_name.is_empty() {
continue;
}
let file_name = field.file_name().map(str::to_string);
let content_type = field.content_type().map(|m| m.to_string());
match file_name {
Some(file_name) => {
let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
.memory_threshold(limits.memory_threshold)
.max_size(limits.file)
.spool_dir(limits.spool_dir.clone());
let mut field = field;
while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
total = charge(&mut total, chunk.len(), limits.multipart_body)?;
builder.write(&chunk).await?;
}
form.push_file(builder.finish().await?);
}
None => {
let mut value: Vec<u8> = Vec::new();
let mut field = field;
while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
total = charge(&mut total, chunk.len(), limits.multipart_body)?;
if value.len() + chunk.len() > limits.field {
return Err(Error::PayloadTooLarge(format!(
"form field `{field_name}` exceeds the {}-byte limit",
limits.field
)));
}
value.extend_from_slice(&chunk);
}
let text = String::from_utf8(value).map_err(|_| {
Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
})?;
form.push_field(field_name, text);
}
}
}
Ok(form)
}
fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
*total = total.saturating_add(len as u64);
if *total > limit {
return Err(Error::PayloadTooLarge(format!(
"multipart request exceeds the {limit}-byte limit"
)));
}
Ok(*total)
}
fn multer_error(err: multer::Error) -> Error {
match err {
multer::Error::FieldSizeExceeded { limit, .. } => {
Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
}
multer::Error::StreamSizeExceeded { limit } => {
Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
}
multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
Error::Validation("multipart body ended unexpectedly".to_string())
}
other => Error::Validation(format!("invalid multipart body: {other}")),
}
}
pub struct HttpResponse {
status: u16,
headers: HashMap<String, String>,
body: Option<ResponseBody>,
}
impl HttpResponse {
pub fn new(status: u16) -> Self {
Self {
status,
headers: HashMap::new(),
body: None,
}
}
}
#[async_trait::async_trait]
impl IHttpResponse for HttpResponse {
fn status(&self) -> u16 {
self.status
}
fn has_body(&self) -> bool {
self.body.is_some()
}
fn set_status(&mut self, code: u16) {
self.status = code;
}
fn set_header(&mut self, key: &str, value: &str) {
self.headers.insert(key.to_string(), value.to_string());
}
fn remove_header(&mut self, key: &str) {
self.headers.remove(key);
}
async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
self.body = Some(ResponseBody::Bytes(data));
Ok(())
}
async fn write_text(&mut self, text: &str) -> Result<()> {
self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
Ok(())
}
async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
self.body = Some(body);
Ok(())
}
fn body_bytes(&self) -> Vec<u8> {
self.body
.as_ref()
.and_then(ResponseBody::as_bytes)
.map(<[u8]>::to_vec)
.unwrap_or_default()
}
fn header(&self, key: &str) -> Option<&str> {
self.headers.get(key).map(|s| s.as_str())
}
}
async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
let HttpResponse {
status,
mut headers,
body,
} = resp;
match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
ResponseBody::Bytes(bytes) => {
default_cache_control(&mut headers, false);
bytes_response(status, headers, bytes)
}
ResponseBody::File(file) => {
default_cache_control(&mut headers, true);
file_response(req, status, headers, file).await
}
}
}
fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
if headers.contains_key("cache-control") {
return;
}
headers.insert(
"cache-control".to_string(),
if is_file {
"public, max-age=0, must-revalidate".to_string()
} else {
"no-store".to_string()
},
);
}
enum PreparedSource {
Path(tokio::fs::File),
Seekable(Box<dyn SeekableReader>),
Stream(Box<dyn AsyncRead + Send + Unpin>),
}
impl PreparedSource {
fn is_seekable(&self) -> bool {
matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
}
}
async fn file_response(
req: &HttpRequest,
status: u16,
mut headers: HashMap<String, String>,
file: FileBody,
) -> hyper::Response<RespBody> {
let FileBody {
source,
content_type,
file_name,
disposition,
enable_range_processing,
last_modified,
entity_tag,
} = file;
let mut modified = last_modified;
let mut etag = entity_tag;
let mut source_path: Option<PathBuf> = None;
let known_len;
let prepared = match source {
FileSource::Path(path) => {
let handle = match tokio::fs::File::open(&path).await {
Ok(handle) => handle,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return problem_bytes(404, headers, "Not Found")
}
Err(err) => {
return problem_bytes(
500,
headers,
&format!("cannot open `{}`: {err}", path.display()),
)
}
};
let metadata = match handle.metadata().await {
Ok(metadata) => metadata,
Err(err) => {
return problem_bytes(
500,
headers,
&format!("cannot stat `{}`: {err}", path.display()),
)
}
};
if !metadata.is_file() {
return problem_bytes(404, headers, "Not Found");
}
let len = metadata.len();
if modified.is_none() {
modified = metadata.modified().ok();
}
if etag.is_none() {
etag = Some(etag_for(len, modified));
}
known_len = Some(len);
source_path = Some(path);
PreparedSource::Path(handle)
}
FileSource::SeekableStream { reader, len } => {
known_len = Some(len);
PreparedSource::Seekable(reader)
}
FileSource::Stream(reader) => {
known_len = None;
PreparedSource::Stream(reader)
}
};
let modified = modified.filter(|value| *value <= SystemTime::now());
if !headers.contains_key("content-type") {
let resolved = content_type
.or_else(|| source_path.as_deref().and_then(guess_content_type))
.unwrap_or_else(|| "application/octet-stream".to_string());
headers.insert("content-type".to_string(), resolved);
}
if !headers.contains_key("content-disposition") {
if let Some(name) = &file_name {
headers.insert(
"content-disposition".to_string(),
content_disposition_value(disposition, name),
);
}
}
if let Some(tag) = &etag {
headers.insert("etag".to_string(), tag.clone());
}
if let Some(modified) = modified {
headers.insert(
"last-modified".to_string(),
httpdate::fmt_http_date(modified),
);
}
let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
headers.insert(
"accept-ranges".to_string(),
if range_capable { "bytes" } else { "none" }.to_string(),
);
if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
headers.remove("content-length");
return bytes_response(304, headers, Vec::new());
}
let mut ranges: Vec<ByteRange> = Vec::new();
if status == 200 && range_capable {
let len = known_len.expect("range-capable sources know their length");
if let Some(raw) = req.header("range") {
if if_range_matches(req, etag.as_deref(), modified) {
match parse_ranges(raw, len) {
RangeOutcome::Ignored => {}
RangeOutcome::Unsatisfiable => {
headers.insert("content-range".to_string(), format!("bytes */{len}"));
return problem_bytes(416, headers, "Range Not Satisfiable");
}
RangeOutcome::Satisfiable(parsed) => ranges = parsed,
}
}
}
}
if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
ranges.clear();
}
if ranges.len() > 1 {
return multipart_byteranges_response(prepared, headers, ranges, known_len);
}
let (status, start, count) = match ranges.first() {
Some(range) => {
let len = known_len.expect("range-capable sources know their length");
headers.insert(
"content-range".to_string(),
format!("bytes {}-{}/{len}", range.start, range.end),
);
(206, range.start, range.len())
}
None => (status, 0, known_len.unwrap_or(0)),
};
match prepared {
PreparedSource::Path(mut handle) => {
headers.insert("content-length".to_string(), count.to_string());
if count == 0 {
return bytes_response(status, headers, Vec::new());
}
if start > 0 {
if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
return problem_bytes(500, headers, &format!("cannot seek: {err}"));
}
}
build_response(status, headers, reader_body(handle.take(count)))
}
PreparedSource::Seekable(mut reader) => {
headers.insert("content-length".to_string(), count.to_string());
if count == 0 {
return bytes_response(status, headers, Vec::new());
}
if start > 0 {
if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
return problem_bytes(500, headers, &format!("cannot seek: {err}"));
}
}
build_response(status, headers, reader_body(reader.take(count)))
}
PreparedSource::Stream(reader) => {
build_response(status, headers, reader_body(reader))
}
}
}
fn representation_content_type(headers: &HashMap<String, String>) -> String {
headers
.get("content-type")
.filter(|value| !value.starts_with("multipart/byteranges"))
.cloned()
.unwrap_or_else(|| "application/octet-stream".to_string())
}
fn multipart_byteranges_response(
prepared: PreparedSource,
mut headers: HashMap<String, String>,
ranges: Vec<ByteRange>,
known_len: Option<u64>,
) -> hyper::Response<RespBody> {
let PreparedSource::Path(handle) = prepared else {
unreachable!("multi-range is only routed here for path sources");
};
let total_len = known_len.unwrap_or(0);
let media_type = representation_content_type(&headers);
let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
let parts: Vec<(Bytes, ByteRange)> = ranges
.iter()
.map(|range| {
let header = format!(
"--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
range.start, range.end
);
(Bytes::from(header), *range)
})
.collect();
let closing = Bytes::from(format!("--{boundary}--\r\n"));
let content_length: u64 = parts
.iter()
.map(|(header, range)| header.len() as u64 + range.len() + 2)
.sum::<u64>()
+ closing.len() as u64;
headers.insert(
"content-type".to_string(),
format!("multipart/byteranges; boundary={boundary}"),
);
headers.insert("content-length".to_string(), content_length.to_string());
headers.remove("content-range");
let stream = multipart_stream(handle, parts, closing);
build_response(
206,
headers,
BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
)
}
const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
struct MultipartState {
handle: tokio::fs::File,
parts: std::vec::IntoIter<(Bytes, ByteRange)>,
closing: Bytes,
pending_header: Bytes,
phase: MultipartPhase,
buffer: bytes::BytesMut,
}
enum MultipartPhase {
NextPart,
Payload(u64),
Terminator,
Closing,
Done,
}
fn multipart_stream(
handle: tokio::fs::File,
parts: Vec<(Bytes, ByteRange)>,
closing: Bytes,
) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
let state = MultipartState {
handle,
parts: parts.into_iter(),
closing,
pending_header: Bytes::new(),
phase: MultipartPhase::NextPart,
buffer: bytes::BytesMut::new(),
};
futures_util::stream::unfold(state, |mut state| async move {
loop {
match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
MultipartPhase::NextPart => match state.parts.next() {
Some((header, range)) => {
if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
return Some((Err(err), state));
}
state.pending_header = header;
state.phase = MultipartPhase::Payload(range.len());
}
None => state.phase = MultipartPhase::Closing,
},
MultipartPhase::Payload(remaining) => {
if remaining == 0 {
state.phase = MultipartPhase::Terminator;
continue;
}
let limit = MULTIPART_CHUNK.min(remaining as usize);
state.buffer.clear();
state.buffer.reserve(limit);
let read = match state.handle.read_buf(&mut state.buffer).await {
Ok(read) => read,
Err(err) => return Some((Err(err), state)),
};
if read == 0 {
return Some((
Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"file changed size while serving byte ranges",
)),
state,
));
}
let keep = read.min(limit);
state.buffer.truncate(keep);
let chunk = state.buffer.split().freeze();
let header = std::mem::take(&mut state.pending_header);
let frame = if header.is_empty() {
chunk
} else {
let mut combined =
bytes::BytesMut::with_capacity(header.len() + chunk.len());
combined.extend_from_slice(&header);
combined.extend_from_slice(&chunk);
combined.freeze()
};
state.phase = MultipartPhase::Payload(remaining - keep as u64);
return Some((Ok(Frame::data(frame)), state));
}
MultipartPhase::Terminator => {
state.phase = MultipartPhase::NextPart;
return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
}
MultipartPhase::Closing => {
state.phase = MultipartPhase::Done;
let closing = std::mem::take(&mut state.closing);
return Some((Ok(Frame::data(closing)), state));
}
MultipartPhase::Done => return None,
}
}
})
}
const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
fn reader_body<R>(reader: R) -> RespBody
where
R: tokio::io::AsyncRead + Send + 'static,
{
let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
BodyExt::boxed_unsync(body)
}
fn bytes_response(
status: u16,
headers: HashMap<String, String>,
bytes: Vec<u8>,
) -> hyper::Response<RespBody> {
build_response(status, headers, full_body(bytes))
}
fn full_body(bytes: Vec<u8>) -> RespBody {
Full::new(Bytes::from(bytes))
.map_err(|never: std::convert::Infallible| match never {})
.boxed_unsync()
}
fn problem_bytes(
status: u16,
mut headers: HashMap<String, String>,
detail: &str,
) -> hyper::Response<RespBody> {
headers.insert(
"content-type".to_string(),
"application/problem+json".to_string(),
);
bytes_response(
status,
headers,
problem_to_bytes(&build_problem(status, detail)),
)
}
fn build_response(
status: u16,
headers: HashMap<String, String>,
body: RespBody,
) -> hyper::Response<RespBody> {
let mut builder = hyper::Response::builder().status(status);
for (key, value) in &headers {
builder = builder.header(key.as_str(), value.as_str());
}
builder.body(body).unwrap_or_else(|_| {
hyper::Response::builder()
.status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
.body(full_body(b"Internal Server Error".to_vec()))
.expect("static error response is valid")
})
}
#[derive(Debug, PartialEq, Eq)]
enum RangeOutcome {
Ignored,
Unsatisfiable,
Satisfiable(Vec<ByteRange>),
}
const MAX_RANGE_PARTS: usize = 8;
fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
let value = value.trim();
let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
return RangeOutcome::Ignored;
};
let mut ranges: Vec<ByteRange> = Vec::new();
let mut saw_valid_spec = false;
for part in spec.split(',') {
let part = part.trim();
if part.is_empty() {
continue;
}
match parse_single_range(part, len) {
SingleRange::Malformed => return RangeOutcome::Ignored,
SingleRange::Unsatisfiable => saw_valid_spec = true,
SingleRange::Satisfiable(range) => {
saw_valid_spec = true;
ranges.push(range);
}
}
}
if ranges.is_empty() {
return if saw_valid_spec {
RangeOutcome::Unsatisfiable
} else {
RangeOutcome::Ignored
};
}
ranges.sort_by_key(|range| range.start);
let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
for range in ranges {
match merged.last_mut() {
Some(last) if range.start <= last.end.saturating_add(1) => {
last.end = last.end.max(range.end);
}
_ => merged.push(range),
}
}
if merged.len() > MAX_RANGE_PARTS {
return RangeOutcome::Ignored;
}
RangeOutcome::Satisfiable(merged)
}
enum SingleRange {
Malformed,
Unsatisfiable,
Satisfiable(ByteRange),
}
fn parse_single_range(spec: &str, len: u64) -> SingleRange {
let Some((start_text, end_text)) = spec.split_once('-') else {
return SingleRange::Malformed;
};
let (start_text, end_text) = (start_text.trim(), end_text.trim());
if start_text.is_empty() {
let Ok(suffix) = end_text.parse::<u64>() else {
return SingleRange::Malformed;
};
if suffix == 0 || len == 0 {
return SingleRange::Unsatisfiable;
}
return SingleRange::Satisfiable(ByteRange {
start: len.saturating_sub(suffix),
end: len - 1,
});
}
let Ok(start) = start_text.parse::<u64>() else {
return SingleRange::Malformed;
};
if start >= len {
return SingleRange::Unsatisfiable;
}
let end = if end_text.is_empty() {
len - 1
} else {
match end_text.parse::<u64>() {
Ok(end) => end.min(len - 1),
Err(_) => return SingleRange::Malformed,
}
};
if end < start {
return SingleRange::Unsatisfiable;
}
SingleRange::Satisfiable(ByteRange { start, end })
}
fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
Some(&value[prefix.len()..])
} else {
None
}
}
fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
match modified {
Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
None => format!("\"{len:x}\""),
}
}
fn epoch_secs(time: SystemTime) -> u64 {
time.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
fn epoch_nanos(time: SystemTime) -> u128 {
time.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0)
}
fn is_weak_etag(tag: &str) -> bool {
tag.trim_start().starts_with("W/")
}
fn etag_value(tag: &str) -> &str {
tag.trim().trim_start_matches("W/")
}
fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
if let Some(if_none_match) = req.header("if-none-match") {
let Some(etag) = etag else {
return false;
};
let expected = etag_value(etag);
return if_none_match.split(',').any(|candidate| {
let candidate = candidate.trim();
candidate == "*" || etag_value(candidate) == expected
});
}
if let Some(if_modified_since) = req.header("if-modified-since") {
if let Some(modified) = modified {
if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
return epoch_secs(modified) <= epoch_secs(since);
}
}
}
false
}
fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
let Some(value) = req.header("if-range") else {
return true;
};
let value = value.trim();
if value.starts_with('"') || value.starts_with("W/") {
let Some(etag) = etag else {
return false;
};
if is_weak_etag(value) || is_weak_etag(etag) {
return false;
}
return value == etag.trim();
}
match (httpdate::parse_http_date(value), modified) {
(Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
_ => false,
}
}
fn guess_content_type(path: &Path) -> Option<String> {
mime_guess::from_path(path)
.first()
.map(|mime| mime.essence_str().to_string())
}
fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
let mut flat = HashMap::new();
let mut multi: HashMap<String, Vec<String>> = HashMap::new();
for pair in query.split('&') {
if pair.is_empty() {
continue;
}
let mut parts = pair.splitn(2, '=');
if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
let key = percent_decode(key);
let value = percent_decode(value);
flat.insert(key.clone(), value.clone());
multi.entry(key).or_default().push(value);
}
}
(flat, multi)
}
fn percent_decode(s: &str) -> String {
if !s.contains('%') && !s.contains('+') {
return s.to_string();
}
let mut bytes = Vec::with_capacity(s.len());
let mut chars = s.bytes();
while let Some(byte) = chars.next() {
match byte {
b'%' => {
let hi = chars.next().and_then(|c| (c as char).to_digit(16));
let lo = chars.next().and_then(|c| (c as char).to_digit(16));
match (hi, lo) {
(Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
_ => bytes.push(b'%'),
}
}
b'+' => bytes.push(b' '),
other => bytes.push(other),
}
}
String::from_utf8_lossy(&bytes).into_owned()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_the_boundary_parameter() {
assert_eq!(
multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
Some("----abc123")
);
assert_eq!(
multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
.as_deref(),
Some("quoted")
);
assert_eq!(multipart_boundary("multipart/form-data"), None);
}
fn ranges(start: u64, end: u64) -> RangeOutcome {
RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
}
#[test]
fn parses_byte_ranges() {
assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
}
#[test]
fn rejects_unsatisfiable_and_malformed_ranges() {
assert_eq!(
parse_ranges("bytes=1000-", 1000),
RangeOutcome::Unsatisfiable
);
assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
assert_eq!(
parse_ranges("bytes=500-100", 1000),
RangeOutcome::Unsatisfiable
);
assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
assert_eq!(
parse_ranges("bytes=0-9,abc-def", 1000),
RangeOutcome::Ignored
);
}
#[test]
fn suffix_range_on_empty_file_is_unsatisfiable() {
assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
}
#[test]
fn multi_range_is_coalesced_into_disjoint_parts() {
assert_eq!(
parse_ranges("bytes=20-30,0-10", 1000),
RangeOutcome::Satisfiable(vec![
ByteRange { start: 0, end: 10 },
ByteRange { start: 20, end: 30 },
])
);
assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
assert_eq!(
parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
ranges(0, 999)
);
assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
}
#[test]
fn too_many_parts_falls_back_to_the_full_representation() {
let mut spec = String::from("bytes=");
for index in 0..(MAX_RANGE_PARTS + 1) {
if index > 0 {
spec.push(',');
}
spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
}
assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
}
#[test]
fn etag_changes_with_size_and_mtime() {
let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
assert_ne!(
etag_for(10, Some(base)),
etag_for(10, Some(base + std::time::Duration::from_secs(1)))
);
assert_ne!(
etag_for(10, Some(base)),
etag_for(10, Some(base + std::time::Duration::from_micros(1)))
);
}
#[test]
fn etag_strength_rules() {
assert!(!is_weak_etag("\"abc\""));
assert!(is_weak_etag("W/\"abc\""));
assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
assert_eq!(etag_value("\"abc\""), "\"abc\"");
}
#[test]
fn parses_query_strings() {
let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
assert_eq!(params.get("a").map(String::as_str), Some("1"));
assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
assert_eq!(
multi.get("a").map(Vec::as_slice),
Some(&["1".to_string()][..])
);
}
#[test]
fn keeps_every_query_value_for_repeated_keys() {
let (params, multi) = parse_query_string("id=1&id=2&id=3");
assert_eq!(params.get("id").map(String::as_str), Some("3"));
assert_eq!(
multi.get("id").map(Vec::as_slice),
Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
);
}
#[test]
fn percent_decodes_utf8() {
let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
assert_eq!(params.get("q").map(String::as_str), Some("中文"));
assert_eq!(params.get("x").map(String::as_str), Some("+"));
}
}