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