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::path::{Path, PathBuf};
31use std::sync::Arc;
32use std::time::{SystemTime, UNIX_EPOCH};
33use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, SeekFrom};
34use tokio::sync::Mutex;
35use webx_core::auth::IClaims;
36use webx_core::config::AppOptions;
37use webx_core::error::{Error, Result};
38use webx_core::form::{FormFileBuilder, MultipartForm};
39use webx_core::http::{
40 content_disposition_value, ByteRange, FileBody, FileSource, IClaimsExt, IHttpContext,
41 IHttpRequest, IHttpResponse, ResponseBody, SeekableReader,
42};
43
44use crate::problem_response::{build_problem, problem_to_bytes};
45
46pub type RespBody = UnsyncBoxBody<Bytes, std::io::Error>;
52
53#[derive(Debug, Clone)]
57pub struct BodyLimits {
58 pub body: usize,
60 pub multipart_body: u64,
62 pub file: u64,
64 pub field: usize,
66 pub memory_threshold: usize,
68 pub spool_dir: PathBuf,
70}
71
72impl Default for BodyLimits {
73 fn default() -> Self {
74 Self {
75 body: 10 * 1024 * 1024,
76 multipart_body: 256 * 1024 * 1024,
77 file: 128 * 1024 * 1024,
78 field: 1024 * 1024,
79 memory_threshold: 1024 * 1024,
80 spool_dir: webx_core::form::spool_root(),
81 }
82 }
83}
84
85impl BodyLimits {
86 pub fn from_options(options: &AppOptions) -> Self {
91 let spool_dir = match options.form.temp_dir.as_deref() {
92 Some(dir) if !dir.trim().is_empty() => {
93 let path = PathBuf::from(dir);
94 if path.is_absolute() {
95 path
96 } else {
97 webx_core::paths::app_base().join(path)
98 }
99 }
100 _ => webx_core::form::spool_root(),
101 };
102
103 Self {
104 body: options.app.max_body_size,
105 multipart_body: options.form.max_request_size as u64,
106 file: options.form.max_file_size,
107 field: options.form.max_field_size,
108 memory_threshold: options.form.memory_threshold,
109 spool_dir,
110 }
111 }
112
113 pub fn apply_spool_root(&self) {
116 webx_core::form::set_spool_root(self.spool_dir.clone());
117 }
118}
119
120pub struct HttpContext {
122 req: HttpRequest,
123 resp: HttpResponse,
124 claims: Option<Box<dyn IClaims>>,
126}
127
128impl HttpContext {
129 pub fn new(req: Request<Incoming>, limits: BodyLimits) -> Self {
134 let (parts, body) = req.into_parts();
135
136 let mut headers = HashMap::new();
137 for (name, value) in parts.headers.iter() {
138 if let Ok(value) = value.to_str() {
139 headers.insert(name.to_string(), value.to_string());
140 }
141 }
142
143 let (query_params, query_multi) = parts
144 .uri
145 .query()
146 .map(parse_query_string)
147 .unwrap_or_default();
148
149 let mut resp = HttpResponse::new(200);
150
151 let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
155 limits.multipart_body
156 } else {
157 limits.body as u64
158 };
159 if let Some(declared) = headers
160 .get("content-length")
161 .and_then(|value| value.parse::<u64>().ok())
162 {
163 if declared > declared_limit {
164 let problem = build_problem(
165 413,
166 format!("request body exceeds the {declared_limit}-byte limit"),
167 );
168 resp.set_status(413);
169 resp.set_header("content-type", "application/problem+json");
170 resp.body = Some(ResponseBody::Bytes(problem_to_bytes(&problem)));
171 }
172 }
173
174 let req = HttpRequest {
175 method: parts.method.to_string(),
176 path: parts.uri.path().to_string(),
177 headers,
178 query_params,
179 query_multi,
180 route_params: HashMap::new(),
181 route_pattern: None,
182 body: Mutex::new(RequestBody::Pending(Some(body))),
183 limits,
184 };
185
186 Self {
187 req,
188 resp,
189 claims: None,
190 }
191 }
192
193 pub async fn into_response(self) -> hyper::Response<RespBody> {
195 let HttpContext { req, resp, .. } = self;
196 finalize_response(&req, resp).await
197 }
198}
199
200impl IClaimsExt for HttpContext {
201 fn set_claims(&mut self, claims: Box<dyn IClaims>) {
202 self.claims = Some(claims);
203 }
204
205 fn claims(&self) -> Option<&dyn IClaims> {
206 self.claims.as_deref()
207 }
208}
209
210impl IHttpContext for HttpContext {
211 fn request(&self) -> &dyn IHttpRequest {
212 &self.req
213 }
214
215 fn request_mut(&mut self) -> &mut dyn IHttpRequest {
216 &mut self.req
217 }
218
219 fn response(&self) -> &dyn IHttpResponse {
220 &self.resp
221 }
222
223 fn response_mut(&mut self) -> &mut dyn IHttpResponse {
224 &mut self.resp
225 }
226}
227
228enum RequestBody {
234 Pending(Option<Incoming>),
236 Buffered(Vec<u8>),
238 Form(Arc<MultipartForm>),
240}
241
242pub struct HttpRequest {
244 method: String,
245 path: String,
246 headers: HashMap<String, String>,
247 query_params: HashMap<String, String>,
248 query_multi: HashMap<String, Vec<String>>,
249 route_params: HashMap<String, String>,
250 route_pattern: Option<String>,
251 body: Mutex<RequestBody>,
252 limits: BodyLimits,
253}
254
255#[async_trait::async_trait]
256impl IHttpRequest for HttpRequest {
257 fn method(&self) -> &str {
258 &self.method
259 }
260
261 fn path(&self) -> &str {
262 &self.path
263 }
264
265 fn header(&self, name: &str) -> Option<&str> {
267 if let Some(value) = self.headers.get(name) {
268 return Some(value.as_str());
269 }
270 let lower = name.to_ascii_lowercase();
271 self.headers.get(&lower).map(|s| s.as_str())
272 }
273
274 fn query(&self) -> &HashMap<String, String> {
275 &self.query_params
276 }
277
278 fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
279 Some(&self.query_multi)
280 }
281
282 fn route_params(&self) -> &HashMap<String, String> {
283 &self.route_params
284 }
285
286 fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
287 &mut self.route_params
288 }
289
290 fn route_pattern(&self) -> Option<&str> {
291 self.route_pattern.as_deref()
292 }
293
294 fn route_pattern_mut(&mut self) -> &mut Option<String> {
295 &mut self.route_pattern
296 }
297
298 async fn body_bytes(&mut self) -> Result<Vec<u8>> {
299 let mut guard = self.body.lock().await;
300 let incoming = match &mut *guard {
301 RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
302 RequestBody::Form(_) => {
303 return Err(Error::Http(
304 "request body was already read as multipart/form-data".to_string(),
305 ))
306 }
307 RequestBody::Pending(slot) => slot.take(),
308 };
309
310 let Some(incoming) = incoming else {
311 return Err(Error::Http(
312 "request body has already been consumed".to_string(),
313 ));
314 };
315
316 let bytes = read_body(incoming, self.limits.body).await?;
317 *guard = RequestBody::Buffered(bytes.clone());
318 Ok(bytes)
319 }
320
321 async fn body_text(&mut self) -> Result<String> {
322 let bytes = self.body_bytes().await?;
323 String::from_utf8(bytes).map_err(|e| Error::Http(e.to_string()))
324 }
325
326 async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
327 let mut guard = self.body.lock().await;
328
329 match &*guard {
330 RequestBody::Form(form) => return Ok(Arc::clone(form)),
331 RequestBody::Buffered(_) => {
332 return Err(Error::UnsupportedMediaType(
333 "request body was already read as a non-multipart body".to_string(),
334 ))
335 }
336 RequestBody::Pending(_) => {}
337 }
338
339 if !self.is_multipart() {
340 return Err(Error::UnsupportedMediaType(
341 "expected Content-Type: multipart/form-data".to_string(),
342 ));
343 }
344
345 let raw_content_type = self.header("content-type").unwrap_or("");
346 let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
347 Error::Validation(
348 "multipart/form-data request is missing its boundary parameter".to_string(),
349 )
350 })?;
351
352 let incoming = match &mut *guard {
353 RequestBody::Pending(slot) => slot.take(),
354 _ => None,
355 };
356
357 let Some(body) = incoming else {
358 return Err(Error::Http(
359 "request body has already been consumed".to_string(),
360 ));
361 };
362
363 let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
364 *guard = RequestBody::Form(Arc::clone(&form));
365 Ok(form)
366 }
367}
368
369async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
371 let mut out = Vec::new();
372 let mut total = 0usize;
373 let mut stream = incoming.into_data_stream();
374
375 while let Some(chunk) = stream.next().await {
376 let chunk =
377 chunk.map_err(|err| Error::Http(format!("failed to read request body: {err}")))?;
378 total = total.saturating_add(chunk.len());
379 if total > limit {
380 return Err(Error::PayloadTooLarge(format!(
381 "request body exceeds the {limit}-byte limit"
382 )));
383 }
384 out.extend_from_slice(&chunk);
385 }
386
387 Ok(out)
388}
389
390fn is_multipart_content_type(content_type: Option<&String>) -> bool {
392 content_type
393 .map(|value| {
394 value
395 .split(';')
396 .next()
397 .unwrap_or("")
398 .trim()
399 .eq_ignore_ascii_case("multipart/form-data")
400 })
401 .unwrap_or(false)
402}
403
404fn multipart_boundary(content_type: &str) -> Option<String> {
406 for part in content_type.split(';').skip(1) {
407 let part = part.trim();
408 let Some((key, value)) = part.split_once('=') else {
409 continue;
410 };
411 if key.trim().eq_ignore_ascii_case("boundary") {
412 let value = value.trim().trim_matches('"');
413 if !value.is_empty() {
414 return Some(value.to_string());
415 }
416 }
417 }
418 None
419}
420
421async fn parse_multipart(
428 incoming: Incoming,
429 boundary: &str,
430 limits: &BodyLimits,
431) -> Result<MultipartForm> {
432 let stream = incoming.into_data_stream();
433 let mut multipart = multer::Multipart::new(stream, boundary);
434 let mut form = MultipartForm::new();
435 let mut total: u64 = 0;
436
437 while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
438 let field_name = field.name().unwrap_or_default().to_string();
439 if field_name.is_empty() {
440 continue;
441 }
442 let file_name = field.file_name().map(str::to_string);
443 let content_type = field.content_type().map(|m| m.to_string());
444
445 match file_name {
446 Some(file_name) => {
447 let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
448 .memory_threshold(limits.memory_threshold)
449 .max_size(limits.file)
450 .spool_dir(limits.spool_dir.clone());
451
452 let mut field = field;
453 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
454 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
455 builder.write(&chunk).await?;
456 }
457 form.push_file(builder.finish().await?);
458 }
459 None => {
460 let mut value: Vec<u8> = Vec::new();
461 let mut field = field;
462 while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
463 total = charge(&mut total, chunk.len(), limits.multipart_body)?;
464 if value.len() + chunk.len() > limits.field {
465 return Err(Error::PayloadTooLarge(format!(
466 "form field `{field_name}` exceeds the {}-byte limit",
467 limits.field
468 )));
469 }
470 value.extend_from_slice(&chunk);
471 }
472 let text = String::from_utf8(value).map_err(|_| {
473 Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
474 })?;
475 form.push_field(field_name, text);
476 }
477 }
478 }
479
480 Ok(form)
481}
482
483fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
485 *total = total.saturating_add(len as u64);
486 if *total > limit {
487 return Err(Error::PayloadTooLarge(format!(
488 "multipart request exceeds the {limit}-byte limit"
489 )));
490 }
491 Ok(*total)
492}
493
494fn multer_error(err: multer::Error) -> Error {
495 match err {
496 multer::Error::FieldSizeExceeded { limit, .. } => {
497 Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
498 }
499 multer::Error::StreamSizeExceeded { limit } => {
500 Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
501 }
502 multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
503 Error::Validation("multipart body ended unexpectedly".to_string())
504 }
505 other => Error::Validation(format!("invalid multipart body: {other}")),
506 }
507}
508
509pub struct HttpResponse {
515 status: u16,
516 headers: HashMap<String, String>,
517 body: Option<ResponseBody>,
518}
519
520impl HttpResponse {
521 pub fn new(status: u16) -> Self {
522 Self {
523 status,
524 headers: HashMap::new(),
525 body: None,
526 }
527 }
528}
529
530#[async_trait::async_trait]
531impl IHttpResponse for HttpResponse {
532 fn status(&self) -> u16 {
533 self.status
534 }
535
536 fn has_body(&self) -> bool {
537 self.body.is_some()
538 }
539
540 fn set_status(&mut self, code: u16) {
541 self.status = code;
542 }
543
544 fn set_header(&mut self, key: &str, value: &str) {
545 self.headers.insert(key.to_string(), value.to_string());
546 }
547
548 fn remove_header(&mut self, key: &str) {
549 self.headers.remove(key);
550 }
551
552 async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
553 self.body = Some(ResponseBody::Bytes(data));
554 Ok(())
555 }
556
557 async fn write_text(&mut self, text: &str) -> Result<()> {
558 self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
559 Ok(())
560 }
561
562 async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
563 self.body = Some(body);
564 Ok(())
565 }
566
567 fn body_bytes(&self) -> Vec<u8> {
568 self.body
569 .as_ref()
570 .and_then(ResponseBody::as_bytes)
571 .map(<[u8]>::to_vec)
572 .unwrap_or_default()
573 }
574
575 fn header(&self, key: &str) -> Option<&str> {
576 self.headers.get(key).map(|s| s.as_str())
577 }
578}
579
580async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
585 let HttpResponse {
586 status,
587 mut headers,
588 body,
589 } = resp;
590
591 match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
592 ResponseBody::Bytes(bytes) => {
593 default_cache_control(&mut headers, false);
594 bytes_response(status, headers, bytes)
595 }
596 ResponseBody::File(file) => {
597 default_cache_control(&mut headers, true);
598 file_response(req, status, headers, file).await
599 }
600 }
601}
602
603fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
609 if headers.contains_key("cache-control") {
610 return;
611 }
612 headers.insert(
613 "cache-control".to_string(),
614 if is_file {
615 "public, max-age=0, must-revalidate".to_string()
616 } else {
617 "no-store".to_string()
618 },
619 );
620}
621
622enum PreparedSource {
624 Path(tokio::fs::File),
626 Seekable(Box<dyn SeekableReader>),
628 Stream(Box<dyn AsyncRead + Send + Unpin>),
630}
631
632impl PreparedSource {
633 fn is_seekable(&self) -> bool {
634 matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
635 }
636}
637
638async fn file_response(
643 req: &HttpRequest,
644 status: u16,
645 mut headers: HashMap<String, String>,
646 file: FileBody,
647) -> hyper::Response<RespBody> {
648 let FileBody {
649 source,
650 content_type,
651 file_name,
652 disposition,
653 enable_range_processing,
654 last_modified,
655 entity_tag,
656 } = file;
657
658 let mut modified = last_modified;
660 let mut etag = entity_tag;
661 let mut source_path: Option<PathBuf> = None;
662 let known_len;
663
664 let prepared = match source {
665 FileSource::Path(path) => {
666 let handle = match tokio::fs::File::open(&path).await {
669 Ok(handle) => handle,
670 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
671 return problem_bytes(404, headers, "Not Found")
672 }
673 Err(err) => {
674 return problem_bytes(
675 500,
676 headers,
677 &format!("cannot open `{}`: {err}", path.display()),
678 )
679 }
680 };
681 let metadata = match handle.metadata().await {
682 Ok(metadata) => metadata,
683 Err(err) => {
684 return problem_bytes(
685 500,
686 headers,
687 &format!("cannot stat `{}`: {err}", path.display()),
688 )
689 }
690 };
691 if !metadata.is_file() {
692 return problem_bytes(404, headers, "Not Found");
693 }
694
695 let len = metadata.len();
696 if modified.is_none() {
697 modified = metadata.modified().ok();
698 }
699 if etag.is_none() {
700 etag = Some(etag_for(len, modified));
701 }
702 known_len = Some(len);
703 source_path = Some(path);
704 PreparedSource::Path(handle)
705 }
706 FileSource::SeekableStream { reader, len } => {
707 known_len = Some(len);
708 PreparedSource::Seekable(reader)
709 }
710 FileSource::Stream(reader) => {
711 known_len = None;
712 PreparedSource::Stream(reader)
713 }
714 };
715
716 let modified = modified.filter(|value| *value <= SystemTime::now());
723
724 if !headers.contains_key("content-type") {
726 let resolved = content_type
727 .or_else(|| source_path.as_deref().and_then(guess_content_type))
728 .unwrap_or_else(|| "application/octet-stream".to_string());
729 headers.insert("content-type".to_string(), resolved);
730 }
731 if !headers.contains_key("content-disposition") {
732 if let Some(name) = &file_name {
733 headers.insert(
734 "content-disposition".to_string(),
735 content_disposition_value(disposition, name),
736 );
737 }
738 }
739 if let Some(tag) = &etag {
740 headers.insert("etag".to_string(), tag.clone());
741 }
742 if let Some(modified) = modified {
743 headers.insert(
744 "last-modified".to_string(),
745 httpdate::fmt_http_date(modified),
746 );
747 }
748
749 let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
752 headers.insert(
753 "accept-ranges".to_string(),
754 if range_capable { "bytes" } else { "none" }.to_string(),
755 );
756
757 if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
759 headers.remove("content-length");
760 return bytes_response(304, headers, Vec::new());
761 }
762
763 let mut ranges: Vec<ByteRange> = Vec::new();
765 if status == 200 && range_capable {
766 let len = known_len.expect("range-capable sources know their length");
767 if let Some(raw) = req.header("range") {
768 if if_range_matches(req, etag.as_deref(), modified) {
772 match parse_ranges(raw, len) {
773 RangeOutcome::Ignored => {}
774 RangeOutcome::Unsatisfiable => {
775 headers.insert("content-range".to_string(), format!("bytes */{len}"));
776 return problem_bytes(416, headers, "Range Not Satisfiable");
777 }
778 RangeOutcome::Satisfiable(parsed) => ranges = parsed,
779 }
780 }
781 }
782 }
783
784 if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
791 ranges.clear();
792 }
793 if ranges.len() > 1 {
794 return multipart_byteranges_response(prepared, headers, ranges, known_len);
795 }
796
797 let (status, start, count) = match ranges.first() {
798 Some(range) => {
799 let len = known_len.expect("range-capable sources know their length");
800 headers.insert(
801 "content-range".to_string(),
802 format!("bytes {}-{}/{len}", range.start, range.end),
803 );
804 (206, range.start, range.len())
805 }
806 None => (status, 0, known_len.unwrap_or(0)),
807 };
808
809 match prepared {
810 PreparedSource::Path(mut handle) => {
811 headers.insert("content-length".to_string(), count.to_string());
812 if count == 0 {
813 return bytes_response(status, headers, Vec::new());
814 }
815 if start > 0 {
816 if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
817 return problem_bytes(500, headers, &format!("cannot seek: {err}"));
818 }
819 }
820 build_response(status, headers, reader_body(handle.take(count)))
821 }
822 PreparedSource::Seekable(mut reader) => {
823 headers.insert("content-length".to_string(), count.to_string());
824 if count == 0 {
825 return bytes_response(status, headers, Vec::new());
826 }
827 if start > 0 {
828 if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
829 return problem_bytes(500, headers, &format!("cannot seek: {err}"));
830 }
831 }
832 build_response(status, headers, reader_body(reader.take(count)))
833 }
834 PreparedSource::Stream(reader) => {
835 build_response(status, headers, reader_body(reader))
838 }
839 }
840}
841
842fn representation_content_type(headers: &HashMap<String, String>) -> String {
847 headers
848 .get("content-type")
849 .filter(|value| !value.starts_with("multipart/byteranges"))
850 .cloned()
851 .unwrap_or_else(|| "application/octet-stream".to_string())
852}
853
854fn multipart_byteranges_response(
861 prepared: PreparedSource,
862 mut headers: HashMap<String, String>,
863 ranges: Vec<ByteRange>,
864 known_len: Option<u64>,
865) -> hyper::Response<RespBody> {
866 let PreparedSource::Path(handle) = prepared else {
867 unreachable!("multi-range is only routed here for path sources");
868 };
869
870 let total_len = known_len.unwrap_or(0);
871 let media_type = representation_content_type(&headers);
872 let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
873
874 let parts: Vec<(Bytes, ByteRange)> = ranges
877 .iter()
878 .map(|range| {
879 let header = format!(
880 "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
881 range.start, range.end
882 );
883 (Bytes::from(header), *range)
884 })
885 .collect();
886 let closing = Bytes::from(format!("--{boundary}--\r\n"));
887
888 let content_length: u64 = parts
889 .iter()
890 .map(|(header, range)| header.len() as u64 + range.len() + 2)
891 .sum::<u64>()
892 + closing.len() as u64;
893
894 headers.insert(
895 "content-type".to_string(),
896 format!("multipart/byteranges; boundary={boundary}"),
897 );
898 headers.insert("content-length".to_string(), content_length.to_string());
899 headers.remove("content-range");
901
902 let stream = multipart_stream(handle, parts, closing);
903 build_response(
904 206,
905 headers,
906 BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
907 )
908}
909
910const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
912
913struct MultipartState {
915 handle: tokio::fs::File,
916 parts: std::vec::IntoIter<(Bytes, ByteRange)>,
917 closing: Bytes,
918 pending_header: Bytes,
920 phase: MultipartPhase,
921 buffer: bytes::BytesMut,
922}
923
924enum MultipartPhase {
925 NextPart,
927 Payload(u64),
929 Terminator,
931 Closing,
933 Done,
934}
935
936fn multipart_stream(
942 handle: tokio::fs::File,
943 parts: Vec<(Bytes, ByteRange)>,
944 closing: Bytes,
945) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
946 let state = MultipartState {
947 handle,
948 parts: parts.into_iter(),
949 closing,
950 pending_header: Bytes::new(),
951 phase: MultipartPhase::NextPart,
952 buffer: bytes::BytesMut::new(),
953 };
954
955 futures_util::stream::unfold(state, |mut state| async move {
956 loop {
957 match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
958 MultipartPhase::NextPart => match state.parts.next() {
959 Some((header, range)) => {
960 if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
961 return Some((Err(err), state));
962 }
963 state.pending_header = header;
964 state.phase = MultipartPhase::Payload(range.len());
965 }
966 None => state.phase = MultipartPhase::Closing,
967 },
968 MultipartPhase::Payload(remaining) => {
969 if remaining == 0 {
970 state.phase = MultipartPhase::Terminator;
971 continue;
972 }
973
974 let limit = MULTIPART_CHUNK.min(remaining as usize);
976 state.buffer.clear();
977 state.buffer.reserve(limit);
978 let read = match state.handle.read_buf(&mut state.buffer).await {
979 Ok(read) => read,
980 Err(err) => return Some((Err(err), state)),
981 };
982 if read == 0 {
983 return Some((
984 Err(std::io::Error::new(
985 std::io::ErrorKind::UnexpectedEof,
986 "file changed size while serving byte ranges",
987 )),
988 state,
989 ));
990 }
991 let keep = read.min(limit);
992 state.buffer.truncate(keep);
993 let chunk = state.buffer.split().freeze();
994
995 let header = std::mem::take(&mut state.pending_header);
998 let frame = if header.is_empty() {
999 chunk
1000 } else {
1001 let mut combined =
1002 bytes::BytesMut::with_capacity(header.len() + chunk.len());
1003 combined.extend_from_slice(&header);
1004 combined.extend_from_slice(&chunk);
1005 combined.freeze()
1006 };
1007
1008 state.phase = MultipartPhase::Payload(remaining - keep as u64);
1009 return Some((Ok(Frame::data(frame)), state));
1010 }
1011 MultipartPhase::Terminator => {
1012 state.phase = MultipartPhase::NextPart;
1013 return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1014 }
1015 MultipartPhase::Closing => {
1016 state.phase = MultipartPhase::Done;
1017 let closing = std::mem::take(&mut state.closing);
1018 return Some((Ok(Frame::data(closing)), state));
1019 }
1020 MultipartPhase::Done => return None,
1021 }
1022 }
1023 })
1024}
1025
1026const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1030
1031fn reader_body<R>(reader: R) -> RespBody
1033where
1034 R: tokio::io::AsyncRead + Send + 'static,
1035{
1036 let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1037 let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1038 BodyExt::boxed_unsync(body)
1039}
1040
1041fn bytes_response(
1042 status: u16,
1043 headers: HashMap<String, String>,
1044 bytes: Vec<u8>,
1045) -> hyper::Response<RespBody> {
1046 build_response(status, headers, full_body(bytes))
1047}
1048
1049fn full_body(bytes: Vec<u8>) -> RespBody {
1050 Full::new(Bytes::from(bytes))
1051 .map_err(|never: std::convert::Infallible| match never {})
1052 .boxed_unsync()
1053}
1054
1055fn problem_bytes(
1057 status: u16,
1058 mut headers: HashMap<String, String>,
1059 detail: &str,
1060) -> hyper::Response<RespBody> {
1061 headers.insert(
1062 "content-type".to_string(),
1063 "application/problem+json".to_string(),
1064 );
1065 bytes_response(
1066 status,
1067 headers,
1068 problem_to_bytes(&build_problem(status, detail)),
1069 )
1070}
1071
1072fn build_response(
1073 status: u16,
1074 headers: HashMap<String, String>,
1075 body: RespBody,
1076) -> hyper::Response<RespBody> {
1077 let mut builder = hyper::Response::builder().status(status);
1078 for (key, value) in &headers {
1079 builder = builder.header(key.as_str(), value.as_str());
1080 }
1081 builder.body(body).unwrap_or_else(|_| {
1082 hyper::Response::builder()
1083 .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1084 .body(full_body(b"Internal Server Error".to_vec()))
1085 .expect("static error response is valid")
1086 })
1087}
1088
1089#[derive(Debug, PartialEq, Eq)]
1091enum RangeOutcome {
1092 Ignored,
1094 Unsatisfiable,
1096 Satisfiable(Vec<ByteRange>),
1098}
1099
1100const MAX_RANGE_PARTS: usize = 8;
1107
1108fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1115 let value = value.trim();
1116 let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1117 return RangeOutcome::Ignored;
1118 };
1119
1120 let mut ranges: Vec<ByteRange> = Vec::new();
1121 let mut saw_valid_spec = false;
1122
1123 for part in spec.split(',') {
1124 let part = part.trim();
1125 if part.is_empty() {
1126 continue;
1127 }
1128 match parse_single_range(part, len) {
1129 SingleRange::Malformed => return RangeOutcome::Ignored,
1131 SingleRange::Unsatisfiable => saw_valid_spec = true,
1132 SingleRange::Satisfiable(range) => {
1133 saw_valid_spec = true;
1134 ranges.push(range);
1135 }
1136 }
1137 }
1138
1139 if ranges.is_empty() {
1140 return if saw_valid_spec {
1141 RangeOutcome::Unsatisfiable
1142 } else {
1143 RangeOutcome::Ignored
1144 };
1145 }
1146
1147 ranges.sort_by_key(|range| range.start);
1148 let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1149 for range in ranges {
1150 match merged.last_mut() {
1151 Some(last) if range.start <= last.end.saturating_add(1) => {
1153 last.end = last.end.max(range.end);
1154 }
1155 _ => merged.push(range),
1156 }
1157 }
1158
1159 if merged.len() > MAX_RANGE_PARTS {
1160 return RangeOutcome::Ignored;
1161 }
1162
1163 RangeOutcome::Satisfiable(merged)
1164}
1165
1166enum SingleRange {
1167 Malformed,
1169 Unsatisfiable,
1171 Satisfiable(ByteRange),
1172}
1173
1174fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1176 let Some((start_text, end_text)) = spec.split_once('-') else {
1177 return SingleRange::Malformed;
1178 };
1179 let (start_text, end_text) = (start_text.trim(), end_text.trim());
1180
1181 if start_text.is_empty() {
1182 let Ok(suffix) = end_text.parse::<u64>() else {
1184 return SingleRange::Malformed;
1185 };
1186 if suffix == 0 || len == 0 {
1187 return SingleRange::Unsatisfiable;
1188 }
1189 return SingleRange::Satisfiable(ByteRange {
1190 start: len.saturating_sub(suffix),
1191 end: len - 1,
1192 });
1193 }
1194
1195 let Ok(start) = start_text.parse::<u64>() else {
1196 return SingleRange::Malformed;
1197 };
1198 if start >= len {
1199 return SingleRange::Unsatisfiable;
1200 }
1201
1202 let end = if end_text.is_empty() {
1203 len - 1
1204 } else {
1205 match end_text.parse::<u64>() {
1206 Ok(end) => end.min(len - 1),
1207 Err(_) => return SingleRange::Malformed,
1208 }
1209 };
1210
1211 if end < start {
1212 return SingleRange::Unsatisfiable;
1213 }
1214
1215 SingleRange::Satisfiable(ByteRange { start, end })
1216}
1217
1218fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1219 if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1220 Some(&value[prefix.len()..])
1221 } else {
1222 None
1223 }
1224}
1225
1226fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1238 match modified {
1239 Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1240 None => format!("\"{len:x}\""),
1241 }
1242}
1243
1244fn epoch_secs(time: SystemTime) -> u64 {
1245 time.duration_since(UNIX_EPOCH)
1246 .map(|d| d.as_secs())
1247 .unwrap_or(0)
1248}
1249
1250fn epoch_nanos(time: SystemTime) -> u128 {
1251 time.duration_since(UNIX_EPOCH)
1252 .map(|d| d.as_nanos())
1253 .unwrap_or(0)
1254}
1255
1256fn is_weak_etag(tag: &str) -> bool {
1258 tag.trim_start().starts_with("W/")
1259}
1260
1261fn etag_value(tag: &str) -> &str {
1263 tag.trim().trim_start_matches("W/")
1264}
1265
1266fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1272 if let Some(if_none_match) = req.header("if-none-match") {
1273 let Some(etag) = etag else {
1274 return false;
1275 };
1276 let expected = etag_value(etag);
1277 return if_none_match.split(',').any(|candidate| {
1278 let candidate = candidate.trim();
1279 candidate == "*" || etag_value(candidate) == expected
1280 });
1281 }
1282
1283 if let Some(if_modified_since) = req.header("if-modified-since") {
1284 if let Some(modified) = modified {
1285 if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1286 return epoch_secs(modified) <= epoch_secs(since);
1287 }
1288 }
1289 }
1290
1291 false
1292}
1293
1294fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1305 let Some(value) = req.header("if-range") else {
1306 return true;
1307 };
1308 let value = value.trim();
1309
1310 if value.starts_with('"') || value.starts_with("W/") {
1311 let Some(etag) = etag else {
1312 return false;
1313 };
1314 if is_weak_etag(value) || is_weak_etag(etag) {
1315 return false;
1316 }
1317 return value == etag.trim();
1318 }
1319
1320 match (httpdate::parse_http_date(value), modified) {
1321 (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1322 _ => false,
1323 }
1324}
1325
1326fn guess_content_type(path: &Path) -> Option<String> {
1327 mime_guess::from_path(path)
1328 .first()
1329 .map(|mime| mime.essence_str().to_string())
1330}
1331
1332fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1342 let mut flat = HashMap::new();
1343 let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1344 for pair in query.split('&') {
1345 if pair.is_empty() {
1346 continue;
1347 }
1348 let mut parts = pair.splitn(2, '=');
1349 if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1350 let key = percent_decode(key);
1351 let value = percent_decode(value);
1352 flat.insert(key.clone(), value.clone());
1353 multi.entry(key).or_default().push(value);
1354 }
1355 }
1356 (flat, multi)
1357}
1358
1359fn percent_decode(s: &str) -> String {
1365 if !s.contains('%') && !s.contains('+') {
1366 return s.to_string();
1367 }
1368
1369 let mut bytes = Vec::with_capacity(s.len());
1370 let mut chars = s.bytes();
1371 while let Some(byte) = chars.next() {
1372 match byte {
1373 b'%' => {
1374 let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1375 let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1376 match (hi, lo) {
1377 (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1378 _ => bytes.push(b'%'),
1379 }
1380 }
1381 b'+' => bytes.push(b' '),
1382 other => bytes.push(other),
1383 }
1384 }
1385 String::from_utf8_lossy(&bytes).into_owned()
1386}
1387
1388#[cfg(test)]
1389mod tests {
1390 use super::*;
1391
1392 #[test]
1393 fn parses_the_boundary_parameter() {
1394 assert_eq!(
1395 multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1396 Some("----abc123")
1397 );
1398 assert_eq!(
1399 multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1400 .as_deref(),
1401 Some("quoted")
1402 );
1403 assert_eq!(multipart_boundary("multipart/form-data"), None);
1404 }
1405
1406 fn ranges(start: u64, end: u64) -> RangeOutcome {
1407 RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1408 }
1409
1410 #[test]
1411 fn parses_byte_ranges() {
1412 assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1413 assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1414 assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1415 assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1417 assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1419 assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1420 }
1421
1422 #[test]
1423 fn rejects_unsatisfiable_and_malformed_ranges() {
1424 assert_eq!(
1425 parse_ranges("bytes=1000-", 1000),
1426 RangeOutcome::Unsatisfiable
1427 );
1428 assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1429 assert_eq!(
1430 parse_ranges("bytes=500-100", 1000),
1431 RangeOutcome::Unsatisfiable
1432 );
1433 assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1434 assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1435 assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1437 assert_eq!(
1438 parse_ranges("bytes=0-9,abc-def", 1000),
1439 RangeOutcome::Ignored
1440 );
1441 }
1442
1443 #[test]
1444 fn suffix_range_on_empty_file_is_unsatisfiable() {
1445 assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1446 }
1447
1448 #[test]
1449 fn multi_range_is_coalesced_into_disjoint_parts() {
1450 assert_eq!(
1452 parse_ranges("bytes=20-30,0-10", 1000),
1453 RangeOutcome::Satisfiable(vec![
1454 ByteRange { start: 0, end: 10 },
1455 ByteRange { start: 20, end: 30 },
1456 ])
1457 );
1458
1459 assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1462
1463 assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1465
1466 assert_eq!(
1468 parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1469 ranges(0, 999)
1470 );
1471
1472 assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1474 }
1475
1476 #[test]
1477 fn too_many_parts_falls_back_to_the_full_representation() {
1478 let mut spec = String::from("bytes=");
1479 for index in 0..(MAX_RANGE_PARTS + 1) {
1480 if index > 0 {
1481 spec.push(',');
1482 }
1483 spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1484 }
1485 assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1486 }
1487
1488 #[test]
1489 fn etag_changes_with_size_and_mtime() {
1490 let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1491 assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1492 assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1493 assert_ne!(
1494 etag_for(10, Some(base)),
1495 etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1496 );
1497 assert_ne!(
1501 etag_for(10, Some(base)),
1502 etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1503 );
1504 }
1505
1506 #[test]
1507 fn etag_strength_rules() {
1508 assert!(!is_weak_etag("\"abc\""));
1509 assert!(is_weak_etag("W/\"abc\""));
1510 assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1511 assert_eq!(etag_value("\"abc\""), "\"abc\"");
1512 }
1513
1514 #[test]
1515 fn parses_query_strings() {
1516 let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1517 assert_eq!(params.get("a").map(String::as_str), Some("1"));
1518 assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1519 assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1520 assert_eq!(
1521 multi.get("a").map(Vec::as_slice),
1522 Some(&["1".to_string()][..])
1523 );
1524 }
1525
1526 #[test]
1527 fn keeps_every_query_value_for_repeated_keys() {
1528 let (params, multi) = parse_query_string("id=1&id=2&id=3");
1529 assert_eq!(params.get("id").map(String::as_str), Some("3"));
1530 assert_eq!(
1531 multi.get("id").map(Vec::as_slice),
1532 Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1533 );
1534 }
1535
1536 #[test]
1537 fn percent_decodes_utf8() {
1538 let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1541 assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1542 assert_eq!(params.get("x").map(String::as_str), Some("+"));
1543 }
1544}