Skip to main content

webx_host/
context.rs

1//! HttpContext — runtime implementation of IHttpContext, IHttpRequest, IHttpResponse.
2//!
3//! # Request body
4//!
5//! The body is *not* read eagerly. It stays as the live hyper stream until
6//! something asks for it:
7//!
8//! * [`IHttpRequest::body_bytes`] buffers the whole body, enforcing
9//!   `App.MaxBodySize`.
10//! * [`IHttpRequest::multipart`] streams the body through a multipart parser,
11//!   enforcing `Form.MaxRequestSize`, `Form.MaxFileSize` and
12//!   `Form.MaxFieldSize`, and spooling large file parts to disk.
13//!
14//! Either way the body can only be consumed once; asking for it twice reports a
15//! clear error rather than silently returning an empty body.
16//!
17//! # Response body
18//!
19//! Responses carry either in-memory bytes or a [`FileBody`]. File bodies are
20//! streamed by the host with `Content-Length`, `ETag`, `Last-Modified`,
21//! `Accept-Ranges` and `Content-Disposition`, and satisfy `Range`,
22//! `If-Range`, `If-None-Match` and `If-Modified-Since` requests, so downloads
23//! can be cached and resumed.
24
25use 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
47/// Response body produced by the HTTP host.
48///
49/// Deliberately the *unsync* box: a streamed body only ever needs to be `Send`,
50/// and requiring `Sync` would rule out readers that are not shared between
51/// threads (most streaming sources).
52pub type RespBody = UnsyncBoxBody<Bytes, std::io::Error>;
53
54/// Limits applied while reading a request body.
55///
56/// Built from `App` and `Form` configuration; see [`BodyLimits::from_options`].
57#[derive(Debug, Clone)]
58pub struct BodyLimits {
59    /// Limit for non-multipart bodies (`App.MaxBodySize`).
60    pub body: usize,
61    /// Limit for the whole multipart body (`Form.MaxRequestSize`).
62    pub multipart_body: u64,
63    /// Limit for one uploaded file (`Form.MaxFileSize`).
64    pub file: u64,
65    /// Limit for one text field (`Form.MaxFieldSize`).
66    pub field: usize,
67    /// Bytes buffered in memory per file before spooling (`Form.MemoryThreshold`).
68    pub memory_threshold: usize,
69    /// Directory for spooled uploads (`Form.TempDir`).
70    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    /// Derive limits from the merged application configuration.
88    ///
89    /// A relative `Form.TempDir` is resolved against the application base
90    /// directory, matching how `wwwroot` and `appsettings.json` are located.
91    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    /// Publish the spool directory so `FormFile` values can be validated
115    /// wherever they are bound.
116    pub fn apply_spool_root(&self) {
117        webx_core::form::set_spool_root(self.spool_dir.clone());
118    }
119}
120
121/// Concrete implementation of IHttpContext wrapping a hyper request and response.
122pub struct HttpContext {
123    req: HttpRequest,
124    resp: HttpResponse,
125    /// Authentication claims set by authentication middleware.
126    claims: Option<Box<dyn IClaims>>,
127    /// A request rejection recorded at construction (e.g. an oversized declared
128    /// `Content-Length`). The pipeline turns it into a normal error response
129    /// from the innermost handler so CORS/security headers/logging still run.
130    early_error: Option<Error>,
131}
132
133impl HttpContext {
134    /// Create a context around `req`.
135    ///
136    /// No I/O happens here: the body stays a live stream until the endpoint
137    /// asks for it. `remote_addr` is the peer address captured by the accept
138    /// loop; `None` when the host cannot report it.
139    pub fn new(
140        req: Request<Incoming>,
141        limits: BodyLimits,
142        remote_addr: Option<SocketAddr>,
143    ) -> Self {
144        let (parts, body) = req.into_parts();
145
146        let mut headers = HashMap::new();
147        for (name, value) in parts.headers.iter() {
148            if let Ok(value) = value.to_str() {
149                headers.insert(name.to_string(), value.to_string());
150            }
151        }
152
153        let (query_params, query_multi) = parts
154            .uri
155            .query()
156            .map(parse_query_string)
157            .unwrap_or_default();
158
159        let resp = HttpResponse::new(200);
160
161        // Refuse an oversized upload from its declared length, without reading
162        // it. Chunked bodies have no `Content-Length` and are bounded while
163        // streaming instead.
164        //
165        // The rejection is recorded, not written eagerly: writing a response
166        // here used to bypass the whole middleware pipeline (no CORS, no
167        // security headers, no access log, no metrics). The router handler
168        // turns it into a normal `PayloadTooLarge` error on the way in.
169        let declared_limit = if is_multipart_content_type(headers.get("content-type")) {
170            limits.multipart_body
171        } else {
172            limits.body as u64
173        };
174        let early_error = headers
175            .get("content-length")
176            .and_then(|value| value.parse::<u64>().ok())
177            .filter(|declared| *declared > declared_limit)
178            .map(|_| {
179                Error::PayloadTooLarge(format!(
180                    "request body exceeds the {declared_limit}-byte limit"
181                ))
182            });
183
184        let req = HttpRequest {
185            method: parts.method.to_string(),
186            path: parts.uri.path().to_string(),
187            headers,
188            query_params,
189            query_multi,
190            route_params: HashMap::new(),
191            route_pattern: None,
192            remote_addr,
193            body: Mutex::new(RequestBody::Pending(Some(body))),
194            limits,
195        };
196
197        Self {
198            req,
199            resp,
200            claims: None,
201            early_error,
202        }
203    }
204
205    /// Finish the request and produce the hyper response.
206    pub async fn into_response(self) -> hyper::Response<RespBody> {
207        let HttpContext { req, resp, .. } = self;
208        finalize_response(&req, resp).await
209    }
210}
211
212impl IClaimsExt for HttpContext {
213    fn set_claims(&mut self, claims: Box<dyn IClaims>) {
214        self.claims = Some(claims);
215    }
216
217    fn claims(&self) -> Option<&dyn IClaims> {
218        self.claims.as_deref()
219    }
220}
221
222impl IHttpContext for HttpContext {
223    fn request(&self) -> &dyn IHttpRequest {
224        &self.req
225    }
226
227    fn request_mut(&mut self) -> &mut dyn IHttpRequest {
228        &mut self.req
229    }
230
231    fn response(&self) -> &dyn IHttpResponse {
232        &self.resp
233    }
234
235    fn response_mut(&mut self) -> &mut dyn IHttpResponse {
236        &mut self.resp
237    }
238
239    fn take_early_error(&mut self) -> Option<Error> {
240        self.early_error.take()
241    }
242}
243
244// ---------------------------------------------------------------------------
245// Request
246// ---------------------------------------------------------------------------
247
248/// State of the request body.
249enum RequestBody {
250    /// Still streaming from the connection.
251    Pending(Option<Incoming>),
252    /// Fully buffered, replayable.
253    Buffered(Vec<u8>),
254    /// Parsed as a multipart form.
255    Form(Arc<MultipartForm>),
256}
257
258/// Concrete implementation of IHttpRequest.
259pub struct HttpRequest {
260    method: String,
261    path: String,
262    headers: HashMap<String, String>,
263    query_params: HashMap<String, String>,
264    query_multi: HashMap<String, Vec<String>>,
265    route_params: HashMap<String, String>,
266    route_pattern: Option<String>,
267    remote_addr: Option<SocketAddr>,
268    body: Mutex<RequestBody>,
269    limits: BodyLimits,
270}
271
272#[async_trait::async_trait]
273impl IHttpRequest for HttpRequest {
274    fn method(&self) -> &str {
275        &self.method
276    }
277
278    fn path(&self) -> &str {
279        &self.path
280    }
281
282    /// Header values are looked up case-insensitively, as HTTP requires.
283    fn header(&self, name: &str) -> Option<&str> {
284        if let Some(value) = self.headers.get(name) {
285            return Some(value.as_str());
286        }
287        let lower = name.to_ascii_lowercase();
288        self.headers.get(&lower).map(|s| s.as_str())
289    }
290
291    fn query(&self) -> &HashMap<String, String> {
292        &self.query_params
293    }
294
295    fn query_all(&self) -> Option<&HashMap<String, Vec<String>>> {
296        Some(&self.query_multi)
297    }
298
299    fn remote_addr(&self) -> Option<SocketAddr> {
300        self.remote_addr
301    }
302
303    fn route_params(&self) -> &HashMap<String, String> {
304        &self.route_params
305    }
306
307    fn route_params_mut(&mut self) -> &mut HashMap<String, String> {
308        &mut self.route_params
309    }
310
311    fn route_pattern(&self) -> Option<&str> {
312        self.route_pattern.as_deref()
313    }
314
315    fn route_pattern_mut(&mut self) -> &mut Option<String> {
316        &mut self.route_pattern
317    }
318
319    async fn body_bytes(&mut self) -> Result<Vec<u8>> {
320        let mut guard = self.body.lock().await;
321        let incoming = match &mut *guard {
322            RequestBody::Buffered(bytes) => return Ok(bytes.clone()),
323            RequestBody::Form(_) => {
324                return Err(Error::State(
325                    "request body was already read as multipart/form-data".to_string(),
326                ))
327            }
328            RequestBody::Pending(slot) => slot.take(),
329        };
330
331        let Some(incoming) = incoming else {
332            return Err(Error::State(
333                "request body has already been consumed".to_string(),
334            ));
335        };
336
337        let bytes = read_body(incoming, self.limits.body).await?;
338        *guard = RequestBody::Buffered(bytes.clone());
339        Ok(bytes)
340    }
341
342    async fn body_text(&mut self) -> Result<String> {
343        let bytes = self.body_bytes().await?;
344        String::from_utf8(bytes)
345            .map_err(|_| Error::Validation("request body is not valid UTF-8".to_string()))
346    }
347
348    async fn multipart(&mut self) -> Result<Arc<MultipartForm>> {
349        let mut guard = self.body.lock().await;
350
351        match &*guard {
352            RequestBody::Form(form) => return Ok(Arc::clone(form)),
353            RequestBody::Buffered(_) => {
354                return Err(Error::UnsupportedMediaType(
355                    "request body was already read as a non-multipart body".to_string(),
356                ))
357            }
358            RequestBody::Pending(_) => {}
359        }
360
361        if !self.is_multipart() {
362            return Err(Error::UnsupportedMediaType(
363                "expected Content-Type: multipart/form-data".to_string(),
364            ));
365        }
366
367        let raw_content_type = self.header("content-type").unwrap_or("");
368        let boundary = multipart_boundary(raw_content_type).ok_or_else(|| {
369            Error::Validation(
370                "multipart/form-data request is missing its boundary parameter".to_string(),
371            )
372        })?;
373
374        let incoming = match &mut *guard {
375            RequestBody::Pending(slot) => slot.take(),
376            _ => None,
377        };
378
379        let Some(body) = incoming else {
380            return Err(Error::State(
381                "request body has already been consumed".to_string(),
382            ));
383        };
384
385        let form = Arc::new(parse_multipart(body, &boundary, &self.limits).await?);
386        *guard = RequestBody::Form(Arc::clone(&form));
387        Ok(form)
388    }
389}
390
391/// Drain a body stream into memory, enforcing `limit`.
392async fn read_body(incoming: Incoming, limit: usize) -> Result<Vec<u8>> {
393    let mut out = Vec::new();
394    let mut total = 0usize;
395    let mut stream = incoming.into_data_stream();
396
397    while let Some(chunk) = stream.next().await {
398        let chunk = chunk.map_err(|err| {
399            // Transport failure while reading the body: the client went away or
400            // the framing is broken. Log the details, never serve them.
401            tracing::debug!(error = %err, "failed to read request body");
402            Error::State("failed to read request body".to_string())
403        })?;
404        total = total.saturating_add(chunk.len());
405        if total > limit {
406            return Err(Error::PayloadTooLarge(format!(
407                "request body exceeds the {limit}-byte limit"
408            )));
409        }
410        out.extend_from_slice(&chunk);
411    }
412
413    Ok(out)
414}
415
416/// Whether a raw `Content-Type` header value declares a multipart form.
417fn is_multipart_content_type(content_type: Option<&String>) -> bool {
418    content_type
419        .map(|value| {
420            value
421                .split(';')
422                .next()
423                .unwrap_or("")
424                .trim()
425                .eq_ignore_ascii_case("multipart/form-data")
426        })
427        .unwrap_or(false)
428}
429
430/// Extract the `boundary` parameter of a `multipart/form-data` content type.
431fn multipart_boundary(content_type: &str) -> Option<String> {
432    for part in content_type.split(';').skip(1) {
433        let part = part.trim();
434        let Some((key, value)) = part.split_once('=') else {
435            continue;
436        };
437        if key.trim().eq_ignore_ascii_case("boundary") {
438            let value = value.trim().trim_matches('"');
439            if !value.is_empty() {
440                return Some(value.to_string());
441            }
442        }
443    }
444    None
445}
446
447/// Stream a multipart body into a [`MultipartForm`].
448///
449/// Text fields are buffered (bounded by `limits.field`); file parts stream into
450/// a [`FormFileBuilder`], which keeps small files in memory and spools large
451/// ones to disk. Nothing is copied twice, so peak memory stays bounded no
452/// matter how large the upload is.
453async fn parse_multipart(
454    incoming: Incoming,
455    boundary: &str,
456    limits: &BodyLimits,
457) -> Result<MultipartForm> {
458    let stream = incoming.into_data_stream();
459    let mut multipart = multer::Multipart::new(stream, boundary);
460    let mut form = MultipartForm::new();
461    let mut total: u64 = 0;
462
463    while let Some(field) = multipart.next_field().await.map_err(multer_error)? {
464        let field_name = field.name().unwrap_or_default().to_string();
465        if field_name.is_empty() {
466            continue;
467        }
468        let file_name = field.file_name().map(str::to_string);
469        let content_type = field.content_type().map(|m| m.to_string());
470
471        match file_name {
472            Some(file_name) => {
473                let mut builder = FormFileBuilder::new(field_name.clone(), file_name, content_type)
474                    .memory_threshold(limits.memory_threshold)
475                    .max_size(limits.file)
476                    .spool_dir(limits.spool_dir.clone());
477
478                let mut field = field;
479                while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
480                    total = charge(&mut total, chunk.len(), limits.multipart_body)?;
481                    builder.write(&chunk).await?;
482                }
483                form.push_file(builder.finish().await?);
484            }
485            None => {
486                let mut value: Vec<u8> = Vec::new();
487                let mut field = field;
488                while let Some(chunk) = field.chunk().await.map_err(multer_error)? {
489                    total = charge(&mut total, chunk.len(), limits.multipart_body)?;
490                    if value.len() + chunk.len() > limits.field {
491                        return Err(Error::PayloadTooLarge(format!(
492                            "form field `{field_name}` exceeds the {}-byte limit",
493                            limits.field
494                        )));
495                    }
496                    value.extend_from_slice(&chunk);
497                }
498                let text = String::from_utf8(value).map_err(|_| {
499                    Error::Validation(format!("form field `{field_name}` is not valid UTF-8"))
500                })?;
501                form.push_field(field_name, text);
502            }
503        }
504    }
505
506    Ok(form)
507}
508
509/// Add `len` to the running total, failing when the request limit is exceeded.
510fn charge(total: &mut u64, len: usize, limit: u64) -> Result<u64> {
511    *total = total.saturating_add(len as u64);
512    if *total > limit {
513        return Err(Error::PayloadTooLarge(format!(
514            "multipart request exceeds the {limit}-byte limit"
515        )));
516    }
517    Ok(*total)
518}
519
520fn multer_error(err: multer::Error) -> Error {
521    match err {
522        multer::Error::FieldSizeExceeded { limit, .. } => {
523            Error::PayloadTooLarge(format!("multipart field exceeds the {limit}-byte limit"))
524        }
525        multer::Error::StreamSizeExceeded { limit } => {
526            Error::PayloadTooLarge(format!("multipart request exceeds the {limit}-byte limit"))
527        }
528        multer::Error::IncompleteFieldData { .. } | multer::Error::IncompleteStream => {
529            Error::Validation("multipart body ended unexpectedly".to_string())
530        }
531        other => Error::Validation(format!("invalid multipart body: {other}")),
532    }
533}
534
535// ---------------------------------------------------------------------------
536// Response
537// ---------------------------------------------------------------------------
538
539/// Concrete implementation of IHttpResponse.
540pub struct HttpResponse {
541    status: u16,
542    headers: HashMap<String, String>,
543    body: Option<ResponseBody>,
544}
545
546impl HttpResponse {
547    pub fn new(status: u16) -> Self {
548        Self {
549            status,
550            headers: HashMap::new(),
551            body: None,
552        }
553    }
554}
555
556#[async_trait::async_trait]
557impl IHttpResponse for HttpResponse {
558    fn status(&self) -> u16 {
559        self.status
560    }
561
562    fn has_body(&self) -> bool {
563        self.body.is_some()
564    }
565
566    fn set_status(&mut self, code: u16) {
567        self.status = code;
568    }
569
570    fn set_header(&mut self, key: &str, value: &str) {
571        self.headers.insert(key.to_string(), value.to_string());
572    }
573
574    fn remove_header(&mut self, key: &str) {
575        self.headers.remove(key);
576    }
577
578    async fn write_bytes(&mut self, data: Vec<u8>) -> Result<()> {
579        self.body = Some(ResponseBody::Bytes(data));
580        Ok(())
581    }
582
583    async fn write_text(&mut self, text: &str) -> Result<()> {
584        self.body = Some(ResponseBody::Bytes(text.as_bytes().to_vec()));
585        Ok(())
586    }
587
588    async fn write_body(&mut self, body: ResponseBody) -> Result<()> {
589        self.body = Some(body);
590        Ok(())
591    }
592
593    fn body_bytes(&self) -> Vec<u8> {
594        self.body
595            .as_ref()
596            .and_then(ResponseBody::as_bytes)
597            .map(<[u8]>::to_vec)
598            .unwrap_or_default()
599    }
600
601    fn header(&self, key: &str) -> Option<&str> {
602        self.headers.get(key).map(|s| s.as_str())
603    }
604}
605
606// ---------------------------------------------------------------------------
607// Response finalisation
608// ---------------------------------------------------------------------------
609
610async fn finalize_response(req: &HttpRequest, resp: HttpResponse) -> hyper::Response<RespBody> {
611    let HttpResponse {
612        status,
613        mut headers,
614        body,
615    } = resp;
616
617    match body.unwrap_or(ResponseBody::Bytes(Vec::new())) {
618        ResponseBody::Bytes(bytes) => {
619            default_cache_control(&mut headers, false);
620            bytes_response(status, headers, bytes)
621        }
622        ResponseBody::File(file) => {
623            default_cache_control(&mut headers, true);
624            file_response(req, status, headers, file).await
625        }
626    }
627}
628
629/// Apply the framework's default `Cache-Control` when the response has none.
630///
631/// Ordinary API responses must not be stored; a file response with validators is
632/// exactly the case where `no-store` would defeat the `ETag` we just sent, so it
633/// revalidates instead. Anything the application set explicitly wins.
634fn default_cache_control(headers: &mut HashMap<String, String>, is_file: bool) {
635    if headers.contains_key("cache-control") {
636        return;
637    }
638    headers.insert(
639        "cache-control".to_string(),
640        if is_file {
641            "public, max-age=0, must-revalidate".to_string()
642        } else {
643            "no-store".to_string()
644        },
645    );
646}
647
648/// A file body whose source has been resolved to something streamable.
649enum PreparedSource {
650    /// An open file handle; seeks to serve ranges.
651    Path(tokio::fs::File),
652    /// A reader that can seek, so ranges are served by seeking.
653    Seekable(Box<dyn SeekableReader>),
654    /// A reader of unknown length; chunked, no ranges.
655    Stream(Box<dyn AsyncRead + Send + Unpin>),
656}
657
658impl PreparedSource {
659    fn is_seekable(&self) -> bool {
660        matches!(self, PreparedSource::Path(_) | PreparedSource::Seekable(_))
661    }
662}
663
664/// Serve a file response with validators, conditional handling and byte ranges.
665///
666/// Works the same way for a path, a seekable stream and a plain stream; the only
667/// differences are which HTTP features the source can support.
668async fn file_response(
669    req: &HttpRequest,
670    status: u16,
671    mut headers: HashMap<String, String>,
672    file: FileBody,
673) -> hyper::Response<RespBody> {
674    let FileBody {
675        source,
676        content_type,
677        file_name,
678        disposition,
679        enable_range_processing,
680        last_modified,
681        entity_tag,
682    } = file;
683
684    // ── Resolve the source, its length and its validators ──
685    let mut modified = last_modified;
686    let mut etag = entity_tag;
687    let mut source_path: Option<PathBuf> = None;
688    let known_len;
689
690    let prepared = match source {
691        FileSource::Path(path) => {
692            // Open once and take length/mtime from the same handle: one syscall
693            // pair instead of `metadata()` followed by `open()`.
694            let handle = match tokio::fs::File::open(&path).await {
695                Ok(handle) => handle,
696                Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
697                    return problem_bytes(404, headers, "Not Found")
698                }
699                Err(err) => {
700                    tracing::error!(
701                        path = %path.display(),
702                        error = %err,
703                        "cannot open file for download"
704                    );
705                    return problem_bytes(500, headers, "an unexpected error occurred");
706                }
707            };
708            let metadata = match handle.metadata().await {
709                Ok(metadata) => metadata,
710                Err(err) => {
711                    tracing::error!(
712                        path = %path.display(),
713                        error = %err,
714                        "cannot stat file for download"
715                    );
716                    return problem_bytes(500, headers, "an unexpected error occurred");
717                }
718            };
719            if !metadata.is_file() {
720                return problem_bytes(404, headers, "Not Found");
721            }
722
723            let len = metadata.len();
724            if modified.is_none() {
725                modified = metadata.modified().ok();
726            }
727            if etag.is_none() {
728                etag = Some(etag_for(len, modified));
729            }
730            known_len = Some(len);
731            source_path = Some(path);
732            PreparedSource::Path(handle)
733        }
734        FileSource::SeekableStream { reader, len } => {
735            known_len = Some(len);
736            PreparedSource::Seekable(reader)
737        }
738        FileSource::Stream(reader) => {
739            known_len = None;
740            PreparedSource::Stream(reader)
741        }
742    };
743
744    // RFC 7232 §2.2.2: an origin server must not send a `Last-Modified` later
745    // than the response generation time, and must not treat such a date as a
746    // precondition. Restored archives and clock skew produce future mtimes, and
747    // they would otherwise make `If-Modified-Since` and date-based `If-Range`
748    // answer nonsense. The ETag above was already derived from the real mtime,
749    // so content changes are still detected.
750    let modified = modified.filter(|value| *value <= SystemTime::now());
751
752    // ── Representation metadata ──
753    if !headers.contains_key("content-type") {
754        let resolved = content_type
755            .or_else(|| source_path.as_deref().and_then(guess_content_type))
756            .unwrap_or_else(|| "application/octet-stream".to_string());
757        headers.insert("content-type".to_string(), resolved);
758    }
759    if !headers.contains_key("content-disposition") {
760        if let Some(name) = &file_name {
761            headers.insert(
762                "content-disposition".to_string(),
763                content_disposition_value(disposition, name),
764            );
765        }
766    }
767    if let Some(tag) = &etag {
768        headers.insert("etag".to_string(), tag.clone());
769    }
770    if let Some(modified) = modified {
771        headers.insert(
772            "last-modified".to_string(),
773            httpdate::fmt_http_date(modified),
774        );
775    }
776
777    // `Accept-Ranges: none` tells download managers not to bother trying, which
778    // is more useful than the header's absence (absence means "unknown").
779    let range_capable = enable_range_processing && prepared.is_seekable() && known_len.is_some();
780    headers.insert(
781        "accept-ranges".to_string(),
782        if range_capable { "bytes" } else { "none" }.to_string(),
783    );
784
785    // ── Conditional GET: the representation is unchanged ──
786    if status == 200 && is_not_modified(req, etag.as_deref(), modified) {
787        headers.remove("content-length");
788        return bytes_response(304, headers, Vec::new());
789    }
790
791    // ── Byte ranges ──
792    let mut ranges: Vec<ByteRange> = Vec::new();
793    if status == 200 && range_capable {
794        let len = known_len.expect("range-capable sources know their length");
795        if let Some(raw) = req.header("range") {
796            // `If-Range` gates the whole range request: if the client's copy is
797            // stale, the correct answer is the *full* representation, not a
798            // range from a different version.
799            if if_range_matches(req, etag.as_deref(), modified) {
800                match parse_ranges(raw, len) {
801                    RangeOutcome::Ignored => {}
802                    RangeOutcome::Unsatisfiable => {
803                        headers.insert("content-range".to_string(), format!("bytes */{len}"));
804                        return problem_bytes(416, headers, "Range Not Satisfiable");
805                    }
806                    RangeOutcome::Satisfiable(parsed) => ranges = parsed,
807                }
808            }
809        }
810    }
811
812    // ── Body ──
813    //
814    // A multi-range request needs to re-read the same source from several
815    // offsets. Only a file handle can do that; a range-capable *stream* cannot
816    // be restarted, so those requests are answered with the full representation
817    // (RFC 7233 §3.1 lets a server ignore `Range`).
818    if ranges.len() > 1 && !matches!(prepared, PreparedSource::Path(_)) {
819        ranges.clear();
820    }
821    if ranges.len() > 1 {
822        return multipart_byteranges_response(prepared, headers, ranges, known_len);
823    }
824
825    let (status, start, count) = match ranges.first() {
826        Some(range) => {
827            let len = known_len.expect("range-capable sources know their length");
828            headers.insert(
829                "content-range".to_string(),
830                format!("bytes {}-{}/{len}", range.start, range.end),
831            );
832            (206, range.start, range.len())
833        }
834        None => (status, 0, known_len.unwrap_or(0)),
835    };
836
837    match prepared {
838        PreparedSource::Path(mut handle) => {
839            headers.insert("content-length".to_string(), count.to_string());
840            if count == 0 {
841                return bytes_response(status, headers, Vec::new());
842            }
843            if start > 0 {
844                if let Err(err) = handle.seek(SeekFrom::Start(start)).await {
845                    tracing::error!(error = %err, "cannot seek file for ranged response");
846                    return problem_bytes(500, headers, "an unexpected error occurred");
847                }
848            }
849            build_response(status, headers, reader_body(handle.take(count)))
850        }
851        PreparedSource::Seekable(mut reader) => {
852            headers.insert("content-length".to_string(), count.to_string());
853            if count == 0 {
854                return bytes_response(status, headers, Vec::new());
855            }
856            if start > 0 {
857                if let Err(err) = reader.seek(SeekFrom::Start(start)).await {
858                    tracing::error!(error = %err, "cannot seek stream for ranged response");
859                    return problem_bytes(500, headers, "an unexpected error occurred");
860                }
861            }
862            build_response(status, headers, reader_body(reader.take(count)))
863        }
864        PreparedSource::Stream(reader) => {
865            // Unknown length: chunked transfer encoding, no `Content-Length`.
866            // Nothing is buffered.
867            build_response(status, headers, reader_body(reader))
868        }
869    }
870}
871
872/// The media type of the parts inside a `multipart/byteranges` response.
873///
874/// The response-level `Content-Type` is about to be replaced, so the original
875/// representation type has to be captured first.
876fn representation_content_type(headers: &HashMap<String, String>) -> String {
877    headers
878        .get("content-type")
879        .filter(|value| !value.starts_with("multipart/byteranges"))
880        .cloned()
881        .unwrap_or_else(|| "application/octet-stream".to_string())
882}
883
884/// Serve several disjoint ranges as one `multipart/byteranges` response.
885///
886/// Ranges are coalesced before this point, so the parts are disjoint and the
887/// total payload never exceeds the file size — no traffic amplification. The
888/// response length is computed exactly rather than falling back to chunked,
889/// because every part's size is known up front.
890fn multipart_byteranges_response(
891    prepared: PreparedSource,
892    mut headers: HashMap<String, String>,
893    ranges: Vec<ByteRange>,
894    known_len: Option<u64>,
895) -> hyper::Response<RespBody> {
896    let PreparedSource::Path(handle) = prepared else {
897        unreachable!("multi-range is only routed here for path sources");
898    };
899
900    let total_len = known_len.unwrap_or(0);
901    let media_type = representation_content_type(&headers);
902    let boundary = format!("webx-{}", uuid::Uuid::new_v4().simple());
903
904    // Pre-render each part's header block so the total length is exact by
905    // construction and the stream only has to emit bytes.
906    let parts: Vec<(Bytes, ByteRange)> = ranges
907        .iter()
908        .map(|range| {
909            let header = format!(
910                "--{boundary}\r\nContent-Type: {media_type}\r\nContent-Range: bytes {}-{}/{total_len}\r\n\r\n",
911                range.start, range.end
912            );
913            (Bytes::from(header), *range)
914        })
915        .collect();
916    let closing = Bytes::from(format!("--{boundary}--\r\n"));
917
918    let content_length: u64 = parts
919        .iter()
920        .map(|(header, range)| header.len() as u64 + range.len() + 2)
921        .sum::<u64>()
922        + closing.len() as u64;
923
924    headers.insert(
925        "content-type".to_string(),
926        format!("multipart/byteranges; boundary={boundary}"),
927    );
928    headers.insert("content-length".to_string(), content_length.to_string());
929    // `Content-Range` is per part here, so the message must not carry one.
930    headers.remove("content-range");
931
932    let stream = multipart_stream(handle, parts, closing);
933    build_response(
934        206,
935        headers,
936        BodyExt::boxed_unsync(http_body_util::StreamBody::new(stream)),
937    )
938}
939
940/// Chunk size for each `multipart/byteranges` part read.
941const MULTIPART_CHUNK: usize = STREAM_BUF_SIZE;
942
943/// State carried between polls of the `multipart/byteranges` stream.
944struct MultipartState {
945    handle: tokio::fs::File,
946    parts: std::vec::IntoIter<(Bytes, ByteRange)>,
947    closing: Bytes,
948    /// Header bytes still to emit for the current part.
949    pending_header: Bytes,
950    phase: MultipartPhase,
951    buffer: bytes::BytesMut,
952}
953
954enum MultipartPhase {
955    /// Move to the next part's header, or close the body.
956    NextPart,
957    /// Stream the current part's payload.
958    Payload(u64),
959    /// Emit the `\r\n` that terminates a part.
960    Terminator,
961    /// Emit the closing delimiter.
962    Closing,
963    Done,
964}
965
966/// Build the `multipart/byteranges` body as a stream of frames.
967///
968/// Written as a `Stream` rather than a hand-rolled `AsyncRead`: the byte layout
969/// is a sequence of independent chunks, so `unfold` expresses it directly and
970/// there is no `poll_*` plumbing to get wrong.
971fn multipart_stream(
972    handle: tokio::fs::File,
973    parts: Vec<(Bytes, ByteRange)>,
974    closing: Bytes,
975) -> impl futures_util::Stream<Item = std::io::Result<Frame<Bytes>>> + Send {
976    let state = MultipartState {
977        handle,
978        parts: parts.into_iter(),
979        closing,
980        pending_header: Bytes::new(),
981        phase: MultipartPhase::NextPart,
982        buffer: bytes::BytesMut::new(),
983    };
984
985    futures_util::stream::unfold(state, |mut state| async move {
986        loop {
987            match std::mem::replace(&mut state.phase, MultipartPhase::Done) {
988                MultipartPhase::NextPart => match state.parts.next() {
989                    Some((header, range)) => {
990                        if let Err(err) = state.handle.seek(SeekFrom::Start(range.start)).await {
991                            return Some((Err(err), state));
992                        }
993                        state.pending_header = header;
994                        state.phase = MultipartPhase::Payload(range.len());
995                    }
996                    None => state.phase = MultipartPhase::Closing,
997                },
998                MultipartPhase::Payload(remaining) => {
999                    if remaining == 0 {
1000                        state.phase = MultipartPhase::Terminator;
1001                        continue;
1002                    }
1003
1004                    // Read at most one chunk, never past the part boundary.
1005                    let limit = MULTIPART_CHUNK.min(remaining as usize);
1006                    state.buffer.clear();
1007                    state.buffer.reserve(limit);
1008                    let read = match state.handle.read_buf(&mut state.buffer).await {
1009                        Ok(read) => read,
1010                        Err(err) => return Some((Err(err), state)),
1011                    };
1012                    if read == 0 {
1013                        return Some((
1014                            Err(std::io::Error::new(
1015                                std::io::ErrorKind::UnexpectedEof,
1016                                "file changed size while serving byte ranges",
1017                            )),
1018                            state,
1019                        ));
1020                    }
1021                    let keep = read.min(limit);
1022                    state.buffer.truncate(keep);
1023                    let chunk = state.buffer.split().freeze();
1024
1025                    // The part header is emitted immediately before its first
1026                    // payload byte, so a part is one contiguous run of frames.
1027                    let header = std::mem::take(&mut state.pending_header);
1028                    let frame = if header.is_empty() {
1029                        chunk
1030                    } else {
1031                        let mut combined =
1032                            bytes::BytesMut::with_capacity(header.len() + chunk.len());
1033                        combined.extend_from_slice(&header);
1034                        combined.extend_from_slice(&chunk);
1035                        combined.freeze()
1036                    };
1037
1038                    state.phase = MultipartPhase::Payload(remaining - keep as u64);
1039                    return Some((Ok(Frame::data(frame)), state));
1040                }
1041                MultipartPhase::Terminator => {
1042                    state.phase = MultipartPhase::NextPart;
1043                    return Some((Ok(Frame::data(Bytes::from_static(b"\r\n"))), state));
1044                }
1045                MultipartPhase::Closing => {
1046                    state.phase = MultipartPhase::Done;
1047                    let closing = std::mem::take(&mut state.closing);
1048                    return Some((Ok(Frame::data(closing)), state));
1049                }
1050                MultipartPhase::Done => return None,
1051            }
1052        }
1053    })
1054}
1055
1056/// Read buffer used for every streamed response body.
1057///
1058/// Shared with the upload path so both sides of a transfer use the same size.
1059const STREAM_BUF_SIZE: usize = webx_core::http::STREAM_BUF_SIZE;
1060
1061/// Turn any async reader into a streaming response body.
1062fn reader_body<R>(reader: R) -> RespBody
1063where
1064    R: tokio::io::AsyncRead + Send + 'static,
1065{
1066    let stream = tokio_util::io::ReaderStream::with_capacity(reader, STREAM_BUF_SIZE);
1067    let body = http_body_util::StreamBody::new(stream.map_ok(Frame::data));
1068    BodyExt::boxed_unsync(body)
1069}
1070
1071fn bytes_response(
1072    status: u16,
1073    headers: HashMap<String, String>,
1074    bytes: Vec<u8>,
1075) -> hyper::Response<RespBody> {
1076    build_response(status, headers, full_body(bytes))
1077}
1078
1079fn full_body(bytes: Vec<u8>) -> RespBody {
1080    Full::new(Bytes::from(bytes))
1081        .map_err(|never: std::convert::Infallible| match never {})
1082        .boxed_unsync()
1083}
1084
1085/// Build an RFC 7807 `application/problem+json` response for a file error.
1086fn problem_bytes(
1087    status: u16,
1088    mut headers: HashMap<String, String>,
1089    detail: &str,
1090) -> hyper::Response<RespBody> {
1091    headers.insert(
1092        "content-type".to_string(),
1093        "application/problem+json".to_string(),
1094    );
1095    bytes_response(
1096        status,
1097        headers,
1098        problem_to_bytes(&build_problem(status, detail)),
1099    )
1100}
1101
1102fn build_response(
1103    status: u16,
1104    headers: HashMap<String, String>,
1105    body: RespBody,
1106) -> hyper::Response<RespBody> {
1107    let mut builder = hyper::Response::builder().status(status);
1108    for (key, value) in &headers {
1109        builder = builder.header(key.as_str(), value.as_str());
1110    }
1111    builder.body(body).unwrap_or_else(|_| {
1112        hyper::Response::builder()
1113            .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)
1114            .body(full_body(b"Internal Server Error".to_vec()))
1115            .expect("static error response is valid")
1116    })
1117}
1118
1119/// Result of parsing a `Range` header.
1120#[derive(Debug, PartialEq, Eq)]
1121enum RangeOutcome {
1122    /// Syntactically invalid: the whole header must be ignored (RFC 7233 §2.1).
1123    Ignored,
1124    /// Valid but nothing overlaps the representation: answer `416`.
1125    Unsatisfiable,
1126    /// One or more disjoint, coalesced ranges, in ascending order.
1127    Satisfiable(Vec<ByteRange>),
1128}
1129
1130/// Maximum number of parts to answer in one `multipart/byteranges` response.
1131///
1132/// Ranges are coalesced first, so the parts are disjoint and the total payload
1133/// can never exceed the file size; this cap only bounds the *framing* overhead
1134/// and the per-part seeks. A client asking for more gets the full
1135/// representation, which RFC 7233 §3.1 permits.
1136const MAX_RANGE_PARTS: usize = 8;
1137
1138/// Parse a byte-range-set (RFC 7233 §2.1) and coalesce it.
1139///
1140/// Coalescing matters for more than tidiness: without it a client could ask for
1141/// the same range eight times and make the server send the file eight times
1142/// over. Merging overlapping and adjacent ranges guarantees the parts are
1143/// disjoint, so the response can never be larger than the file.
1144fn parse_ranges(value: &str, len: u64) -> RangeOutcome {
1145    let value = value.trim();
1146    let Some(spec) = strip_prefix_ignore_ascii_case(value, "bytes=") else {
1147        return RangeOutcome::Ignored;
1148    };
1149
1150    let mut ranges: Vec<ByteRange> = Vec::new();
1151    let mut saw_valid_spec = false;
1152
1153    for part in spec.split(',') {
1154        let part = part.trim();
1155        if part.is_empty() {
1156            continue;
1157        }
1158        match parse_single_range(part, len) {
1159            // One malformed member invalidates the whole header.
1160            SingleRange::Malformed => return RangeOutcome::Ignored,
1161            SingleRange::Unsatisfiable => saw_valid_spec = true,
1162            SingleRange::Satisfiable(range) => {
1163                saw_valid_spec = true;
1164                ranges.push(range);
1165            }
1166        }
1167    }
1168
1169    if ranges.is_empty() {
1170        return if saw_valid_spec {
1171            RangeOutcome::Unsatisfiable
1172        } else {
1173            RangeOutcome::Ignored
1174        };
1175    }
1176
1177    ranges.sort_by_key(|range| range.start);
1178    let mut merged: Vec<ByteRange> = Vec::with_capacity(ranges.len());
1179    for range in ranges {
1180        match merged.last_mut() {
1181            // `+ 1` merges adjacent ranges too (`0-9,10-19` is one read).
1182            Some(last) if range.start <= last.end.saturating_add(1) => {
1183                last.end = last.end.max(range.end);
1184            }
1185            _ => merged.push(range),
1186        }
1187    }
1188
1189    if merged.len() > MAX_RANGE_PARTS {
1190        return RangeOutcome::Ignored;
1191    }
1192
1193    RangeOutcome::Satisfiable(merged)
1194}
1195
1196enum SingleRange {
1197    /// Not a valid `byte-range-spec`.
1198    Malformed,
1199    /// Valid syntax, but outside the representation.
1200    Unsatisfiable,
1201    Satisfiable(ByteRange),
1202}
1203
1204/// Parse one `first-byte-pos "-" [last-byte-pos]` or `"-" suffix-length`.
1205fn parse_single_range(spec: &str, len: u64) -> SingleRange {
1206    let Some((start_text, end_text)) = spec.split_once('-') else {
1207        return SingleRange::Malformed;
1208    };
1209    let (start_text, end_text) = (start_text.trim(), end_text.trim());
1210
1211    if start_text.is_empty() {
1212        // Suffix range: the last N bytes.
1213        let Ok(suffix) = end_text.parse::<u64>() else {
1214            return SingleRange::Malformed;
1215        };
1216        if suffix == 0 || len == 0 {
1217            return SingleRange::Unsatisfiable;
1218        }
1219        return SingleRange::Satisfiable(ByteRange {
1220            start: len.saturating_sub(suffix),
1221            end: len - 1,
1222        });
1223    }
1224
1225    let Ok(start) = start_text.parse::<u64>() else {
1226        return SingleRange::Malformed;
1227    };
1228    if start >= len {
1229        return SingleRange::Unsatisfiable;
1230    }
1231
1232    let end = if end_text.is_empty() {
1233        len - 1
1234    } else {
1235        match end_text.parse::<u64>() {
1236            Ok(end) => end.min(len - 1),
1237            Err(_) => return SingleRange::Malformed,
1238        }
1239    };
1240
1241    if end < start {
1242        return SingleRange::Unsatisfiable;
1243    }
1244
1245    SingleRange::Satisfiable(ByteRange { start, end })
1246}
1247
1248fn strip_prefix_ignore_ascii_case<'a>(value: &'a str, prefix: &str) -> Option<&'a str> {
1249    if value.len() >= prefix.len() && value[..prefix.len()].eq_ignore_ascii_case(prefix) {
1250        Some(&value[prefix.len()..])
1251    } else {
1252        None
1253    }
1254}
1255
1256/// A weak validator derived from size and modification time.
1257///
1258/// This is the same scheme nginx and most static servers use: free to compute,
1259/// and it changes whenever the file does. Full `SystemTime` precision is used so
1260/// an in-place edit that keeps the size is still detected within the same
1261/// second — `Last-Modified` cannot express that, but an `ETag` can.
1262///
1263/// It is emitted as a **strong** tag because download resumption needs one:
1264/// RFC 7233 §3.2 requires a strong validator for `If-Range`, so a weak tag would
1265/// silently disable resumable downloads. Callers that need a byte-exact
1266/// validator should set one explicitly with `FileBody::entity_tag`.
1267fn etag_for(len: u64, modified: Option<SystemTime>) -> String {
1268    match modified {
1269        Some(modified) => format!("\"{:x}-{:x}\"", len, epoch_nanos(modified)),
1270        None => format!("\"{len:x}\""),
1271    }
1272}
1273
1274fn epoch_secs(time: SystemTime) -> u64 {
1275    time.duration_since(UNIX_EPOCH)
1276        .map(|d| d.as_secs())
1277        .unwrap_or(0)
1278}
1279
1280fn epoch_nanos(time: SystemTime) -> u128 {
1281    time.duration_since(UNIX_EPOCH)
1282        .map(|d| d.as_nanos())
1283        .unwrap_or(0)
1284}
1285
1286/// Whether an entity-tag is weak (`W/"..."`).
1287fn is_weak_etag(tag: &str) -> bool {
1288    tag.trim_start().starts_with("W/")
1289}
1290
1291/// Strip the weakness marker so two tags can be compared by value.
1292fn etag_value(tag: &str) -> &str {
1293    tag.trim().trim_start_matches("W/")
1294}
1295
1296/// Whether the client's cached representation is still current.
1297///
1298/// Uses **weak** comparison, per RFC 7232 §3.2: `If-None-Match` only asks
1299/// whether the content is semantically the same, so a weak tag is enough. A
1300/// response without validators can never answer `304`.
1301fn is_not_modified(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1302    if let Some(if_none_match) = req.header("if-none-match") {
1303        let Some(etag) = etag else {
1304            return false;
1305        };
1306        let expected = etag_value(etag);
1307        return if_none_match.split(',').any(|candidate| {
1308            let candidate = candidate.trim();
1309            candidate == "*" || etag_value(candidate) == expected
1310        });
1311    }
1312
1313    if let Some(if_modified_since) = req.header("if-modified-since") {
1314        if let Some(modified) = modified {
1315            if let Ok(since) = httpdate::parse_http_date(if_modified_since) {
1316                return epoch_secs(modified) <= epoch_secs(since);
1317            }
1318        }
1319    }
1320
1321    false
1322}
1323
1324/// Whether a `Range` header should be honoured, per RFC 7233 §3.2.
1325///
1326/// `If-Range` is the precondition that makes resumption safe, so it is strict:
1327///
1328/// * an entity-tag must match by **strong** comparison — both tags must be
1329///   strong and byte-equal. A weak tag (ours or the client's) can never match,
1330///   because a weak validator does not promise byte-for-byte identity;
1331/// * a date must be at least as new as `Last-Modified`.
1332///
1333/// Returning `false` means "ignore `Range` and send the full representation".
1334fn if_range_matches(req: &HttpRequest, etag: Option<&str>, modified: Option<SystemTime>) -> bool {
1335    let Some(value) = req.header("if-range") else {
1336        return true;
1337    };
1338    let value = value.trim();
1339
1340    if value.starts_with('"') || value.starts_with("W/") {
1341        let Some(etag) = etag else {
1342            return false;
1343        };
1344        if is_weak_etag(value) || is_weak_etag(etag) {
1345            return false;
1346        }
1347        return value == etag.trim();
1348    }
1349
1350    match (httpdate::parse_http_date(value), modified) {
1351        (Ok(instant), Some(modified)) => epoch_secs(instant) >= epoch_secs(modified),
1352        _ => false,
1353    }
1354}
1355
1356fn guess_content_type(path: &Path) -> Option<String> {
1357    mime_guess::from_path(path)
1358        .first()
1359        .map(|mime| mime.essence_str().to_string())
1360}
1361
1362// ---------------------------------------------------------------------------
1363// Query string helpers
1364// ---------------------------------------------------------------------------
1365
1366/// Parse a query string like "key1=val1&key2=val2" into both views.
1367///
1368/// Returns the last value per key (the compatibility view used by
1369/// `IHttpRequest::query`) and every value per key (used for `Vec<T>` fields).
1370/// Repeated keys keep their order; empty pairs are ignored.
1371fn parse_query_string(query: &str) -> (HashMap<String, String>, HashMap<String, Vec<String>>) {
1372    let mut flat = HashMap::new();
1373    let mut multi: HashMap<String, Vec<String>> = HashMap::new();
1374    for pair in query.split('&') {
1375        if pair.is_empty() {
1376            continue;
1377        }
1378        let mut parts = pair.splitn(2, '=');
1379        if let (Some(key), Some(value)) = (parts.next(), parts.next()) {
1380            let key = percent_decode(key);
1381            let value = percent_decode(value);
1382            flat.insert(key.clone(), value.clone());
1383            multi.entry(key).or_default().push(value);
1384        }
1385    }
1386    (flat, multi)
1387}
1388
1389/// Decode `%XX` escapes and `+` into a UTF-8 string.
1390///
1391/// Escapes are decoded as bytes and then validated as UTF-8, so multi-byte
1392/// characters (`%E4%B8%AD` → `中`) survive; the previous byte-per-char loop
1393/// mangled anything outside ASCII.
1394fn percent_decode(s: &str) -> String {
1395    if !s.contains('%') && !s.contains('+') {
1396        return s.to_string();
1397    }
1398
1399    let mut bytes = Vec::with_capacity(s.len());
1400    let mut chars = s.bytes();
1401    while let Some(byte) = chars.next() {
1402        match byte {
1403            b'%' => {
1404                let hi = chars.next().and_then(|c| (c as char).to_digit(16));
1405                let lo = chars.next().and_then(|c| (c as char).to_digit(16));
1406                match (hi, lo) {
1407                    (Some(hi), Some(lo)) => bytes.push((hi * 16 + lo) as u8),
1408                    _ => bytes.push(b'%'),
1409                }
1410            }
1411            b'+' => bytes.push(b' '),
1412            other => bytes.push(other),
1413        }
1414    }
1415    String::from_utf8_lossy(&bytes).into_owned()
1416}
1417
1418#[cfg(test)]
1419mod tests {
1420    use super::*;
1421
1422    #[test]
1423    fn parses_the_boundary_parameter() {
1424        assert_eq!(
1425            multipart_boundary("multipart/form-data; boundary=----abc123").as_deref(),
1426            Some("----abc123")
1427        );
1428        assert_eq!(
1429            multipart_boundary("multipart/form-data; charset=utf-8; boundary=\"quoted\"")
1430                .as_deref(),
1431            Some("quoted")
1432        );
1433        assert_eq!(multipart_boundary("multipart/form-data"), None);
1434    }
1435
1436    fn ranges(start: u64, end: u64) -> RangeOutcome {
1437        RangeOutcome::Satisfiable(vec![ByteRange { start, end }])
1438    }
1439
1440    #[test]
1441    fn parses_byte_ranges() {
1442        assert_eq!(parse_ranges("bytes=0-99", 1000), ranges(0, 99));
1443        assert_eq!(parse_ranges("bytes=500-", 1000), ranges(500, 999));
1444        assert_eq!(parse_ranges("bytes=-200", 1000), ranges(800, 999));
1445        // Clamped to the end of the representation.
1446        assert_eq!(parse_ranges("bytes=0-99999", 1000), ranges(0, 999));
1447        // Case-insensitive unit, and whitespace around members is tolerated.
1448        assert_eq!(parse_ranges("BYTES=0-9", 1000), ranges(0, 9));
1449        assert_eq!(parse_ranges("bytes= 0-9 ", 1000), ranges(0, 9));
1450    }
1451
1452    #[test]
1453    fn rejects_unsatisfiable_and_malformed_ranges() {
1454        assert_eq!(
1455            parse_ranges("bytes=1000-", 1000),
1456            RangeOutcome::Unsatisfiable
1457        );
1458        assert_eq!(parse_ranges("bytes=-0", 1000), RangeOutcome::Unsatisfiable);
1459        assert_eq!(
1460            parse_ranges("bytes=500-100", 1000),
1461            RangeOutcome::Unsatisfiable
1462        );
1463        assert_eq!(parse_ranges("items=0-10", 1000), RangeOutcome::Ignored);
1464        assert_eq!(parse_ranges("bytes=", 1000), RangeOutcome::Ignored);
1465        // RFC 7233 §2.1: one malformed member invalidates the whole header.
1466        assert_eq!(parse_ranges("bytes=abc-def", 1000), RangeOutcome::Ignored);
1467        assert_eq!(
1468            parse_ranges("bytes=0-9,abc-def", 1000),
1469            RangeOutcome::Ignored
1470        );
1471    }
1472
1473    #[test]
1474    fn suffix_range_on_empty_file_is_unsatisfiable() {
1475        assert_eq!(parse_ranges("bytes=-10", 0), RangeOutcome::Unsatisfiable);
1476    }
1477
1478    #[test]
1479    fn multi_range_is_coalesced_into_disjoint_parts() {
1480        // Disjoint ranges stay separate, in ascending order.
1481        assert_eq!(
1482            parse_ranges("bytes=20-30,0-10", 1000),
1483            RangeOutcome::Satisfiable(vec![
1484                ByteRange { start: 0, end: 10 },
1485                ByteRange { start: 20, end: 30 },
1486            ])
1487        );
1488
1489        // Overlapping ranges merge, so a client cannot ask for the same bytes
1490        // twice and make the server send them twice.
1491        assert_eq!(parse_ranges("bytes=0-99,50-149", 1000), ranges(0, 149));
1492
1493        // Adjacent ranges merge too — one read instead of two.
1494        assert_eq!(parse_ranges("bytes=0-9,10-19", 1000), ranges(0, 19));
1495
1496        // Duplicates collapse completely: no traffic amplification.
1497        assert_eq!(
1498            parse_ranges("bytes=0-,0-,0-,0-,0-,0-,0-,0-", 1000),
1499            ranges(0, 999)
1500        );
1501
1502        // An unsatisfiable member next to a satisfiable one is dropped, not fatal.
1503        assert_eq!(parse_ranges("bytes=0-9,5000-6000", 1000), ranges(0, 9));
1504    }
1505
1506    #[test]
1507    fn too_many_parts_falls_back_to_the_full_representation() {
1508        let mut spec = String::from("bytes=");
1509        for index in 0..(MAX_RANGE_PARTS + 1) {
1510            if index > 0 {
1511                spec.push(',');
1512            }
1513            spec.push_str(&format!("{}-{}", index * 4, index * 4 + 1));
1514        }
1515        assert_eq!(parse_ranges(&spec, 1000), RangeOutcome::Ignored);
1516    }
1517
1518    #[test]
1519    fn etag_changes_with_size_and_mtime() {
1520        let base = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
1521        assert_eq!(etag_for(10, Some(base)), etag_for(10, Some(base)));
1522        assert_ne!(etag_for(10, Some(base)), etag_for(11, Some(base)));
1523        assert_ne!(
1524            etag_for(10, Some(base)),
1525            etag_for(10, Some(base + std::time::Duration::from_secs(1)))
1526        );
1527        // Sub-second precision matters: an in-place edit that keeps the size is
1528        // invisible to `Last-Modified`, but must change the ETag. (1 µs rather
1529        // than 1 ns, because Windows filesystem timestamps are 100 ns ticks.)
1530        assert_ne!(
1531            etag_for(10, Some(base)),
1532            etag_for(10, Some(base + std::time::Duration::from_micros(1)))
1533        );
1534    }
1535
1536    #[test]
1537    fn etag_strength_rules() {
1538        assert!(!is_weak_etag("\"abc\""));
1539        assert!(is_weak_etag("W/\"abc\""));
1540        assert_eq!(etag_value("W/\"abc\""), "\"abc\"");
1541        assert_eq!(etag_value("\"abc\""), "\"abc\"");
1542    }
1543
1544    #[test]
1545    fn parses_query_strings() {
1546        let (params, multi) = parse_query_string("a=1&b=hello+world&c=%2Ftmp");
1547        assert_eq!(params.get("a").map(String::as_str), Some("1"));
1548        assert_eq!(params.get("b").map(String::as_str), Some("hello world"));
1549        assert_eq!(params.get("c").map(String::as_str), Some("/tmp"));
1550        assert_eq!(
1551            multi.get("a").map(Vec::as_slice),
1552            Some(&["1".to_string()][..])
1553        );
1554    }
1555
1556    #[test]
1557    fn keeps_every_query_value_for_repeated_keys() {
1558        let (params, multi) = parse_query_string("id=1&id=2&id=3");
1559        assert_eq!(params.get("id").map(String::as_str), Some("3"));
1560        assert_eq!(
1561            multi.get("id").map(Vec::as_slice),
1562            Some(&["1".to_string(), "2".to_string(), "3".to_string()][..])
1563        );
1564    }
1565
1566    #[test]
1567    fn percent_decodes_utf8() {
1568        // Multi-byte characters must survive: the byte-per-char loop turned
1569        // `%E4%B8%AD` into mojibake.
1570        let (params, _) = parse_query_string("q=%E4%B8%AD%E6%96%87&x=%2B");
1571        assert_eq!(params.get("q").map(String::as_str), Some("中文"));
1572        assert_eq!(params.get("x").map(String::as_str), Some("+"));
1573    }
1574}