1use futures_util::{StreamExt, TryStreamExt};
26use http_body_util::{combinators::UnsyncBoxBody, BodyExt, Full};
27use hyper::body::{Bytes, Frame, Incoming};
28use hyper::Request;
29use std::collections::HashMap;
30use std::net::SocketAddr;
31use std::path::{Path, PathBuf};
32use std::sync::Arc;
33use std::time::{SystemTime, UNIX_EPOCH};
34use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, SeekFrom};
35use tokio::sync::Mutex;
36use webx_core::auth::IClaims;
37use webx_core::config::AppOptions;
38use webx_core::error::{Error, Result};
39use webx_core::form::{FormFileBuilder, MultipartForm};
40use webx_core::http::{
41 content_disposition_value, ByteRange, FileBody, FileSource, IClaimsExt, IHttpContext,
42 IHttpRequest, IHttpResponse, ResponseBody, SeekableReader,
43};
44
45use crate::problem_response::{build_problem, problem_to_bytes};
46
47pub type RespBody = UnsyncBoxBody<Bytes, std::io::Error>;
53
54#[derive(Debug, Clone)]
58pub struct BodyLimits {
59 pub body: usize,
61 pub multipart_body: u64,
63 pub file: u64,
65 pub field: usize,
67 pub memory_threshold: usize,
69 pub spool_dir: PathBuf,
71}
72
73impl Default for BodyLimits {
74 fn default() -> Self {
75 Self {
76 body: 10 * 1024 * 1024,
77 multipart_body: 256 * 1024 * 1024,
78 file: 128 * 1024 * 1024,
79 field: 1024 * 1024,
80 memory_threshold: 1024 * 1024,
81 spool_dir: webx_core::form::spool_root(),
82 }
83 }
84}
85
86impl BodyLimits {
87 pub fn from_options(options: &AppOptions) -> Self {
92 let spool_dir = match options.form.temp_dir.as_deref() {
93 Some(dir) if !dir.trim().is_empty() => {
94 let path = PathBuf::from(dir);
95 if path.is_absolute() {
96 path
97 } else {
98 webx_core::paths::app_base().join(path)
99 }
100 }
101 _ => webx_core::form::spool_root(),
102 };
103
104 Self {
105 body: options.app.max_body_size,
106 multipart_body: options.form.max_request_size as u64,
107 file: options.form.max_file_size,
108 field: options.form.max_field_size,
109 memory_threshold: options.form.memory_threshold,
110 spool_dir,
111 }
112 }
113
114 pub fn apply_spool_root(&self) {
117 webx_core::form::set_spool_root(self.spool_dir.clone());
118 }
119}
120
121pub struct HttpContext {
123 req: HttpRequest,
124 resp: HttpResponse,
125 claims: Option<Box<dyn IClaims>>,
127 early_error: Option<Error>,
131 dispatch_runtime: Option<Arc<webx_core::DispatchRuntime>>,
133}
134
135impl HttpContext {
136 pub fn new(
142 req: Request<Incoming>,
143 limits: BodyLimits,
144 remote_addr: Option<SocketAddr>,
145 dispatch_runtime: Arc<webx_core::DispatchRuntime>,
146 ) -> Self {
147 let (parts, body) = req.into_parts();
148
149 let mut headers = HashMap::new();
150 for (name, value) in parts.headers.iter() {
151 if let Ok(value) = value.to_str() {
152 headers.insert(name.to_string(), value.to_string());
153 }
154 }
155
156 let (query_params, query_multi) = parts
157 .uri
158 .query()
159 .map(parse_query_string)
160 .unwrap_or_default();
161
162 let resp = HttpResponse::new(200);
163
164 let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
173 limits.multipart_body
174 } else {
175 limits.body as u64
176 };
177 let early_error = headers
178 .get("content-length")
179 .and_then(|value| value.parse::<u64>().ok())
180 .filter(|declared| *declared > declared_limit)
181 .map(|_| {
182 Error::PayloadTooLarge(format!(
183 "request body exceeds the {declared_limit}-byte limit"
184 ))
185 });
186
187 let req = HttpRequest {
188 method: parts.method.to_string(),
189 path: parts.uri.path().to_string(),
190 headers,
191 query_params,
192 query_multi,
193 route_params: HashMap::new(),
194 route_pattern: None,
195 remote_addr,
196 body: Mutex::new(RequestBody::Pending(Some(body))),
197 limits,
198 };
199
200 Self {
201 req,
202 resp,
203 claims: None,
204 early_error,
205 dispatch_runtime: Some(dispatch_runtime),
206 }
207 }
208
209 pub async fn into_response(self) -> hyper::Response<RespBody> {
211 let HttpContext { req, resp, .. } = self;
212 finalize_response(&req, resp).await
213 }
214}
215
216impl IClaimsExt for HttpContext {
217 fn set_claims(&mut self, claims: Box<dyn IClaims>) {
218 self.claims = Some(claims);
219 }
220
221 fn claims(&self) -> Option<&dyn IClaims> {
222 self.claims.as_deref()
223 }
224}
225
226impl IHttpContext for HttpContext {
227 fn request(&self) -> &dyn IHttpRequest {
228 &self.req
229 }
230
231 fn request_mut(&mut self) -> &mut dyn IHttpRequest {
232 &mut self.req
233 }
234
235 fn response(&self) -> &dyn IHttpResponse {
236 &self.resp
237 }
238
239 fn response_mut(&mut self) -> &mut dyn IHttpResponse {
240 &mut self.resp
241 }
242
243 fn take_early_error(&mut self) -> Option<Error> {
244 self.early_error.take()
245 }
246
247 fn dispatch_runtime(&self) -> Option<&Arc<webx_core::DispatchRuntime>> {
248 self.dispatch_runtime.as_ref()
249 }
250}
251
252enum RequestBody {
258 Pending(Option<Incoming>),
260 Buffered(Vec<u8>),
262 Form(Arc<MultipartForm>),
264}
265
266pub struct HttpRequest {
268 method: String,
269 path: String,
270 headers: HashMap<String, String>,
271 query_params: HashMap<String, String>,
272 query_multi: HashMap<String, Vec<String>>,
273 route_params: HashMap<String, String>,
274 route_pattern: Option<String>,
275 remote_addr: Option<SocketAddr>,
276 body: Mutex<RequestBody>,
277 limits: BodyLimits,
278}
279
280#[async_trait::async_trait]
281impl IHttpRequest for HttpRequest {
282 fn method(&self) -> &str {
283 &self.method
284 }
285
286 fn path(&self) -> &str {
287 &self.path
288 }
289
290 fn header(&self, name: &str) -> Option<&str> {
292 if let Some(value) = self.headers.get(name) {
293 return Some(value.as_str());
294 }
295 let lower = name.to_ascii_lowercase();
296 self.headers.get(&lower).map(|s| s.as_str())
297 }
298
299 fn query(&self) -> &HashMap<String, String> {
300 &self.query_params
301 }
302
303 fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
304 Some(&self.query_multi)
305 }
306
307 fn remote_addr(&self) -> Option<SocketAddr> {
308 self.remote_addr
309 }
310
311 fn route_params(&self) -> &HashMap<String, String> {
312 &self.route_params
313 }
314
315 fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
316 &mut self.route_params
317 }
318
319 fn route_pattern(&self) -> Option<&str> {
320 self.route_pattern.as_deref()
321 }
322
323 fn route_pattern_mut(&mut self) -> &mut Option<String> {
324 &mut self.route_pattern
325 }
326
327 async fn body_bytes(&mut self) -> Result<Vec<u8>> {
328 let mut guard = self.body.lock().await;
329 let incoming = match &mut *guard {
330 RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
331 RequestBody::Form(_) => {
332 return Err(Error::State(
333 "request body was already read as multipart/form-data".to_string(),
334 ))
335 }
336 RequestBody::Pending(slot) => slot.take(),
337 };
338
339 let Some(incoming) = incoming else {
340 return Err(Error::State(
341 "request body has already been consumed".to_string(),
342 ));
343 };
344
345 let bytes = read_body(incoming, self.limits.body).await?;
346 *guard = RequestBody::Buffered(bytes.clone());
347 Ok(bytes)
348 }
349
350 async fn body_text(&mut self) -> Result<String> {
351 let bytes = self.body_bytes().await?;
352 String::from_utf8(bytes)
353 .map_err(|_| Error::Validation("request body is not valid UTF-8".to_string()))
354 }
355
356 async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
357 let mut guard = self.body.lock().await;
358
359 match &*guard {
360 RequestBody::Form(form) => return Ok(Arc::clone(form)),
361 RequestBody::Buffered(_) => {
362 return Err(Error::UnsupportedMediaType(
363 "request body was already read as a non-multipart body".to_string(),
364 ))
365 }
366 RequestBody::Pending(_) => {}
367 }
368
369 if !self.is_multipart() {
370 return Err(Error::UnsupportedMediaType(
371 "expected Content-Type: multipart/form-data".to_string(),
372 ));
373 }
374
375 let raw_content_type = self.header("content-type").unwrap_or("");
376 let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
377 Error::Validation(
378 "multipart/form-data request is missing its boundary parameter".to_string(),
379 )
380 })?;
381
382 let incoming = match &mut *guard {
383 RequestBody::Pending(slot) => slot.take(),
384 _ => None,
385 };
386
387 let Some(body) = incoming else {
388 return Err(Error::State(
389 "request body has already been consumed".to_string(),
390 ));
391 };
392
393 let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
394 *guard = RequestBody::Form(Arc::clone(&form));
395 Ok(form)
396 }
397}
398
399async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
401 let mut out = Vec::new();
402 let mut total = 0usize;
403 let mut stream = incoming.into_data_stream();
404
405 while let Some(chunk) = stream.next().await {
406 let chunk = chunk.map_err(|err| {
407 tracing::debug!(error = %err, "failed to read request body");
410 Error::State("failed to read request body".to_string())
411 })?;
412 total = total.saturating_add(chunk.len());
413 if total > limit {
414 return Err(Error::PayloadTooLarge(format!(
415 "request body exceeds the {limit}-byte limit"
416 )));
417 }
418 out.extend_from_slice(&chunk);
419 }
420
421 Ok(out)
422}
423
424fn is_multipart_content_type(content_type: Option<&String>) -> bool {
426 content_type
427 .map(|value| {
428 value
429 .split(';')
430 .next()
431 .unwrap_or("")
432 .trim()
433 .eq_ignore_ascii_case("multipart/form-data")
434 })
435 .unwrap_or(false)
436}
437
438fn multipart_boundary(content_type: &str) -> Option<String> {
440 for part in content_type.split(';').skip(1) {
441 let part = part.trim();
442 let Some((key, value)) = part.split_once('=') else {
443 continue;
444 };
445 if key.trim().eq_ignore_ascii_case("boundary") {
446 let value = value.trim().trim_matches('"');
447 if !value.is_empty() {
448 return Some(value.to_string());
449 }
450 }
451 }
452 None
453}
454
455async fn parse_multipart(
462 incoming: Incoming,
463 boundary: &str,
464 limits: &BodyLimits,
465) -> Result<MultipartForm> {
466 let stream = incoming.into_data_stream();
467 let mut multipart = multer::Multipart::new(stream, boundary);
468 let mut form = MultipartForm::new();
469 let mut total: u64 = 0;
470
471 while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
472 let field_name = field.name().unwrap_or_default().to_string();
473 if field_name.is_empty() {
474 continue;
475 }
476 let file_name = field.file_name().map(str::to_string);
477 let content_type = field.content_type().map(|m| m.to_string());
478
479 match file_name {
480 Some(file_name) => {
481 let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
482 .memory_threshold(limits.memory_threshold)
483 .max_size(limits.file)
484 .spool_dir(limits.spool_dir.clone());
485
486 let mut field = field;
487 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
488 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
489 builder.write(&chunk).await?;
490 }
491 form.push_file(builder.finish().await?);
492 }
493 None => {
494 let mut value: Vec<u8> = Vec::new();
495 let mut field = field;
496 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
497 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
498 if value.len() + chunk.len() > limits.field {
499 return Err(Error::PayloadTooLarge(format!(
500 "form field `{field_name}` exceeds the {}-byte limit",
501 limits.field
502 )));
503 }
504 value.extend_from_slice(&chunk);
505 }
506 let text = String::from_utf8(value).map_err(|_| {
507 Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
508 })?;
509 form.push_field(field_name, text);
510 }
511 }
512 }
513
514 Ok(form)
515}
516
517fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
519 *total = total.saturating_add(len as u64);
520 if *total > limit {
521 return Err(Error::PayloadTooLarge(format!(
522 "multipart request exceeds the {limit}-byte limit"
523 )));
524 }
525 Ok(*total)
526}
527
528fn multer_error(err: multer::Error) -> Error {
529 match err {
530 multer::Error::FieldSizeExceeded { limit, .. } => {
531 Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
532 }
533 multer::Error::StreamSizeExceeded { limit } => {
534 Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
535 }
536 multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
537 Error::Validation("multipart body ended unexpectedly".to_string())
538 }
539 other => Error::Validation(format!("invalid multipart body: {other}")),
540 }
541}
542
543pub struct HttpResponse {
549 status: u16,
550 headers: HashMap<String, String>,
551 body: Option<ResponseBody>,
552}
553
554impl HttpResponse {
555 pub fn new(status: u16) -> Self {
556 Self {
557 status,
558 headers: HashMap::new(),
559 body: None,
560 }
561 }
562}
563
564#[async_trait::async_trait]
565impl IHttpResponse for HttpResponse {
566 fn status(&self) -> u16 {
567 self.status
568 }
569
570 fn has_body(&self) -> bool {
571 self.body.is_some()
572 }
573
574 fn set_status(&mut self, code: u16) {
575 self.status = code;
576 }
577
578 fn set_header(&mut self, key: &str, value: &str) {
579 self.headers
583 .insert(key.to_ascii_lowercase(), value.to_string());
584 }
585
586 fn remove_header(&mut self, key: &str) {
587 self.headers.remove(&key.to_ascii_lowercase());
588 }
589
590 async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
591 self.body = Some(ResponseBody::Bytes(data));
592 Ok(())
593 }
594
595 async fn write_text(&mut self, text: &str) -> Result<()> {
596 self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
597 Ok(())
598 }
599
600 async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
601 self.body = Some(body);
602 Ok(())
603 }
604
605 fn body_bytes(&self) -> Vec<u8> {
606 self.body
607 .as_ref()
608 .and_then(ResponseBody::as_bytes)
609 .map(<[u8]>::to_vec)
610 .unwrap_or_default()
611 }
612
613 fn header(&self, key: &str) -> Option<&str> {
614 self.headers
615 .get(&key.to_ascii_lowercase())
616 .map(|s| s.as_str())
617 }
618}
619
620async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
625 let HttpResponse {
626 status,
627 mut headers,
628 body,
629 } = resp;
630
631 match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
632 ResponseBody::Bytes(bytes) => {
633 default_cache_control(&mut headers, false);
634 bytes_response(status, headers, bytes)
635 }
636 ResponseBody::File(file) => {
637 default_cache_control(&mut headers, true);
638 file_response(req, status, headers, file).await
639 }
640 }
641}
642
643fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
649 if headers.contains_key("cache-control") {
650 return;
651 }
652 headers.insert(
653 "cache-control".to_string(),
654 if is_file {
655 "public, max-age=0, must-revalidate".to_string()
656 } else {
657 "no-store".to_string()
658 },
659 );
660}
661
662enum PreparedSource {
664 Path(tokio::fs::File),
666 Seekable(Box<dyn SeekableReader>),
668 Stream(Box<dyn AsyncRead + Send + Unpin>),
670}
671
672impl PreparedSource {
673 fn is_seekable(&self) -> bool {
674 matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
675 }
676}
677
678async fn file_response(
683 req: &HttpRequest,
684 status: u16,
685 mut headers: HashMap<String, String>,
686 file: FileBody,
687) -> hyper::Response<RespBody> {
688 let FileBody {
689 source,
690 content_type,
691 file_name,
692 disposition,
693 enable_range_processing,
694 last_modified,
695 entity_tag,
696 } = file;
697
698 let mut modified = last_modified;
700 let mut etag = entity_tag;
701 let mut source_path: Option<PathBuf> = None;
702 let known_len;
703
704 let prepared = match source {
705 FileSource::Path(path) => {
706 let handle = match tokio::fs::File::open(&path).await {
709 Ok(handle) => handle,
710 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
711 return problem_bytes(404, headers, "Not Found")
712 }
713 Err(err) => {
714 tracing::error!(
715 path = %path.display(),
716 error = %err,
717 "cannot open file for download"
718 );
719 return problem_bytes(500, headers, "an unexpected error occurred");
720 }
721 };
722 let metadata = match handle.metadata().await {
723 Ok(metadata) => metadata,
724 Err(err) => {
725 tracing::error!(
726 path = %path.display(),
727 error = %err,
728 "cannot stat file for download"
729 );
730 return problem_bytes(500, headers, "an unexpected error occurred");
731 }
732 };
733 if !metadata.is_file() {
734 return problem_bytes(404, headers, "Not Found");
735 }
736
737 let len = metadata.len();
738 if modified.is_none() {
739 modified = metadata.modified().ok();
740 }
741 if etag.is_none() {
742 etag = Some(etag_for(len, modified));
743 }
744 known_len = Some(len);
745 source_path = Some(path);
746 PreparedSource::Path(handle)
747 }
748 FileSource::SeekableStream { reader, len } => {
749 known_len = Some(len);
750 PreparedSource::Seekable(reader)
751 }
752 FileSource::Stream(reader) => {
753 known_len = None;
754 PreparedSource::Stream(reader)
755 }
756 };
757
758 let modified = modified.filter(|value| *value <= SystemTime::now());
765
766 if !headers.contains_key("content-type") {
768 let resolved = content_type
769 .or_else(|| source_path.as_deref().and_then(guess_content_type))
770 .unwrap_or_else(|| "application/octet-stream".to_string());
771 headers.insert("content-type".to_string(), resolved);
772 }
773 if !headers.contains_key("content-disposition") {
774 if let Some(name) = &file_name {
775 headers.insert(
776 "content-disposition".to_string(),
777 content_disposition_value(disposition, name),
778 );
779 }
780 }
781 if let Some(tag) = &etag {
782 headers.insert("etag".to_string(), tag.clone());
783 }
784 if let Some(modified) = modified {
785 headers.insert(
786 "last-modified".to_string(),
787 httpdate::fmt_http_date(modified),
788 );
789 }
790
791 let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
794 headers.insert(
795 "accept-ranges".to_string(),
796 if range_capable { "bytes" } else { "none" }.to_string(),
797 );
798
799 if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
801 headers.remove("content-length");
802 return bytes_response(304, headers, Vec::new());
803 }
804
805 let mut ranges: Vec<ByteRange> = Vec::new();
807 if status == 200 && range_capable {
808 let len = known_len.expect("range-capable sources know their length");
809 if let Some(raw) = req.header("range") {
810 if if_range_matches(req, etag.as_deref(), modified) {
814 match parse_ranges(raw, len) {
815 RangeOutcome::Ignored => {}
816 RangeOutcome::Unsatisfiable => {
817 headers.insert("content-range".to_string(), format!("bytes */{len}"));
818 return problem_bytes(416, headers, "Range Not Satisfiable");
819 }
820 RangeOutcome::Satisfiable(parsed) => ranges = parsed,
821 }
822 }
823 }
824 }
825
826 if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
833 ranges.clear();
834 }
835 if ranges.len() > 1 {
836 return multipart_byteranges_response(prepared, headers, ranges, known_len);
837 }
838
839 let (status, start, count) = match ranges.first() {
840 Some(range) => {
841 let len = known_len.expect("range-capable sources know their length");
842 headers.insert(
843 "content-range".to_string(),
844 format!("bytes {}-{}/{len}", range.start, range.end),
845 );
846 (206, range.start, range.len())
847 }
848 None => (status, 0, known_len.unwrap_or(0)),
849 };
850
851 match prepared {
852 PreparedSource::Path(mut handle) => {
853 headers.insert("content-length".to_string(), count.to_string());
854 if count == 0 {
855 return bytes_response(status, headers, Vec::new());
856 }
857 if start > 0 {
858 if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
859 tracing::error!(error = %err, "cannot seek file for ranged response");
860 return problem_bytes(500, headers, "an unexpected error occurred");
861 }
862 }
863 build_response(status, headers, reader_body(handle.take(count)))
864 }
865 PreparedSource::Seekable(mut reader) => {
866 headers.insert("content-length".to_string(), count.to_string());
867 if count == 0 {
868 return bytes_response(status, headers, Vec::new());
869 }
870 if start > 0 {
871 if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
872 tracing::error!(error = %err, "cannot seek stream for ranged response");
873 return problem_bytes(500, headers, "an unexpected error occurred");
874 }
875 }
876 build_response(status, headers, reader_body(reader.take(count)))
877 }
878 PreparedSource::Stream(reader) => {
879 build_response(status, headers, reader_body(reader))
882 }
883 }
884}
885
886fn representation_content_type(headers: &HashMap<String, String>) -> String {
891 headers
892 .get("content-type")
893 .filter(|value| !value.starts_with("multipart/byteranges"))
894 .cloned()
895 .unwrap_or_else(|| "application/octet-stream".to_string())
896}
897
898fn multipart_byteranges_response(
905 prepared: PreparedSource,
906 mut headers: HashMap<String, String>,
907 ranges: Vec<ByteRange>,
908 known_len: Option<u64>,
909) -> hyper::Response<RespBody> {
910 let PreparedSource::Path(handle) = prepared else {
911 unreachable!("multi-range is only routed here for path sources");
912 };
913
914 let total_len = known_len.unwrap_or(0);
915 let media_type = representation_content_type(&headers);
916 let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
917
918 let parts: Vec<(Bytes, ByteRange)> = ranges
921 .iter()
922 .map(|range| {
923 let header = format!(
924 "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
925 range.start, range.end
926 );
927 (Bytes::from(header), *range)
928 })
929 .collect();
930 let closing = Bytes::from(format!("--{boundary}--\r\n"));
931
932 let content_length: u64 = parts
933 .iter()
934 .map(|(header, range)| header.len() as u64 + range.len() + 2)
935 .sum::<u64>()
936 + closing.len() as u64;
937
938 headers.insert(
939 "content-type".to_string(),
940 format!("multipart/byteranges; boundary={boundary}"),
941 );
942 headers.insert("content-length".to_string(), content_length.to_string());
943 headers.remove("content-range");
945
946 let stream = multipart_stream(handle, parts, closing);
947 build_response(
948 206,
949 headers,
950 BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
951 )
952}
953
954const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
956
957struct MultipartState {
959 handle: tokio::fs::File,
960 parts: std::vec::IntoIter<(Bytes, ByteRange)>,
961 closing: Bytes,
962 pending_header: Bytes,
964 phase: MultipartPhase,
965 buffer: bytes::BytesMut,
966}
967
968enum MultipartPhase {
969 NextPart,
971 Payload(u64),
973 Terminator,
975 Closing,
977 Done,
978}
979
980fn multipart_stream(
986 handle: tokio::fs::File,
987 parts: Vec<(Bytes, ByteRange)>,
988 closing: Bytes,
989) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
990 let state = MultipartState {
991 handle,
992 parts: parts.into_iter(),
993 closing,
994 pending_header: Bytes::new(),
995 phase: MultipartPhase::NextPart,
996 buffer: bytes::BytesMut::new(),
997 };
998
999 futures_util::stream::unfold(state, |mut state| async move {
1000 loop {
1001 match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
1002 MultipartPhase::NextPart => match state.parts.next() {
1003 Some((header, range)) => {
1004 if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
1005 return Some((Err(err), state));
1006 }
1007 state.pending_header = header;
1008 state.phase = MultipartPhase::Payload(range.len());
1009 }
1010 None => state.phase = MultipartPhase::Closing,
1011 },
1012 MultipartPhase::Payload(remaining) => {
1013 if remaining == 0 {
1014 state.phase = MultipartPhase::Terminator;
1015 continue;
1016 }
1017
1018 let limit = MULTIPART_CHUNK.min(remaining as usize);
1020 state.buffer.clear();
1021 state.buffer.reserve(limit);
1022 let read = match state.handle.read_buf(&mut state.buffer).await {
1023 Ok(read) => read,
1024 Err(err) => return Some((Err(err), state)),
1025 };
1026 if read == 0 {
1027 return Some((
1028 Err(std::io::Error::new(
1029 std::io::ErrorKind::UnexpectedEof,
1030 "file changed size while serving byte ranges",
1031 )),
1032 state,
1033 ));
1034 }
1035 let keep = read.min(limit);
1036 state.buffer.truncate(keep);
1037 let chunk = state.buffer.split().freeze();
1038
1039 let header = std::mem::take(&mut state.pending_header);
1042 let frame = if header.is_empty() {
1043 chunk
1044 } else {
1045 let mut combined =
1046 bytes::BytesMut::with_capacity(header.len() + chunk.len());
1047 combined.extend_from_slice(&header);
1048 combined.extend_from_slice(&chunk);
1049 combined.freeze()
1050 };
1051
1052 state.phase = MultipartPhase::Payload(remaining - keep as u64);
1053 return Some((Ok(Frame::data(frame)), state));
1054 }
1055 MultipartPhase::Terminator => {
1056 state.phase = MultipartPhase::NextPart;
1057 return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1058 }
1059 MultipartPhase::Closing => {
1060 state.phase = MultipartPhase::Done;
1061 let closing = std::mem::take(&mut state.closing);
1062 return Some((Ok(Frame::data(closing)), state));
1063 }
1064 MultipartPhase::Done => return None,
1065 }
1066 }
1067 })
1068}
1069
1070const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1074
1075fn reader_body<R>(reader: R) -> RespBody
1077where
1078 R: tokio::io::AsyncRead + Send + 'static,
1079{
1080 let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1081 let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1082 BodyExt::boxed_unsync(body)
1083}
1084
1085fn bytes_response(
1086 status: u16,
1087 headers: HashMap<String, String>,
1088 bytes: Vec<u8>,
1089) -> hyper::Response<RespBody> {
1090 build_response(status, headers, full_body(bytes))
1091}
1092
1093fn full_body(bytes: Vec<u8>) -> RespBody {
1094 Full::new(Bytes::from(bytes))
1095 .map_err(|never: std::convert::Infallible| match never {})
1096 .boxed_unsync()
1097}
1098
1099fn problem_bytes(
1101 status: u16,
1102 mut headers: HashMap<String, String>,
1103 detail: &str,
1104) -> hyper::Response<RespBody> {
1105 headers.insert(
1106 "content-type".to_string(),
1107 "application/problem+json".to_string(),
1108 );
1109 bytes_response(
1110 status,
1111 headers,
1112 problem_to_bytes(&build_problem(status, detail)),
1113 )
1114}
1115
1116fn build_response(
1117 status: u16,
1118 headers: HashMap<String, String>,
1119 body: RespBody,
1120) -> hyper::Response<RespBody> {
1121 let mut builder = hyper::Response::builder().status(status);
1122 for (key, value) in &headers {
1123 builder = builder.header(key.as_str(), value.as_str());
1124 }
1125 builder.body(body).unwrap_or_else(|_| {
1126 hyper::Response::builder()
1127 .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1128 .body(full_body(b"Internal Server Error".to_vec()))
1129 .expect("static error response is valid")
1130 })
1131}
1132
1133#[derive(Debug, PartialEq, Eq)]
1135enum RangeOutcome {
1136 Ignored,
1138 Unsatisfiable,
1140 Satisfiable(Vec<ByteRange>),
1142}
1143
1144const MAX_RANGE_PARTS: usize = 8;
1151
1152fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1159 let value = value.trim();
1160 let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1161 return RangeOutcome::Ignored;
1162 };
1163
1164 let mut ranges: Vec<ByteRange> = Vec::new();
1165 let mut saw_valid_spec = false;
1166
1167 for part in spec.split(',') {
1168 let part = part.trim();
1169 if part.is_empty() {
1170 continue;
1171 }
1172 match parse_single_range(part, len) {
1173 SingleRange::Malformed => return RangeOutcome::Ignored,
1175 SingleRange::Unsatisfiable => saw_valid_spec = true,
1176 SingleRange::Satisfiable(range) => {
1177 saw_valid_spec = true;
1178 ranges.push(range);
1179 }
1180 }
1181 }
1182
1183 if ranges.is_empty() {
1184 return if saw_valid_spec {
1185 RangeOutcome::Unsatisfiable
1186 } else {
1187 RangeOutcome::Ignored
1188 };
1189 }
1190
1191 ranges.sort_by_key(|range| range.start);
1192 let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1193 for range in ranges {
1194 match merged.last_mut() {
1195 Some(last) if range.start <= last.end.saturating_add(1) => {
1197 last.end = last.end.max(range.end);
1198 }
1199 _ => merged.push(range),
1200 }
1201 }
1202
1203 if merged.len() > MAX_RANGE_PARTS {
1204 return RangeOutcome::Ignored;
1205 }
1206
1207 RangeOutcome::Satisfiable(merged)
1208}
1209
1210enum SingleRange {
1211 Malformed,
1213 Unsatisfiable,
1215 Satisfiable(ByteRange),
1216}
1217
1218fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1220 let Some((start_text, end_text)) = spec.split_once('-') else {
1221 return SingleRange::Malformed;
1222 };
1223 let (start_text, end_text) = (start_text.trim(), end_text.trim());
1224
1225 if start_text.is_empty() {
1226 let Ok(suffix) = end_text.parse::<u64>() else {
1228 return SingleRange::Malformed;
1229 };
1230 if suffix == 0 || len == 0 {
1231 return SingleRange::Unsatisfiable;
1232 }
1233 return SingleRange::Satisfiable(ByteRange {
1234 start: len.saturating_sub(suffix),
1235 end: len - 1,
1236 });
1237 }
1238
1239 let Ok(start) = start_text.parse::<u64>() else {
1240 return SingleRange::Malformed;
1241 };
1242 if start >= len {
1243 return SingleRange::Unsatisfiable;
1244 }
1245
1246 let end = if end_text.is_empty() {
1247 len - 1
1248 } else {
1249 match end_text.parse::<u64>() {
1250 Ok(end) => end.min(len - 1),
1251 Err(_) => return SingleRange::Malformed,
1252 }
1253 };
1254
1255 if end < start {
1256 return SingleRange::Unsatisfiable;
1257 }
1258
1259 SingleRange::Satisfiable(ByteRange { start, end })
1260}
1261
1262fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1263 if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1264 Some(&value[prefix.len()..])
1265 } else {
1266 None
1267 }
1268}
1269
1270fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1282 match modified {
1283 Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1284 None => format!("\"{len:x}\""),
1285 }
1286}
1287
1288fn epoch_secs(time: SystemTime) -> u64 {
1289 time.duration_since(UNIX_EPOCH)
1290 .map(|d| d.as_secs())
1291 .unwrap_or(0)
1292}
1293
1294fn epoch_nanos(time: SystemTime) -> u128 {
1295 time.duration_since(UNIX_EPOCH)
1296 .map(|d| d.as_nanos())
1297 .unwrap_or(0)
1298}
1299
1300fn is_weak_etag(tag: &str) -> bool {
1302 tag.trim_start().starts_with("W/")
1303}
1304
1305fn etag_value(tag: &str) -> &str {
1307 tag.trim().trim_start_matches("W/")
1308}
1309
1310fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1316 if let Some(if_none_match) = req.header("if-none-match") {
1317 let Some(etag) = etag else {
1318 return false;
1319 };
1320 let expected = etag_value(etag);
1321 return if_none_match.split(',').any(|candidate| {
1322 let candidate = candidate.trim();
1323 candidate == "*" || etag_value(candidate) == expected
1324 });
1325 }
1326
1327 if let Some(if_modified_since) = req.header("if-modified-since") {
1328 if let Some(modified) = modified {
1329 if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1330 return epoch_secs(modified) <= epoch_secs(since);
1331 }
1332 }
1333 }
1334
1335 false
1336}
1337
1338fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1349 let Some(value) = req.header("if-range") else {
1350 return true;
1351 };
1352 let value = value.trim();
1353
1354 if value.starts_with('"') || value.starts_with("W/") {
1355 let Some(etag) = etag else {
1356 return false;
1357 };
1358 if is_weak_etag(value) || is_weak_etag(etag) {
1359 return false;
1360 }
1361 return value == etag.trim();
1362 }
1363
1364 match (httpdate::parse_http_date(value), modified) {
1365 (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1366 _ => false,
1367 }
1368}
1369
1370fn guess_content_type(path: &Path) -> Option<String> {
1371 mime_guess::from_path(path)
1372 .first()
1373 .map(|mime| mime.essence_str().to_string())
1374}
1375
1376fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1386 let mut flat = HashMap::new();
1387 let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1388 for pair in query.split('&') {
1389 if pair.is_empty() {
1390 continue;
1391 }
1392 let mut parts = pair.splitn(2, '=');
1393 if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1394 let key = percent_decode(key);
1395 let value = percent_decode(value);
1396 flat.insert(key.clone(), value.clone());
1397 multi.entry(key).or_default().push(value);
1398 }
1399 }
1400 (flat, multi)
1401}
1402
1403fn percent_decode(s: &str) -> String {
1409 if !s.contains('%') && !s.contains('+') {
1410 return s.to_string();
1411 }
1412
1413 let mut bytes = Vec::with_capacity(s.len());
1414 let mut chars = s.bytes();
1415 while let Some(byte) = chars.next() {
1416 match byte {
1417 b'%' => {
1418 let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1419 let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1420 match (hi, lo) {
1421 (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1422 _ => bytes.push(b'%'),
1423 }
1424 }
1425 b'+' => bytes.push(b' '),
1426 other => bytes.push(other),
1427 }
1428 }
1429 String::from_utf8_lossy(&bytes).into_owned()
1430}
1431
1432#[cfg(test)]
1433mod tests {
1434 use super::*;
1435
1436 #[test]
1437 fn parses_the_boundary_parameter() {
1438 assert_eq!(
1439 multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1440 Some("----abc123")
1441 );
1442 assert_eq!(
1443 multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1444 .as_deref(),
1445 Some("quoted")
1446 );
1447 assert_eq!(multipart_boundary("multipart/form-data"), None);
1448 }
1449
1450 fn ranges(start: u64, end: u64) -> RangeOutcome {
1451 RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1452 }
1453
1454 #[test]
1455 fn parses_byte_ranges() {
1456 assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1457 assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1458 assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1459 assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1461 assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1463 assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1464 }
1465
1466 #[test]
1467 fn rejects_unsatisfiable_and_malformed_ranges() {
1468 assert_eq!(
1469 parse_ranges("bytes=1000-", 1000),
1470 RangeOutcome::Unsatisfiable
1471 );
1472 assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1473 assert_eq!(
1474 parse_ranges("bytes=500-100", 1000),
1475 RangeOutcome::Unsatisfiable
1476 );
1477 assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1478 assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1479 assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1481 assert_eq!(
1482 parse_ranges("bytes=0-9,abc-def", 1000),
1483 RangeOutcome::Ignored
1484 );
1485 }
1486
1487 #[test]
1488 fn suffix_range_on_empty_file_is_unsatisfiable() {
1489 assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1490 }
1491
1492 #[test]
1493 fn multi_range_is_coalesced_into_disjoint_parts() {
1494 assert_eq!(
1496 parse_ranges("bytes=20-30,0-10", 1000),
1497 RangeOutcome::Satisfiable(vec![
1498 ByteRange { start: 0, end: 10 },
1499 ByteRange { start: 20, end: 30 },
1500 ])
1501 );
1502
1503 assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1506
1507 assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1509
1510 assert_eq!(
1512 parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1513 ranges(0, 999)
1514 );
1515
1516 assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1518 }
1519
1520 #[test]
1521 fn too_many_parts_falls_back_to_the_full_representation() {
1522 let mut spec = String::from("bytes=");
1523 for index in 0..(MAX_RANGE_PARTS + 1) {
1524 if index > 0 {
1525 spec.push(',');
1526 }
1527 spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1528 }
1529 assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1530 }
1531
1532 #[test]
1533 fn etag_changes_with_size_and_mtime() {
1534 let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1535 assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1536 assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1537 assert_ne!(
1538 etag_for(10, Some(base)),
1539 etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1540 );
1541 assert_ne!(
1545 etag_for(10, Some(base)),
1546 etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1547 );
1548 }
1549
1550 #[test]
1551 fn etag_strength_rules() {
1552 assert!(!is_weak_etag("\"abc\""));
1553 assert!(is_weak_etag("W/\"abc\""));
1554 assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1555 assert_eq!(etag_value("\"abc\""), "\"abc\"");
1556 }
1557
1558 #[test]
1559 fn parses_query_strings() {
1560 let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1561 assert_eq!(params.get("a").map(String::as_str), Some("1"));
1562 assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1563 assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1564 assert_eq!(
1565 multi.get("a").map(Vec::as_slice),
1566 Some(&["1".to_string()][..])
1567 );
1568 }
1569
1570 #[test]
1571 fn keeps_every_query_value_for_repeated_keys() {
1572 let (params, multi) = parse_query_string("id=1&id=2&id=3");
1573 assert_eq!(params.get("id").map(String::as_str), Some("3"));
1574 assert_eq!(
1575 multi.get("id").map(Vec::as_slice),
1576 Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1577 );
1578 }
1579
1580 #[test]
1581 fn percent_decodes_utf8() {
1582 let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1585 assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1586 assert_eq!(params.get("x").map(String::as_str), Some("+"));
1587 }
1588}