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}
132
133impl HttpContext {
134 pub fn new(
140 req: Request<Incoming>,
141 limits: BodyLimits,
142 remote_addr: Option<SocketAddr>,
143 ) -> Self {
144 let (parts, body) = req.into_parts();
145
146 let mut headers = HashMap::new();
147 for (name, value) in parts.headers.iter() {
148 if let Ok(value) = value.to_str() {
149 headers.insert(name.to_string(), value.to_string());
150 }
151 }
152
153 let (query_params, query_multi) = parts
154 .uri
155 .query()
156 .map(parse_query_string)
157 .unwrap_or_default();
158
159 let resp = HttpResponse::new(200);
160
161 let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
170 limits.multipart_body
171 } else {
172 limits.body as u64
173 };
174 let early_error = headers
175 .get("content-length")
176 .and_then(|value| value.parse::<u64>().ok())
177 .filter(|declared| *declared > declared_limit)
178 .map(|_| {
179 Error::PayloadTooLarge(format!(
180 "request body exceeds the {declared_limit}-byte limit"
181 ))
182 });
183
184 let req = HttpRequest {
185 method: parts.method.to_string(),
186 path: parts.uri.path().to_string(),
187 headers,
188 query_params,
189 query_multi,
190 route_params: HashMap::new(),
191 route_pattern: None,
192 remote_addr,
193 body: Mutex::new(RequestBody::Pending(Some(body))),
194 limits,
195 };
196
197 Self {
198 req,
199 resp,
200 claims: None,
201 early_error,
202 }
203 }
204
205 pub async fn into_response(self) -> hyper::Response<RespBody> {
207 let HttpContext { req, resp, .. } = self;
208 finalize_response(&req, resp).await
209 }
210}
211
212impl IClaimsExt for HttpContext {
213 fn set_claims(&mut self, claims: Box<dyn IClaims>) {
214 self.claims = Some(claims);
215 }
216
217 fn claims(&self) -> Option<&dyn IClaims> {
218 self.claims.as_deref()
219 }
220}
221
222impl IHttpContext for HttpContext {
223 fn request(&self) -> &dyn IHttpRequest {
224 &self.req
225 }
226
227 fn request_mut(&mut self) -> &mut dyn IHttpRequest {
228 &mut self.req
229 }
230
231 fn response(&self) -> &dyn IHttpResponse {
232 &self.resp
233 }
234
235 fn response_mut(&mut self) -> &mut dyn IHttpResponse {
236 &mut self.resp
237 }
238
239 fn take_early_error(&mut self) -> Option<Error> {
240 self.early_error.take()
241 }
242}
243
244enum RequestBody {
250 Pending(Option<Incoming>),
252 Buffered(Vec<u8>),
254 Form(Arc<MultipartForm>),
256}
257
258pub struct HttpRequest {
260 method: String,
261 path: String,
262 headers: HashMap<String, String>,
263 query_params: HashMap<String, String>,
264 query_multi: HashMap<String, Vec<String>>,
265 route_params: HashMap<String, String>,
266 route_pattern: Option<String>,
267 remote_addr: Option<SocketAddr>,
268 body: Mutex<RequestBody>,
269 limits: BodyLimits,
270}
271
272#[async_trait::async_trait]
273impl IHttpRequest for HttpRequest {
274 fn method(&self) -> &str {
275 &self.method
276 }
277
278 fn path(&self) -> &str {
279 &self.path
280 }
281
282 fn header(&self, name: &str) -> Option<&str> {
284 if let Some(value) = self.headers.get(name) {
285 return Some(value.as_str());
286 }
287 let lower = name.to_ascii_lowercase();
288 self.headers.get(&lower).map(|s| s.as_str())
289 }
290
291 fn query(&self) -> &HashMap<String, String> {
292 &self.query_params
293 }
294
295 fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
296 Some(&self.query_multi)
297 }
298
299 fn remote_addr(&self) -> Option<SocketAddr> {
300 self.remote_addr
301 }
302
303 fn route_params(&self) -> &HashMap<String, String> {
304 &self.route_params
305 }
306
307 fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
308 &mut self.route_params
309 }
310
311 fn route_pattern(&self) -> Option<&str> {
312 self.route_pattern.as_deref()
313 }
314
315 fn route_pattern_mut(&mut self) -> &mut Option<String> {
316 &mut self.route_pattern
317 }
318
319 async fn body_bytes(&mut self) -> Result<Vec<u8>> {
320 let mut guard = self.body.lock().await;
321 let incoming = match &mut *guard {
322 RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
323 RequestBody::Form(_) => {
324 return Err(Error::State(
325 "request body was already read as multipart/form-data".to_string(),
326 ))
327 }
328 RequestBody::Pending(slot) => slot.take(),
329 };
330
331 let Some(incoming) = incoming else {
332 return Err(Error::State(
333 "request body has already been consumed".to_string(),
334 ));
335 };
336
337 let bytes = read_body(incoming, self.limits.body).await?;
338 *guard = RequestBody::Buffered(bytes.clone());
339 Ok(bytes)
340 }
341
342 async fn body_text(&mut self) -> Result<String> {
343 let bytes = self.body_bytes().await?;
344 String::from_utf8(bytes)
345 .map_err(|_| Error::Validation("request body is not valid UTF-8".to_string()))
346 }
347
348 async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
349 let mut guard = self.body.lock().await;
350
351 match &*guard {
352 RequestBody::Form(form) => return Ok(Arc::clone(form)),
353 RequestBody::Buffered(_) => {
354 return Err(Error::UnsupportedMediaType(
355 "request body was already read as a non-multipart body".to_string(),
356 ))
357 }
358 RequestBody::Pending(_) => {}
359 }
360
361 if !self.is_multipart() {
362 return Err(Error::UnsupportedMediaType(
363 "expected Content-Type: multipart/form-data".to_string(),
364 ));
365 }
366
367 let raw_content_type = self.header("content-type").unwrap_or("");
368 let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
369 Error::Validation(
370 "multipart/form-data request is missing its boundary parameter".to_string(),
371 )
372 })?;
373
374 let incoming = match &mut *guard {
375 RequestBody::Pending(slot) => slot.take(),
376 _ => None,
377 };
378
379 let Some(body) = incoming else {
380 return Err(Error::State(
381 "request body has already been consumed".to_string(),
382 ));
383 };
384
385 let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
386 *guard = RequestBody::Form(Arc::clone(&form));
387 Ok(form)
388 }
389}
390
391async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
393 let mut out = Vec::new();
394 let mut total = 0usize;
395 let mut stream = incoming.into_data_stream();
396
397 while let Some(chunk) = stream.next().await {
398 let chunk = chunk.map_err(|err| {
399 tracing::debug!(error = %err, "failed to read request body");
402 Error::State("failed to read request body".to_string())
403 })?;
404 total = total.saturating_add(chunk.len());
405 if total > limit {
406 return Err(Error::PayloadTooLarge(format!(
407 "request body exceeds the {limit}-byte limit"
408 )));
409 }
410 out.extend_from_slice(&chunk);
411 }
412
413 Ok(out)
414}
415
416fn is_multipart_content_type(content_type: Option<&String>) -> bool {
418 content_type
419 .map(|value| {
420 value
421 .split(';')
422 .next()
423 .unwrap_or("")
424 .trim()
425 .eq_ignore_ascii_case("multipart/form-data")
426 })
427 .unwrap_or(false)
428}
429
430fn multipart_boundary(content_type: &str) -> Option<String> {
432 for part in content_type.split(';').skip(1) {
433 let part = part.trim();
434 let Some((key, value)) = part.split_once('=') else {
435 continue;
436 };
437 if key.trim().eq_ignore_ascii_case("boundary") {
438 let value = value.trim().trim_matches('"');
439 if !value.is_empty() {
440 return Some(value.to_string());
441 }
442 }
443 }
444 None
445}
446
447async fn parse_multipart(
454 incoming: Incoming,
455 boundary: &str,
456 limits: &BodyLimits,
457) -> Result<MultipartForm> {
458 let stream = incoming.into_data_stream();
459 let mut multipart = multer::Multipart::new(stream, boundary);
460 let mut form = MultipartForm::new();
461 let mut total: u64 = 0;
462
463 while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
464 let field_name = field.name().unwrap_or_default().to_string();
465 if field_name.is_empty() {
466 continue;
467 }
468 let file_name = field.file_name().map(str::to_string);
469 let content_type = field.content_type().map(|m| m.to_string());
470
471 match file_name {
472 Some(file_name) => {
473 let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
474 .memory_threshold(limits.memory_threshold)
475 .max_size(limits.file)
476 .spool_dir(limits.spool_dir.clone());
477
478 let mut field = field;
479 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
480 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
481 builder.write(&chunk).await?;
482 }
483 form.push_file(builder.finish().await?);
484 }
485 None => {
486 let mut value: Vec<u8> = Vec::new();
487 let mut field = field;
488 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
489 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
490 if value.len() + chunk.len() > limits.field {
491 return Err(Error::PayloadTooLarge(format!(
492 "form field `{field_name}` exceeds the {}-byte limit",
493 limits.field
494 )));
495 }
496 value.extend_from_slice(&chunk);
497 }
498 let text = String::from_utf8(value).map_err(|_| {
499 Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
500 })?;
501 form.push_field(field_name, text);
502 }
503 }
504 }
505
506 Ok(form)
507}
508
509fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
511 *total = total.saturating_add(len as u64);
512 if *total > limit {
513 return Err(Error::PayloadTooLarge(format!(
514 "multipart request exceeds the {limit}-byte limit"
515 )));
516 }
517 Ok(*total)
518}
519
520fn multer_error(err: multer::Error) -> Error {
521 match err {
522 multer::Error::FieldSizeExceeded { limit, .. } => {
523 Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
524 }
525 multer::Error::StreamSizeExceeded { limit } => {
526 Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
527 }
528 multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
529 Error::Validation("multipart body ended unexpectedly".to_string())
530 }
531 other => Error::Validation(format!("invalid multipart body: {other}")),
532 }
533}
534
535pub struct HttpResponse {
541 status: u16,
542 headers: HashMap<String, String>,
543 body: Option<ResponseBody>,
544}
545
546impl HttpResponse {
547 pub fn new(status: u16) -> Self {
548 Self {
549 status,
550 headers: HashMap::new(),
551 body: None,
552 }
553 }
554}
555
556#[async_trait::async_trait]
557impl IHttpResponse for HttpResponse {
558 fn status(&self) -> u16 {
559 self.status
560 }
561
562 fn has_body(&self) -> bool {
563 self.body.is_some()
564 }
565
566 fn set_status(&mut self, code: u16) {
567 self.status = code;
568 }
569
570 fn set_header(&mut self, key: &str, value: &str) {
571 self.headers.insert(key.to_string(), value.to_string());
572 }
573
574 fn remove_header(&mut self, key: &str) {
575 self.headers.remove(key);
576 }
577
578 async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
579 self.body = Some(ResponseBody::Bytes(data));
580 Ok(())
581 }
582
583 async fn write_text(&mut self, text: &str) -> Result<()> {
584 self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
585 Ok(())
586 }
587
588 async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
589 self.body = Some(body);
590 Ok(())
591 }
592
593 fn body_bytes(&self) -> Vec<u8> {
594 self.body
595 .as_ref()
596 .and_then(ResponseBody::as_bytes)
597 .map(<[u8]>::to_vec)
598 .unwrap_or_default()
599 }
600
601 fn header(&self, key: &str) -> Option<&str> {
602 self.headers.get(key).map(|s| s.as_str())
603 }
604}
605
606async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
611 let HttpResponse {
612 status,
613 mut headers,
614 body,
615 } = resp;
616
617 match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
618 ResponseBody::Bytes(bytes) => {
619 default_cache_control(&mut headers, false);
620 bytes_response(status, headers, bytes)
621 }
622 ResponseBody::File(file) => {
623 default_cache_control(&mut headers, true);
624 file_response(req, status, headers, file).await
625 }
626 }
627}
628
629fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
635 if headers.contains_key("cache-control") {
636 return;
637 }
638 headers.insert(
639 "cache-control".to_string(),
640 if is_file {
641 "public, max-age=0, must-revalidate".to_string()
642 } else {
643 "no-store".to_string()
644 },
645 );
646}
647
648enum PreparedSource {
650 Path(tokio::fs::File),
652 Seekable(Box<dyn SeekableReader>),
654 Stream(Box<dyn AsyncRead + Send + Unpin>),
656}
657
658impl PreparedSource {
659 fn is_seekable(&self) -> bool {
660 matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
661 }
662}
663
664async fn file_response(
669 req: &HttpRequest,
670 status: u16,
671 mut headers: HashMap<String, String>,
672 file: FileBody,
673) -> hyper::Response<RespBody> {
674 let FileBody {
675 source,
676 content_type,
677 file_name,
678 disposition,
679 enable_range_processing,
680 last_modified,
681 entity_tag,
682 } = file;
683
684 let mut modified = last_modified;
686 let mut etag = entity_tag;
687 let mut source_path: Option<PathBuf> = None;
688 let known_len;
689
690 let prepared = match source {
691 FileSource::Path(path) => {
692 let handle = match tokio::fs::File::open(&path).await {
695 Ok(handle) => handle,
696 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
697 return problem_bytes(404, headers, "Not Found")
698 }
699 Err(err) => {
700 tracing::error!(
701 path = %path.display(),
702 error = %err,
703 "cannot open file for download"
704 );
705 return problem_bytes(500, headers, "an unexpected error occurred");
706 }
707 };
708 let metadata = match handle.metadata().await {
709 Ok(metadata) => metadata,
710 Err(err) => {
711 tracing::error!(
712 path = %path.display(),
713 error = %err,
714 "cannot stat file for download"
715 );
716 return problem_bytes(500, headers, "an unexpected error occurred");
717 }
718 };
719 if !metadata.is_file() {
720 return problem_bytes(404, headers, "Not Found");
721 }
722
723 let len = metadata.len();
724 if modified.is_none() {
725 modified = metadata.modified().ok();
726 }
727 if etag.is_none() {
728 etag = Some(etag_for(len, modified));
729 }
730 known_len = Some(len);
731 source_path = Some(path);
732 PreparedSource::Path(handle)
733 }
734 FileSource::SeekableStream { reader, len } => {
735 known_len = Some(len);
736 PreparedSource::Seekable(reader)
737 }
738 FileSource::Stream(reader) => {
739 known_len = None;
740 PreparedSource::Stream(reader)
741 }
742 };
743
744 let modified = modified.filter(|value| *value <= SystemTime::now());
751
752 if !headers.contains_key("content-type") {
754 let resolved = content_type
755 .or_else(|| source_path.as_deref().and_then(guess_content_type))
756 .unwrap_or_else(|| "application/octet-stream".to_string());
757 headers.insert("content-type".to_string(), resolved);
758 }
759 if !headers.contains_key("content-disposition") {
760 if let Some(name) = &file_name {
761 headers.insert(
762 "content-disposition".to_string(),
763 content_disposition_value(disposition, name),
764 );
765 }
766 }
767 if let Some(tag) = &etag {
768 headers.insert("etag".to_string(), tag.clone());
769 }
770 if let Some(modified) = modified {
771 headers.insert(
772 "last-modified".to_string(),
773 httpdate::fmt_http_date(modified),
774 );
775 }
776
777 let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
780 headers.insert(
781 "accept-ranges".to_string(),
782 if range_capable { "bytes" } else { "none" }.to_string(),
783 );
784
785 if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
787 headers.remove("content-length");
788 return bytes_response(304, headers, Vec::new());
789 }
790
791 let mut ranges: Vec<ByteRange> = Vec::new();
793 if status == 200 && range_capable {
794 let len = known_len.expect("range-capable sources know their length");
795 if let Some(raw) = req.header("range") {
796 if if_range_matches(req, etag.as_deref(), modified) {
800 match parse_ranges(raw, len) {
801 RangeOutcome::Ignored => {}
802 RangeOutcome::Unsatisfiable => {
803 headers.insert("content-range".to_string(), format!("bytes */{len}"));
804 return problem_bytes(416, headers, "Range Not Satisfiable");
805 }
806 RangeOutcome::Satisfiable(parsed) => ranges = parsed,
807 }
808 }
809 }
810 }
811
812 if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
819 ranges.clear();
820 }
821 if ranges.len() > 1 {
822 return multipart_byteranges_response(prepared, headers, ranges, known_len);
823 }
824
825 let (status, start, count) = match ranges.first() {
826 Some(range) => {
827 let len = known_len.expect("range-capable sources know their length");
828 headers.insert(
829 "content-range".to_string(),
830 format!("bytes {}-{}/{len}", range.start, range.end),
831 );
832 (206, range.start, range.len())
833 }
834 None => (status, 0, known_len.unwrap_or(0)),
835 };
836
837 match prepared {
838 PreparedSource::Path(mut handle) => {
839 headers.insert("content-length".to_string(), count.to_string());
840 if count == 0 {
841 return bytes_response(status, headers, Vec::new());
842 }
843 if start > 0 {
844 if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
845 tracing::error!(error = %err, "cannot seek file for ranged response");
846 return problem_bytes(500, headers, "an unexpected error occurred");
847 }
848 }
849 build_response(status, headers, reader_body(handle.take(count)))
850 }
851 PreparedSource::Seekable(mut reader) => {
852 headers.insert("content-length".to_string(), count.to_string());
853 if count == 0 {
854 return bytes_response(status, headers, Vec::new());
855 }
856 if start > 0 {
857 if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
858 tracing::error!(error = %err, "cannot seek stream for ranged response");
859 return problem_bytes(500, headers, "an unexpected error occurred");
860 }
861 }
862 build_response(status, headers, reader_body(reader.take(count)))
863 }
864 PreparedSource::Stream(reader) => {
865 build_response(status, headers, reader_body(reader))
868 }
869 }
870}
871
872fn representation_content_type(headers: &HashMap<String, String>) -> String {
877 headers
878 .get("content-type")
879 .filter(|value| !value.starts_with("multipart/byteranges"))
880 .cloned()
881 .unwrap_or_else(|| "application/octet-stream".to_string())
882}
883
884fn multipart_byteranges_response(
891 prepared: PreparedSource,
892 mut headers: HashMap<String, String>,
893 ranges: Vec<ByteRange>,
894 known_len: Option<u64>,
895) -> hyper::Response<RespBody> {
896 let PreparedSource::Path(handle) = prepared else {
897 unreachable!("multi-range is only routed here for path sources");
898 };
899
900 let total_len = known_len.unwrap_or(0);
901 let media_type = representation_content_type(&headers);
902 let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
903
904 let parts: Vec<(Bytes, ByteRange)> = ranges
907 .iter()
908 .map(|range| {
909 let header = format!(
910 "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
911 range.start, range.end
912 );
913 (Bytes::from(header), *range)
914 })
915 .collect();
916 let closing = Bytes::from(format!("--{boundary}--\r\n"));
917
918 let content_length: u64 = parts
919 .iter()
920 .map(|(header, range)| header.len() as u64 + range.len() + 2)
921 .sum::<u64>()
922 + closing.len() as u64;
923
924 headers.insert(
925 "content-type".to_string(),
926 format!("multipart/byteranges; boundary={boundary}"),
927 );
928 headers.insert("content-length".to_string(), content_length.to_string());
929 headers.remove("content-range");
931
932 let stream = multipart_stream(handle, parts, closing);
933 build_response(
934 206,
935 headers,
936 BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
937 )
938}
939
940const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
942
943struct MultipartState {
945 handle: tokio::fs::File,
946 parts: std::vec::IntoIter<(Bytes, ByteRange)>,
947 closing: Bytes,
948 pending_header: Bytes,
950 phase: MultipartPhase,
951 buffer: bytes::BytesMut,
952}
953
954enum MultipartPhase {
955 NextPart,
957 Payload(u64),
959 Terminator,
961 Closing,
963 Done,
964}
965
966fn multipart_stream(
972 handle: tokio::fs::File,
973 parts: Vec<(Bytes, ByteRange)>,
974 closing: Bytes,
975) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
976 let state = MultipartState {
977 handle,
978 parts: parts.into_iter(),
979 closing,
980 pending_header: Bytes::new(),
981 phase: MultipartPhase::NextPart,
982 buffer: bytes::BytesMut::new(),
983 };
984
985 futures_util::stream::unfold(state, |mut state| async move {
986 loop {
987 match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
988 MultipartPhase::NextPart => match state.parts.next() {
989 Some((header, range)) => {
990 if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
991 return Some((Err(err), state));
992 }
993 state.pending_header = header;
994 state.phase = MultipartPhase::Payload(range.len());
995 }
996 None => state.phase = MultipartPhase::Closing,
997 },
998 MultipartPhase::Payload(remaining) => {
999 if remaining == 0 {
1000 state.phase = MultipartPhase::Terminator;
1001 continue;
1002 }
1003
1004 let limit = MULTIPART_CHUNK.min(remaining as usize);
1006 state.buffer.clear();
1007 state.buffer.reserve(limit);
1008 let read = match state.handle.read_buf(&mut state.buffer).await {
1009 Ok(read) => read,
1010 Err(err) => return Some((Err(err), state)),
1011 };
1012 if read == 0 {
1013 return Some((
1014 Err(std::io::Error::new(
1015 std::io::ErrorKind::UnexpectedEof,
1016 "file changed size while serving byte ranges",
1017 )),
1018 state,
1019 ));
1020 }
1021 let keep = read.min(limit);
1022 state.buffer.truncate(keep);
1023 let chunk = state.buffer.split().freeze();
1024
1025 let header = std::mem::take(&mut state.pending_header);
1028 let frame = if header.is_empty() {
1029 chunk
1030 } else {
1031 let mut combined =
1032 bytes::BytesMut::with_capacity(header.len() + chunk.len());
1033 combined.extend_from_slice(&header);
1034 combined.extend_from_slice(&chunk);
1035 combined.freeze()
1036 };
1037
1038 state.phase = MultipartPhase::Payload(remaining - keep as u64);
1039 return Some((Ok(Frame::data(frame)), state));
1040 }
1041 MultipartPhase::Terminator => {
1042 state.phase = MultipartPhase::NextPart;
1043 return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1044 }
1045 MultipartPhase::Closing => {
1046 state.phase = MultipartPhase::Done;
1047 let closing = std::mem::take(&mut state.closing);
1048 return Some((Ok(Frame::data(closing)), state));
1049 }
1050 MultipartPhase::Done => return None,
1051 }
1052 }
1053 })
1054}
1055
1056const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1060
1061fn reader_body<R>(reader: R) -> RespBody
1063where
1064 R: tokio::io::AsyncRead + Send + 'static,
1065{
1066 let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1067 let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1068 BodyExt::boxed_unsync(body)
1069}
1070
1071fn bytes_response(
1072 status: u16,
1073 headers: HashMap<String, String>,
1074 bytes: Vec<u8>,
1075) -> hyper::Response<RespBody> {
1076 build_response(status, headers, full_body(bytes))
1077}
1078
1079fn full_body(bytes: Vec<u8>) -> RespBody {
1080 Full::new(Bytes::from(bytes))
1081 .map_err(|never: std::convert::Infallible| match never {})
1082 .boxed_unsync()
1083}
1084
1085fn problem_bytes(
1087 status: u16,
1088 mut headers: HashMap<String, String>,
1089 detail: &str,
1090) -> hyper::Response<RespBody> {
1091 headers.insert(
1092 "content-type".to_string(),
1093 "application/problem+json".to_string(),
1094 );
1095 bytes_response(
1096 status,
1097 headers,
1098 problem_to_bytes(&build_problem(status, detail)),
1099 )
1100}
1101
1102fn build_response(
1103 status: u16,
1104 headers: HashMap<String, String>,
1105 body: RespBody,
1106) -> hyper::Response<RespBody> {
1107 let mut builder = hyper::Response::builder().status(status);
1108 for (key, value) in &headers {
1109 builder = builder.header(key.as_str(), value.as_str());
1110 }
1111 builder.body(body).unwrap_or_else(|_| {
1112 hyper::Response::builder()
1113 .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1114 .body(full_body(b"Internal Server Error".to_vec()))
1115 .expect("static error response is valid")
1116 })
1117}
1118
1119#[derive(Debug, PartialEq, Eq)]
1121enum RangeOutcome {
1122 Ignored,
1124 Unsatisfiable,
1126 Satisfiable(Vec<ByteRange>),
1128}
1129
1130const MAX_RANGE_PARTS: usize = 8;
1137
1138fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1145 let value = value.trim();
1146 let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1147 return RangeOutcome::Ignored;
1148 };
1149
1150 let mut ranges: Vec<ByteRange> = Vec::new();
1151 let mut saw_valid_spec = false;
1152
1153 for part in spec.split(',') {
1154 let part = part.trim();
1155 if part.is_empty() {
1156 continue;
1157 }
1158 match parse_single_range(part, len) {
1159 SingleRange::Malformed => return RangeOutcome::Ignored,
1161 SingleRange::Unsatisfiable => saw_valid_spec = true,
1162 SingleRange::Satisfiable(range) => {
1163 saw_valid_spec = true;
1164 ranges.push(range);
1165 }
1166 }
1167 }
1168
1169 if ranges.is_empty() {
1170 return if saw_valid_spec {
1171 RangeOutcome::Unsatisfiable
1172 } else {
1173 RangeOutcome::Ignored
1174 };
1175 }
1176
1177 ranges.sort_by_key(|range| range.start);
1178 let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1179 for range in ranges {
1180 match merged.last_mut() {
1181 Some(last) if range.start <= last.end.saturating_add(1) => {
1183 last.end = last.end.max(range.end);
1184 }
1185 _ => merged.push(range),
1186 }
1187 }
1188
1189 if merged.len() > MAX_RANGE_PARTS {
1190 return RangeOutcome::Ignored;
1191 }
1192
1193 RangeOutcome::Satisfiable(merged)
1194}
1195
1196enum SingleRange {
1197 Malformed,
1199 Unsatisfiable,
1201 Satisfiable(ByteRange),
1202}
1203
1204fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1206 let Some((start_text, end_text)) = spec.split_once('-') else {
1207 return SingleRange::Malformed;
1208 };
1209 let (start_text, end_text) = (start_text.trim(), end_text.trim());
1210
1211 if start_text.is_empty() {
1212 let Ok(suffix) = end_text.parse::<u64>() else {
1214 return SingleRange::Malformed;
1215 };
1216 if suffix == 0 || len == 0 {
1217 return SingleRange::Unsatisfiable;
1218 }
1219 return SingleRange::Satisfiable(ByteRange {
1220 start: len.saturating_sub(suffix),
1221 end: len - 1,
1222 });
1223 }
1224
1225 let Ok(start) = start_text.parse::<u64>() else {
1226 return SingleRange::Malformed;
1227 };
1228 if start >= len {
1229 return SingleRange::Unsatisfiable;
1230 }
1231
1232 let end = if end_text.is_empty() {
1233 len - 1
1234 } else {
1235 match end_text.parse::<u64>() {
1236 Ok(end) => end.min(len - 1),
1237 Err(_) => return SingleRange::Malformed,
1238 }
1239 };
1240
1241 if end < start {
1242 return SingleRange::Unsatisfiable;
1243 }
1244
1245 SingleRange::Satisfiable(ByteRange { start, end })
1246}
1247
1248fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1249 if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1250 Some(&value[prefix.len()..])
1251 } else {
1252 None
1253 }
1254}
1255
1256fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1268 match modified {
1269 Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1270 None => format!("\"{len:x}\""),
1271 }
1272}
1273
1274fn epoch_secs(time: SystemTime) -> u64 {
1275 time.duration_since(UNIX_EPOCH)
1276 .map(|d| d.as_secs())
1277 .unwrap_or(0)
1278}
1279
1280fn epoch_nanos(time: SystemTime) -> u128 {
1281 time.duration_since(UNIX_EPOCH)
1282 .map(|d| d.as_nanos())
1283 .unwrap_or(0)
1284}
1285
1286fn is_weak_etag(tag: &str) -> bool {
1288 tag.trim_start().starts_with("W/")
1289}
1290
1291fn etag_value(tag: &str) -> &str {
1293 tag.trim().trim_start_matches("W/")
1294}
1295
1296fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1302 if let Some(if_none_match) = req.header("if-none-match") {
1303 let Some(etag) = etag else {
1304 return false;
1305 };
1306 let expected = etag_value(etag);
1307 return if_none_match.split(',').any(|candidate| {
1308 let candidate = candidate.trim();
1309 candidate == "*" || etag_value(candidate) == expected
1310 });
1311 }
1312
1313 if let Some(if_modified_since) = req.header("if-modified-since") {
1314 if let Some(modified) = modified {
1315 if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1316 return epoch_secs(modified) <= epoch_secs(since);
1317 }
1318 }
1319 }
1320
1321 false
1322}
1323
1324fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1335 let Some(value) = req.header("if-range") else {
1336 return true;
1337 };
1338 let value = value.trim();
1339
1340 if value.starts_with('"') || value.starts_with("W/") {
1341 let Some(etag) = etag else {
1342 return false;
1343 };
1344 if is_weak_etag(value) || is_weak_etag(etag) {
1345 return false;
1346 }
1347 return value == etag.trim();
1348 }
1349
1350 match (httpdate::parse_http_date(value), modified) {
1351 (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1352 _ => false,
1353 }
1354}
1355
1356fn guess_content_type(path: &Path) -> Option<String> {
1357 mime_guess::from_path(path)
1358 .first()
1359 .map(|mime| mime.essence_str().to_string())
1360}
1361
1362fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1372 let mut flat = HashMap::new();
1373 let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1374 for pair in query.split('&') {
1375 if pair.is_empty() {
1376 continue;
1377 }
1378 let mut parts = pair.splitn(2, '=');
1379 if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1380 let key = percent_decode(key);
1381 let value = percent_decode(value);
1382 flat.insert(key.clone(), value.clone());
1383 multi.entry(key).or_default().push(value);
1384 }
1385 }
1386 (flat, multi)
1387}
1388
1389fn percent_decode(s: &str) -> String {
1395 if !s.contains('%') && !s.contains('+') {
1396 return s.to_string();
1397 }
1398
1399 let mut bytes = Vec::with_capacity(s.len());
1400 let mut chars = s.bytes();
1401 while let Some(byte) = chars.next() {
1402 match byte {
1403 b'%' => {
1404 let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1405 let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1406 match (hi, lo) {
1407 (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1408 _ => bytes.push(b'%'),
1409 }
1410 }
1411 b'+' => bytes.push(b' '),
1412 other => bytes.push(other),
1413 }
1414 }
1415 String::from_utf8_lossy(&bytes).into_owned()
1416}
1417
1418#[cfg(test)]
1419mod tests {
1420 use super::*;
1421
1422 #[test]
1423 fn parses_the_boundary_parameter() {
1424 assert_eq!(
1425 multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1426 Some("----abc123")
1427 );
1428 assert_eq!(
1429 multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1430 .as_deref(),
1431 Some("quoted")
1432 );
1433 assert_eq!(multipart_boundary("multipart/form-data"), None);
1434 }
1435
1436 fn ranges(start: u64, end: u64) -> RangeOutcome {
1437 RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1438 }
1439
1440 #[test]
1441 fn parses_byte_ranges() {
1442 assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1443 assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1444 assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1445 assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1447 assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1449 assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1450 }
1451
1452 #[test]
1453 fn rejects_unsatisfiable_and_malformed_ranges() {
1454 assert_eq!(
1455 parse_ranges("bytes=1000-", 1000),
1456 RangeOutcome::Unsatisfiable
1457 );
1458 assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1459 assert_eq!(
1460 parse_ranges("bytes=500-100", 1000),
1461 RangeOutcome::Unsatisfiable
1462 );
1463 assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1464 assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1465 assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1467 assert_eq!(
1468 parse_ranges("bytes=0-9,abc-def", 1000),
1469 RangeOutcome::Ignored
1470 );
1471 }
1472
1473 #[test]
1474 fn suffix_range_on_empty_file_is_unsatisfiable() {
1475 assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1476 }
1477
1478 #[test]
1479 fn multi_range_is_coalesced_into_disjoint_parts() {
1480 assert_eq!(
1482 parse_ranges("bytes=20-30,0-10", 1000),
1483 RangeOutcome::Satisfiable(vec![
1484 ByteRange { start: 0, end: 10 },
1485 ByteRange { start: 20, end: 30 },
1486 ])
1487 );
1488
1489 assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1492
1493 assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1495
1496 assert_eq!(
1498 parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1499 ranges(0, 999)
1500 );
1501
1502 assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1504 }
1505
1506 #[test]
1507 fn too_many_parts_falls_back_to_the_full_representation() {
1508 let mut spec = String::from("bytes=");
1509 for index in 0..(MAX_RANGE_PARTS + 1) {
1510 if index > 0 {
1511 spec.push(',');
1512 }
1513 spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1514 }
1515 assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1516 }
1517
1518 #[test]
1519 fn etag_changes_with_size_and_mtime() {
1520 let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1521 assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1522 assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1523 assert_ne!(
1524 etag_for(10, Some(base)),
1525 etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1526 );
1527 assert_ne!(
1531 etag_for(10, Some(base)),
1532 etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1533 );
1534 }
1535
1536 #[test]
1537 fn etag_strength_rules() {
1538 assert!(!is_weak_etag("\"abc\""));
1539 assert!(is_weak_etag("W/\"abc\""));
1540 assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1541 assert_eq!(etag_value("\"abc\""), "\"abc\"");
1542 }
1543
1544 #[test]
1545 fn parses_query_strings() {
1546 let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1547 assert_eq!(params.get("a").map(String::as_str), Some("1"));
1548 assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1549 assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1550 assert_eq!(
1551 multi.get("a").map(Vec::as_slice),
1552 Some(&["1".to_string()][..])
1553 );
1554 }
1555
1556 #[test]
1557 fn keeps_every_query_value_for_repeated_keys() {
1558 let (params, multi) = parse_query_string("id=1&id=2&id=3");
1559 assert_eq!(params.get("id").map(String::as_str), Some("3"));
1560 assert_eq!(
1561 multi.get("id").map(Vec::as_slice),
1562 Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1563 );
1564 }
1565
1566 #[test]
1567 fn percent_decodes_utf8() {
1568 let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1571 assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1572 assert_eq!(params.get("x").map(String::as_str), Some("+"));
1573 }
1574}