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