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