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.insert(key.to_string(), value.to_string());
580 }
581
582 fn remove_header(&mut self, key: &str) {
583 self.headers.remove(key);
584 }
585
586 async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
587 self.body = Some(ResponseBody::Bytes(data));
588 Ok(())
589 }
590
591 async fn write_text(&mut self, text: &str) -> Result<()> {
592 self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
593 Ok(())
594 }
595
596 async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
597 self.body = Some(body);
598 Ok(())
599 }
600
601 fn body_bytes(&self) -> Vec<u8> {
602 self.body
603 .as_ref()
604 .and_then(ResponseBody::as_bytes)
605 .map(<[u8]>::to_vec)
606 .unwrap_or_default()
607 }
608
609 fn header(&self, key: &str) -> Option<&str> {
610 self.headers.get(key).map(|s| s.as_str())
611 }
612}
613
614async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
619 let HttpResponse {
620 status,
621 mut headers,
622 body,
623 } = resp;
624
625 match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
626 ResponseBody::Bytes(bytes) => {
627 default_cache_control(&mut headers, false);
628 bytes_response(status, headers, bytes)
629 }
630 ResponseBody::File(file) => {
631 default_cache_control(&mut headers, true);
632 file_response(req, status, headers, file).await
633 }
634 }
635}
636
637fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
643 if headers.contains_key("cache-control") {
644 return;
645 }
646 headers.insert(
647 "cache-control".to_string(),
648 if is_file {
649 "public, max-age=0, must-revalidate".to_string()
650 } else {
651 "no-store".to_string()
652 },
653 );
654}
655
656enum PreparedSource {
658 Path(tokio::fs::File),
660 Seekable(Box<dyn SeekableReader>),
662 Stream(Box<dyn AsyncRead + Send + Unpin>),
664}
665
666impl PreparedSource {
667 fn is_seekable(&self) -> bool {
668 matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
669 }
670}
671
672async fn file_response(
677 req: &HttpRequest,
678 status: u16,
679 mut headers: HashMap<String, String>,
680 file: FileBody,
681) -> hyper::Response<RespBody> {
682 let FileBody {
683 source,
684 content_type,
685 file_name,
686 disposition,
687 enable_range_processing,
688 last_modified,
689 entity_tag,
690 } = file;
691
692 let mut modified = last_modified;
694 let mut etag = entity_tag;
695 let mut source_path: Option<PathBuf> = None;
696 let known_len;
697
698 let prepared = match source {
699 FileSource::Path(path) => {
700 let handle = match tokio::fs::File::open(&path).await {
703 Ok(handle) => handle,
704 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
705 return problem_bytes(404, headers, "Not Found")
706 }
707 Err(err) => {
708 tracing::error!(
709 path = %path.display(),
710 error = %err,
711 "cannot open file for download"
712 );
713 return problem_bytes(500, headers, "an unexpected error occurred");
714 }
715 };
716 let metadata = match handle.metadata().await {
717 Ok(metadata) => metadata,
718 Err(err) => {
719 tracing::error!(
720 path = %path.display(),
721 error = %err,
722 "cannot stat file for download"
723 );
724 return problem_bytes(500, headers, "an unexpected error occurred");
725 }
726 };
727 if !metadata.is_file() {
728 return problem_bytes(404, headers, "Not Found");
729 }
730
731 let len = metadata.len();
732 if modified.is_none() {
733 modified = metadata.modified().ok();
734 }
735 if etag.is_none() {
736 etag = Some(etag_for(len, modified));
737 }
738 known_len = Some(len);
739 source_path = Some(path);
740 PreparedSource::Path(handle)
741 }
742 FileSource::SeekableStream { reader, len } => {
743 known_len = Some(len);
744 PreparedSource::Seekable(reader)
745 }
746 FileSource::Stream(reader) => {
747 known_len = None;
748 PreparedSource::Stream(reader)
749 }
750 };
751
752 let modified = modified.filter(|value| *value <= SystemTime::now());
759
760 if !headers.contains_key("content-type") {
762 let resolved = content_type
763 .or_else(|| source_path.as_deref().and_then(guess_content_type))
764 .unwrap_or_else(|| "application/octet-stream".to_string());
765 headers.insert("content-type".to_string(), resolved);
766 }
767 if !headers.contains_key("content-disposition") {
768 if let Some(name) = &file_name {
769 headers.insert(
770 "content-disposition".to_string(),
771 content_disposition_value(disposition, name),
772 );
773 }
774 }
775 if let Some(tag) = &etag {
776 headers.insert("etag".to_string(), tag.clone());
777 }
778 if let Some(modified) = modified {
779 headers.insert(
780 "last-modified".to_string(),
781 httpdate::fmt_http_date(modified),
782 );
783 }
784
785 let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
788 headers.insert(
789 "accept-ranges".to_string(),
790 if range_capable { "bytes" } else { "none" }.to_string(),
791 );
792
793 if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
795 headers.remove("content-length");
796 return bytes_response(304, headers, Vec::new());
797 }
798
799 let mut ranges: Vec<ByteRange> = Vec::new();
801 if status == 200 && range_capable {
802 let len = known_len.expect("range-capable sources know their length");
803 if let Some(raw) = req.header("range") {
804 if if_range_matches(req, etag.as_deref(), modified) {
808 match parse_ranges(raw, len) {
809 RangeOutcome::Ignored => {}
810 RangeOutcome::Unsatisfiable => {
811 headers.insert("content-range".to_string(), format!("bytes */{len}"));
812 return problem_bytes(416, headers, "Range Not Satisfiable");
813 }
814 RangeOutcome::Satisfiable(parsed) => ranges = parsed,
815 }
816 }
817 }
818 }
819
820 if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
827 ranges.clear();
828 }
829 if ranges.len() > 1 {
830 return multipart_byteranges_response(prepared, headers, ranges, known_len);
831 }
832
833 let (status, start, count) = match ranges.first() {
834 Some(range) => {
835 let len = known_len.expect("range-capable sources know their length");
836 headers.insert(
837 "content-range".to_string(),
838 format!("bytes {}-{}/{len}", range.start, range.end),
839 );
840 (206, range.start, range.len())
841 }
842 None => (status, 0, known_len.unwrap_or(0)),
843 };
844
845 match prepared {
846 PreparedSource::Path(mut handle) => {
847 headers.insert("content-length".to_string(), count.to_string());
848 if count == 0 {
849 return bytes_response(status, headers, Vec::new());
850 }
851 if start > 0 {
852 if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
853 tracing::error!(error = %err, "cannot seek file for ranged response");
854 return problem_bytes(500, headers, "an unexpected error occurred");
855 }
856 }
857 build_response(status, headers, reader_body(handle.take(count)))
858 }
859 PreparedSource::Seekable(mut reader) => {
860 headers.insert("content-length".to_string(), count.to_string());
861 if count == 0 {
862 return bytes_response(status, headers, Vec::new());
863 }
864 if start > 0 {
865 if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
866 tracing::error!(error = %err, "cannot seek stream for ranged response");
867 return problem_bytes(500, headers, "an unexpected error occurred");
868 }
869 }
870 build_response(status, headers, reader_body(reader.take(count)))
871 }
872 PreparedSource::Stream(reader) => {
873 build_response(status, headers, reader_body(reader))
876 }
877 }
878}
879
880fn representation_content_type(headers: &HashMap<String, String>) -> String {
885 headers
886 .get("content-type")
887 .filter(|value| !value.starts_with("multipart/byteranges"))
888 .cloned()
889 .unwrap_or_else(|| "application/octet-stream".to_string())
890}
891
892fn multipart_byteranges_response(
899 prepared: PreparedSource,
900 mut headers: HashMap<String, String>,
901 ranges: Vec<ByteRange>,
902 known_len: Option<u64>,
903) -> hyper::Response<RespBody> {
904 let PreparedSource::Path(handle) = prepared else {
905 unreachable!("multi-range is only routed here for path sources");
906 };
907
908 let total_len = known_len.unwrap_or(0);
909 let media_type = representation_content_type(&headers);
910 let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
911
912 let parts: Vec<(Bytes, ByteRange)> = ranges
915 .iter()
916 .map(|range| {
917 let header = format!(
918 "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
919 range.start, range.end
920 );
921 (Bytes::from(header), *range)
922 })
923 .collect();
924 let closing = Bytes::from(format!("--{boundary}--\r\n"));
925
926 let content_length: u64 = parts
927 .iter()
928 .map(|(header, range)| header.len() as u64 + range.len() + 2)
929 .sum::<u64>()
930 + closing.len() as u64;
931
932 headers.insert(
933 "content-type".to_string(),
934 format!("multipart/byteranges; boundary={boundary}"),
935 );
936 headers.insert("content-length".to_string(), content_length.to_string());
937 headers.remove("content-range");
939
940 let stream = multipart_stream(handle, parts, closing);
941 build_response(
942 206,
943 headers,
944 BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
945 )
946}
947
948const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
950
951struct MultipartState {
953 handle: tokio::fs::File,
954 parts: std::vec::IntoIter<(Bytes, ByteRange)>,
955 closing: Bytes,
956 pending_header: Bytes,
958 phase: MultipartPhase,
959 buffer: bytes::BytesMut,
960}
961
962enum MultipartPhase {
963 NextPart,
965 Payload(u64),
967 Terminator,
969 Closing,
971 Done,
972}
973
974fn multipart_stream(
980 handle: tokio::fs::File,
981 parts: Vec<(Bytes, ByteRange)>,
982 closing: Bytes,
983) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
984 let state = MultipartState {
985 handle,
986 parts: parts.into_iter(),
987 closing,
988 pending_header: Bytes::new(),
989 phase: MultipartPhase::NextPart,
990 buffer: bytes::BytesMut::new(),
991 };
992
993 futures_util::stream::unfold(state, |mut state| async move {
994 loop {
995 match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
996 MultipartPhase::NextPart => match state.parts.next() {
997 Some((header, range)) => {
998 if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
999 return Some((Err(err), state));
1000 }
1001 state.pending_header = header;
1002 state.phase = MultipartPhase::Payload(range.len());
1003 }
1004 None => state.phase = MultipartPhase::Closing,
1005 },
1006 MultipartPhase::Payload(remaining) => {
1007 if remaining == 0 {
1008 state.phase = MultipartPhase::Terminator;
1009 continue;
1010 }
1011
1012 let limit = MULTIPART_CHUNK.min(remaining as usize);
1014 state.buffer.clear();
1015 state.buffer.reserve(limit);
1016 let read = match state.handle.read_buf(&mut state.buffer).await {
1017 Ok(read) => read,
1018 Err(err) => return Some((Err(err), state)),
1019 };
1020 if read == 0 {
1021 return Some((
1022 Err(std::io::Error::new(
1023 std::io::ErrorKind::UnexpectedEof,
1024 "file changed size while serving byte ranges",
1025 )),
1026 state,
1027 ));
1028 }
1029 let keep = read.min(limit);
1030 state.buffer.truncate(keep);
1031 let chunk = state.buffer.split().freeze();
1032
1033 let header = std::mem::take(&mut state.pending_header);
1036 let frame = if header.is_empty() {
1037 chunk
1038 } else {
1039 let mut combined =
1040 bytes::BytesMut::with_capacity(header.len() + chunk.len());
1041 combined.extend_from_slice(&header);
1042 combined.extend_from_slice(&chunk);
1043 combined.freeze()
1044 };
1045
1046 state.phase = MultipartPhase::Payload(remaining - keep as u64);
1047 return Some((Ok(Frame::data(frame)), state));
1048 }
1049 MultipartPhase::Terminator => {
1050 state.phase = MultipartPhase::NextPart;
1051 return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1052 }
1053 MultipartPhase::Closing => {
1054 state.phase = MultipartPhase::Done;
1055 let closing = std::mem::take(&mut state.closing);
1056 return Some((Ok(Frame::data(closing)), state));
1057 }
1058 MultipartPhase::Done => return None,
1059 }
1060 }
1061 })
1062}
1063
1064const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1068
1069fn reader_body<R>(reader: R) -> RespBody
1071where
1072 R: tokio::io::AsyncRead + Send + 'static,
1073{
1074 let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1075 let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1076 BodyExt::boxed_unsync(body)
1077}
1078
1079fn bytes_response(
1080 status: u16,
1081 headers: HashMap<String, String>,
1082 bytes: Vec<u8>,
1083) -> hyper::Response<RespBody> {
1084 build_response(status, headers, full_body(bytes))
1085}
1086
1087fn full_body(bytes: Vec<u8>) -> RespBody {
1088 Full::new(Bytes::from(bytes))
1089 .map_err(|never: std::convert::Infallible| match never {})
1090 .boxed_unsync()
1091}
1092
1093fn problem_bytes(
1095 status: u16,
1096 mut headers: HashMap<String, String>,
1097 detail: &str,
1098) -> hyper::Response<RespBody> {
1099 headers.insert(
1100 "content-type".to_string(),
1101 "application/problem+json".to_string(),
1102 );
1103 bytes_response(
1104 status,
1105 headers,
1106 problem_to_bytes(&build_problem(status, detail)),
1107 )
1108}
1109
1110fn build_response(
1111 status: u16,
1112 headers: HashMap<String, String>,
1113 body: RespBody,
1114) -> hyper::Response<RespBody> {
1115 let mut builder = hyper::Response::builder().status(status);
1116 for (key, value) in &headers {
1117 builder = builder.header(key.as_str(), value.as_str());
1118 }
1119 builder.body(body).unwrap_or_else(|_| {
1120 hyper::Response::builder()
1121 .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1122 .body(full_body(b"Internal Server Error".to_vec()))
1123 .expect("static error response is valid")
1124 })
1125}
1126
1127#[derive(Debug, PartialEq, Eq)]
1129enum RangeOutcome {
1130 Ignored,
1132 Unsatisfiable,
1134 Satisfiable(Vec<ByteRange>),
1136}
1137
1138const MAX_RANGE_PARTS: usize = 8;
1145
1146fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1153 let value = value.trim();
1154 let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1155 return RangeOutcome::Ignored;
1156 };
1157
1158 let mut ranges: Vec<ByteRange> = Vec::new();
1159 let mut saw_valid_spec = false;
1160
1161 for part in spec.split(',') {
1162 let part = part.trim();
1163 if part.is_empty() {
1164 continue;
1165 }
1166 match parse_single_range(part, len) {
1167 SingleRange::Malformed => return RangeOutcome::Ignored,
1169 SingleRange::Unsatisfiable => saw_valid_spec = true,
1170 SingleRange::Satisfiable(range) => {
1171 saw_valid_spec = true;
1172 ranges.push(range);
1173 }
1174 }
1175 }
1176
1177 if ranges.is_empty() {
1178 return if saw_valid_spec {
1179 RangeOutcome::Unsatisfiable
1180 } else {
1181 RangeOutcome::Ignored
1182 };
1183 }
1184
1185 ranges.sort_by_key(|range| range.start);
1186 let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1187 for range in ranges {
1188 match merged.last_mut() {
1189 Some(last) if range.start <= last.end.saturating_add(1) => {
1191 last.end = last.end.max(range.end);
1192 }
1193 _ => merged.push(range),
1194 }
1195 }
1196
1197 if merged.len() > MAX_RANGE_PARTS {
1198 return RangeOutcome::Ignored;
1199 }
1200
1201 RangeOutcome::Satisfiable(merged)
1202}
1203
1204enum SingleRange {
1205 Malformed,
1207 Unsatisfiable,
1209 Satisfiable(ByteRange),
1210}
1211
1212fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1214 let Some((start_text, end_text)) = spec.split_once('-') else {
1215 return SingleRange::Malformed;
1216 };
1217 let (start_text, end_text) = (start_text.trim(), end_text.trim());
1218
1219 if start_text.is_empty() {
1220 let Ok(suffix) = end_text.parse::<u64>() else {
1222 return SingleRange::Malformed;
1223 };
1224 if suffix == 0 || len == 0 {
1225 return SingleRange::Unsatisfiable;
1226 }
1227 return SingleRange::Satisfiable(ByteRange {
1228 start: len.saturating_sub(suffix),
1229 end: len - 1,
1230 });
1231 }
1232
1233 let Ok(start) = start_text.parse::<u64>() else {
1234 return SingleRange::Malformed;
1235 };
1236 if start >= len {
1237 return SingleRange::Unsatisfiable;
1238 }
1239
1240 let end = if end_text.is_empty() {
1241 len - 1
1242 } else {
1243 match end_text.parse::<u64>() {
1244 Ok(end) => end.min(len - 1),
1245 Err(_) => return SingleRange::Malformed,
1246 }
1247 };
1248
1249 if end < start {
1250 return SingleRange::Unsatisfiable;
1251 }
1252
1253 SingleRange::Satisfiable(ByteRange { start, end })
1254}
1255
1256fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1257 if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1258 Some(&value[prefix.len()..])
1259 } else {
1260 None
1261 }
1262}
1263
1264fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1276 match modified {
1277 Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1278 None => format!("\"{len:x}\""),
1279 }
1280}
1281
1282fn epoch_secs(time: SystemTime) -> u64 {
1283 time.duration_since(UNIX_EPOCH)
1284 .map(|d| d.as_secs())
1285 .unwrap_or(0)
1286}
1287
1288fn epoch_nanos(time: SystemTime) -> u128 {
1289 time.duration_since(UNIX_EPOCH)
1290 .map(|d| d.as_nanos())
1291 .unwrap_or(0)
1292}
1293
1294fn is_weak_etag(tag: &str) -> bool {
1296 tag.trim_start().starts_with("W/")
1297}
1298
1299fn etag_value(tag: &str) -> &str {
1301 tag.trim().trim_start_matches("W/")
1302}
1303
1304fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1310 if let Some(if_none_match) = req.header("if-none-match") {
1311 let Some(etag) = etag else {
1312 return false;
1313 };
1314 let expected = etag_value(etag);
1315 return if_none_match.split(',').any(|candidate| {
1316 let candidate = candidate.trim();
1317 candidate == "*" || etag_value(candidate) == expected
1318 });
1319 }
1320
1321 if let Some(if_modified_since) = req.header("if-modified-since") {
1322 if let Some(modified) = modified {
1323 if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1324 return epoch_secs(modified) <= epoch_secs(since);
1325 }
1326 }
1327 }
1328
1329 false
1330}
1331
1332fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1343 let Some(value) = req.header("if-range") else {
1344 return true;
1345 };
1346 let value = value.trim();
1347
1348 if value.starts_with('"') || value.starts_with("W/") {
1349 let Some(etag) = etag else {
1350 return false;
1351 };
1352 if is_weak_etag(value) || is_weak_etag(etag) {
1353 return false;
1354 }
1355 return value == etag.trim();
1356 }
1357
1358 match (httpdate::parse_http_date(value), modified) {
1359 (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1360 _ => false,
1361 }
1362}
1363
1364fn guess_content_type(path: &Path) -> Option<String> {
1365 mime_guess::from_path(path)
1366 .first()
1367 .map(|mime| mime.essence_str().to_string())
1368}
1369
1370fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1380 let mut flat = HashMap::new();
1381 let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1382 for pair in query.split('&') {
1383 if pair.is_empty() {
1384 continue;
1385 }
1386 let mut parts = pair.splitn(2, '=');
1387 if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1388 let key = percent_decode(key);
1389 let value = percent_decode(value);
1390 flat.insert(key.clone(), value.clone());
1391 multi.entry(key).or_default().push(value);
1392 }
1393 }
1394 (flat, multi)
1395}
1396
1397fn percent_decode(s: &str) -> String {
1403 if !s.contains('%') && !s.contains('+') {
1404 return s.to_string();
1405 }
1406
1407 let mut bytes = Vec::with_capacity(s.len());
1408 let mut chars = s.bytes();
1409 while let Some(byte) = chars.next() {
1410 match byte {
1411 b'%' => {
1412 let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1413 let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1414 match (hi, lo) {
1415 (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1416 _ => bytes.push(b'%'),
1417 }
1418 }
1419 b'+' => bytes.push(b' '),
1420 other => bytes.push(other),
1421 }
1422 }
1423 String::from_utf8_lossy(&bytes).into_owned()
1424}
1425
1426#[cfg(test)]
1427mod tests {
1428 use super::*;
1429
1430 #[test]
1431 fn parses_the_boundary_parameter() {
1432 assert_eq!(
1433 multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1434 Some("----abc123")
1435 );
1436 assert_eq!(
1437 multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1438 .as_deref(),
1439 Some("quoted")
1440 );
1441 assert_eq!(multipart_boundary("multipart/form-data"), None);
1442 }
1443
1444 fn ranges(start: u64, end: u64) -> RangeOutcome {
1445 RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1446 }
1447
1448 #[test]
1449 fn parses_byte_ranges() {
1450 assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1451 assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1452 assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1453 assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1455 assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1457 assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1458 }
1459
1460 #[test]
1461 fn rejects_unsatisfiable_and_malformed_ranges() {
1462 assert_eq!(
1463 parse_ranges("bytes=1000-", 1000),
1464 RangeOutcome::Unsatisfiable
1465 );
1466 assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1467 assert_eq!(
1468 parse_ranges("bytes=500-100", 1000),
1469 RangeOutcome::Unsatisfiable
1470 );
1471 assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1472 assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1473 assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1475 assert_eq!(
1476 parse_ranges("bytes=0-9,abc-def", 1000),
1477 RangeOutcome::Ignored
1478 );
1479 }
1480
1481 #[test]
1482 fn suffix_range_on_empty_file_is_unsatisfiable() {
1483 assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1484 }
1485
1486 #[test]
1487 fn multi_range_is_coalesced_into_disjoint_parts() {
1488 assert_eq!(
1490 parse_ranges("bytes=20-30,0-10", 1000),
1491 RangeOutcome::Satisfiable(vec![
1492 ByteRange { start: 0, end: 10 },
1493 ByteRange { start: 20, end: 30 },
1494 ])
1495 );
1496
1497 assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1500
1501 assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1503
1504 assert_eq!(
1506 parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1507 ranges(0, 999)
1508 );
1509
1510 assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1512 }
1513
1514 #[test]
1515 fn too_many_parts_falls_back_to_the_full_representation() {
1516 let mut spec = String::from("bytes=");
1517 for index in 0..(MAX_RANGE_PARTS + 1) {
1518 if index > 0 {
1519 spec.push(',');
1520 }
1521 spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1522 }
1523 assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1524 }
1525
1526 #[test]
1527 fn etag_changes_with_size_and_mtime() {
1528 let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1529 assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1530 assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1531 assert_ne!(
1532 etag_for(10, Some(base)),
1533 etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1534 );
1535 assert_ne!(
1539 etag_for(10, Some(base)),
1540 etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1541 );
1542 }
1543
1544 #[test]
1545 fn etag_strength_rules() {
1546 assert!(!is_weak_etag("\"abc\""));
1547 assert!(is_weak_etag("W/\"abc\""));
1548 assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1549 assert_eq!(etag_value("\"abc\""), "\"abc\"");
1550 }
1551
1552 #[test]
1553 fn parses_query_strings() {
1554 let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1555 assert_eq!(params.get("a").map(String::as_str), Some("1"));
1556 assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1557 assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1558 assert_eq!(
1559 multi.get("a").map(Vec::as_slice),
1560 Some(&["1".to_string()][..])
1561 );
1562 }
1563
1564 #[test]
1565 fn keeps_every_query_value_for_repeated_keys() {
1566 let (params, multi) = parse_query_string("id=1&id=2&id=3");
1567 assert_eq!(params.get("id").map(String::as_str), Some("3"));
1568 assert_eq!(
1569 multi.get("id").map(Vec::as_slice),
1570 Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1571 );
1572 }
1573
1574 #[test]
1575 fn percent_decodes_utf8() {
1576 let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1579 assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1580 assert_eq!(params.get("x").map(String::as_str), Some("+"));
1581 }
1582}