Skip to main content

rust_web_server/proxy_config/
mod.rs

1//! Config-driven proxy application.
2//!
3//! When `rws.config.toml` contains `[[route]]` or `[[upstream]]` sections,
4//! `ConfigDrivenApp` is used as the top-level `Application` instead of the
5//! hardcoded `build_app()` in `main.rs`.
6//!
7//! # Quick start
8//!
9//! ```toml
10//! # rws.config.toml
11//! [[upstream]]
12//! name = "api"
13//! backends = ["localhost:3000"]
14//!
15//! [[route]]
16//! name = "api-proxy"
17//!
18//! [route.match]
19//! path = "/api/*"
20//!
21//! [route.action]
22//! type = "proxy"
23//!
24//! [route.action.proxy]
25//! upstream = "api"
26//! ```
27
28pub mod parser;
29pub mod health;
30pub(crate) mod upstream;
31
32pub use upstream::{BreakerConfig, RetryConfig, StickyConfig};
33pub mod builder;
34pub mod validate;
35
36#[cfg(test)]
37mod tests;
38
39use std::sync::atomic::{AtomicBool, Ordering};
40use std::sync::{Arc, RwLock};
41
42use crate::app::App;
43use crate::application::Application;
44use crate::core::New;
45use crate::request::Request;
46use crate::header::Header;
47use crate::response::{Response, STATUS_CODE_REASON_PHRASE};
48use crate::server::ConnectionInfo;
49use crate::server_config::ServerConfig;
50
51// ── Public config types ────────────────────────────────────────────────────────
52
53#[derive(Debug, Clone)]
54pub struct ProxyConfig {
55    pub upstreams: Vec<UpstreamConfig>,
56    pub routes: Vec<RouteConfig>,
57    pub tcp_proxies: Vec<TcpProxyConfig>,
58    pub udp_proxies: Vec<UdpProxyConfig>,
59    pub ws_proxies: Vec<WsProxyConfig>,
60    pub global_middleware: MiddlewareConfig,
61}
62
63#[derive(Debug, Clone)]
64pub struct UpstreamConfig {
65    pub name: String,
66    pub backends: Vec<String>,
67    pub strategy: String, // "round_robin" | "random" | "ip_hash"
68    pub health_check: Option<HealthCheckConfig>,
69    /// `true` when all backends use `https://` scheme — connections to the
70    /// upstream are made over TLS. Requires the `http-client` or `http2`
71    /// feature (which bring in `rustls` + `webpki-roots`).
72    pub tls: bool,
73    /// `tls_ca_file` — PEM file of CA certificates to trust for this
74    /// upstream's `https://` backends **instead of** the public WebPKI roots
75    /// (a private CA or a self-signed certificate). `None` = WebPKI.
76    pub tls_ca_file: Option<String>,
77    /// `tls_client_cert_file` / `tls_client_key_file` — a client certificate
78    /// (PEM chain and key) presented to TLS backends that require one.
79    /// Since 18.3.0.
80    pub tls_client_cert_file: Option<String>,
81    pub tls_client_key_file: Option<String>,
82    /// `weights` — one per backend, in order (default 1 each). Since 18.6.0.
83    pub weights: Vec<u32>,
84    /// `[upstream.sticky]` — keep a client on one backend. Since 18.6.0.
85    pub sticky: Option<StickyConfig>,
86    /// `[upstream.circuit_breaker]` — take failing backends out of rotation
87    /// (passive health checks). Since 18.6.0.
88    pub circuit_breaker: Option<BreakerConfig>,
89    /// `[upstream.retry]` — try another backend. Since 18.6.0.
90    pub retry: Option<RetryConfig>,
91}
92
93#[derive(Debug, Clone)]
94pub struct HealthCheckConfig {
95    pub path: String,
96    pub interval_secs: u64,
97    pub timeout_ms: u64,
98    pub healthy_threshold: u32,
99    pub unhealthy_threshold: u32,
100}
101
102#[derive(Debug, Clone)]
103pub struct RouteConfig {
104    pub name: String,
105    pub match_: MatchConfig,
106    pub action: ActionConfig,
107    pub middleware: MiddlewareConfig,
108}
109
110#[derive(Debug, Clone, Default)]
111pub struct MatchConfig {
112    pub host: Option<String>,
113    pub path: Option<String>,
114    pub method: Option<String>,
115    pub content_type: Option<String>,
116}
117
118#[derive(Debug, Clone)]
119pub enum ActionConfig {
120    Proxy {
121        upstream: String,
122        connect_timeout_ms: u64,
123        read_timeout_ms: u64,
124        strip_path_prefix: Option<String>,
125        add_path_prefix: Option<String>,
126    },
127    Grpc {
128        upstream: String,
129        connect_timeout_ms: u64,
130        read_timeout_ms: u64,
131    },
132    Static {
133        root: String,
134        index: Vec<String>,
135    },
136    Redirect {
137        location: String,
138        status: u16,
139    },
140    Respond {
141        status: u16,
142        body: String,
143        content_type: String,
144    },
145    Mcp,
146    Unknown(String),
147}
148
149#[derive(Debug, Clone, Default)]
150pub struct MiddlewareConfig {
151    pub rate_limit: Option<RateLimitConfig>,
152    pub cache: Option<CacheConfig>,
153    pub auth: Option<AuthConfig>,
154    pub rewrite_request: Vec<RewriteRuleConfig>,
155    pub rewrite_response: Vec<RewriteRuleConfig>,
156    pub ip_allow: Vec<String>,
157    pub ip_deny: Vec<String>,
158    /// `timeout_ms` — if the route (including all its other middleware)
159    /// doesn't produce a response within this many milliseconds, the client
160    /// gets `504 Gateway Timeout` instead of waiting further. See
161    /// `crate::timeout` for the underlying mechanism and its limitations.
162    pub timeout_ms: Option<u64>,
163    /// `max_body_size` — reject this route's request with `413 Payload Too
164    /// Large` if its body exceeds this many bytes.
165    ///
166    /// This is a *route-specific* ceiling on top of the global, process-wide
167    /// `RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES` (see `crate::entry_point::get_max_body_size`)
168    /// — e.g. a small JSON API route can be capped stricter than a
169    /// file-upload route without lowering the global limit for everyone.
170    /// It is **not** a substitute for the global limit: by the time any
171    /// route-level middleware runs, `Request` (including its body) has
172    /// already been fully read off the socket — route matching itself
173    /// requires the complete request, so a per-route check can only run
174    /// after that memory is already spent, not before. The global limit is
175    /// what protects total server memory from an oversized upload;
176    /// `max_body_size` protects only this route's own logic/upstream from
177    /// an oversized-but-still-within-the-global-limit body.
178    pub max_body_size: Option<u64>,
179}
180
181#[derive(Debug, Clone)]
182pub struct RateLimitConfig {
183    pub max_requests: u32,
184    pub window_secs: u64,
185}
186
187#[derive(Debug, Clone)]
188pub struct CacheConfig {
189    pub ttl_secs: u64,
190    pub vary_by: Vec<String>,
191}
192
193#[derive(Debug, Clone)]
194pub enum AuthConfig {
195    /// `auth = { type = "basic", htpasswd_file = ".htpasswd" }`. See
196    /// `crate::auth::BasicAuthLayer::from_htpasswd_file` for the supported
197    /// file format (plain text and `{SHA256}` only — not Apache's `{SHA}`,
198    /// `$apr1$`, or bcrypt).
199    Basic { htpasswd_file: String },
200    /// `auth = { type = "jwt", secret_env = "JWT_SECRET" }`. Requires the
201    /// `auth` feature; verifies HS256 JWTs via `crate::auth::JwtLayer`.
202    Jwt { secret_env: String },
203    Bearer { token_env: String },
204}
205
206#[derive(Debug, Clone, Default)]
207pub struct RewriteRuleConfig {
208    pub type_: String,
209    pub name: Option<String>,
210    pub value: Option<String>,
211    pub prefix: Option<String>,
212    pub from: Option<String>,
213    pub to: Option<String>,
214    pub code: Option<u16>,
215    pub reason: Option<String>,
216}
217
218#[derive(Debug, Clone)]
219pub struct TcpProxyConfig {
220    pub name: String,
221    pub listen: String,
222    pub backends: Vec<String>,
223    pub connect_timeout_ms: u64,
224}
225
226#[derive(Debug, Clone)]
227pub struct UdpProxyConfig {
228    pub name: String,
229    pub listen: String,
230    pub backends: Vec<String>,
231    pub reply_timeout_ms: u64,
232    pub buffer_size: usize,
233}
234
235#[derive(Debug, Clone)]
236pub struct WsProxyConfig {
237    pub name: String,
238    pub listen: String,
239    pub backends: Vec<String>,
240    pub connect_timeout_ms: u64,
241    pub read_timeout_ms: u64,
242    /// Optional `[ws_proxy.health_check]` — same shape and semantics as
243    /// `[upstream.health_check]`. The probe is a plain HTTP `GET {path}`
244    /// against the backend's `host:port` (the `ws://`/`wss://` scheme only
245    /// determines whether the probe connects over TLS); most WebSocket
246    /// backends serve a regular HTTP health endpoint alongside their upgrade
247    /// route, the same way nginx/Traefik health-check WS upstreams.
248    pub health_check: Option<HealthCheckConfig>,
249}
250
251// ── ProxyConfig loading ────────────────────────────────────────────────────────
252
253
254/// An action's settings may sit directly under `[route.action]` next to
255/// `type` (the short form README.md and the docs use), or in a nested
256/// `[route.action.<type>]` table (the long form). Copy short-form keys into
257/// the nested table so the parser reads one place; a key set in both keeps
258/// the nested value.
259fn merge_action_keys(map: &mut parser::SectionMap, action_section: &str, action_type: &str) {
260    if action_type.is_empty() {
261        return;
262    }
263    let short_form: Vec<(String, String)> = match map.get(action_section) {
264        Some(entries) => entries.iter().filter(|(k, _)| k != "type").cloned().collect(),
265        None => return,
266    };
267    let nested = map.entry(format!("{action_section}.{action_type}")).or_default();
268    for (key, value) in short_form {
269        if !nested.iter().any(|(k, _)| *k == key) {
270            nested.push((key, value));
271        }
272    }
273}
274
275impl ProxyConfig {
276    /// Returns `true` if `rws.config.toml` (or `RWS_CONFIG_FILE`) contains
277    /// `[[route]]` or `[[upstream]]` sections, meaning config-driven mode
278    /// should be used.
279    pub fn is_proxy_mode() -> bool {
280        let path = config_file_path();
281        match std::fs::read_to_string(&path) {
282            Ok(contents) => {
283                contents.contains("[[route]]") || contents.contains("[[upstream]]")
284            }
285            Err(_) => false,
286        }
287    }
288
289    /// Parse the config file and return a `ProxyConfig`.
290    pub fn load() -> Self {
291        let path = config_file_path();
292        let contents = std::fs::read_to_string(&path).unwrap_or_default();
293        Self::from_str(&contents)
294    }
295
296    /// Parse `toml` text directly into a `ProxyConfig`. Used in tests.
297    pub fn from_str(toml: &str) -> Self {
298        use parser::{get_array, get_str, get_u32, get_u64, section_exists};
299
300        let mut map = parser::parse(toml);
301
302        // ── upstreams ──────────────────────────────────────────────────────────
303        let mut upstreams = Vec::new();
304        let mut i = 0;
305        loop {
306            let sec = format!("upstream[{}]", i);
307            if !section_exists(&map, &sec) {
308                break;
309            }
310            let name = get_str(&map, &sec, "name");
311            let backends = get_array(&map, &sec, "backends");
312            let strategy = {
313                let s = get_str(&map, &sec, "strategy");
314                if s.is_empty() { "round_robin".to_string() } else { s }
315            };
316            let hc_sec = format!("{}.health_check", sec);
317            let health_check = if section_exists(&map, &hc_sec) {
318                Some(HealthCheckConfig {
319                    path: {
320                        let p = get_str(&map, &hc_sec, "path");
321                        if p.is_empty() { "/health".to_string() } else { p }
322                    },
323                    interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
324                    timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
325                    healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
326                    unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
327                })
328            } else {
329                None
330            };
331            let tls = backends.iter().any(|b| ["https://", "h2s://", "grpcs://"].iter().any(|s| b.starts_with(s)));
332            let tls_ca_file = Some(get_str(&map, &sec, "tls_ca_file")).filter(|v| !v.is_empty());
333            let tls_client_cert_file = Some(get_str(&map, &sec, "tls_client_cert_file")).filter(|v| !v.is_empty());
334            let tls_client_key_file = Some(get_str(&map, &sec, "tls_client_key_file")).filter(|v| !v.is_empty());
335            let weights: Vec<u32> = get_array(&map, &sec, "weights").iter().map(|w| w.trim().parse().unwrap_or(0)).collect();
336            let sticky_sec = format!("{}.sticky", sec);
337            let sticky = section_exists(&map, &sticky_sec).then(|| StickyConfig {
338                cookie: Some(get_str(&map, &sticky_sec, "cookie")).filter(|c| !c.is_empty()).unwrap_or_else(|| "rws_backend".to_string()),
339                max_age_secs: Some(get_u64(&map, &sticky_sec, "max_age_secs", 0)).filter(|s| *s > 0),
340            });
341            let cb_sec = format!("{}.circuit_breaker", sec);
342            let circuit_breaker = section_exists(&map, &cb_sec).then(|| BreakerConfig {
343                failure_threshold: get_u32(&map, &cb_sec, "failure_threshold", 5),
344                recovery_secs: get_u64(&map, &cb_sec, "recovery_secs", 30),
345                half_open_probes: get_u32(&map, &cb_sec, "half_open_probes", 1),
346            });
347            let retry_sec = format!("{}.retry", sec);
348            let retry = section_exists(&map, &retry_sec).then(|| {
349                let status: Vec<i16> = get_array(&map, &retry_sec, "status").iter().filter_map(|s| s.trim().parse().ok()).collect();
350                RetryConfig { attempts: get_u32(&map, &retry_sec, "attempts", 1), status: if status.is_empty() { vec![502, 503, 504] } else { status } }
351            });
352            upstreams.push(UpstreamConfig {
353                name, backends, strategy, health_check, tls, tls_ca_file, tls_client_cert_file, tls_client_key_file,
354                weights, sticky, circuit_breaker, retry,
355            });
356            i += 1;
357        }
358
359        // ── routes ─────────────────────────────────────────────────────────────
360        let mut routes = Vec::new();
361        let mut i = 0;
362        loop {
363            let sec = format!("route[{}]", i);
364            if !section_exists(&map, &sec) {
365                break;
366            }
367            let name = get_str(&map, &sec, "name");
368
369            // match
370            let m_sec = format!("{}.match", sec);
371            let match_ = MatchConfig {
372                host: {
373                    let h = get_str(&map, &m_sec, "host");
374                    if h.is_empty() { None } else { Some(h) }
375                },
376                path: {
377                    let p = get_str(&map, &m_sec, "path");
378                    if p.is_empty() { None } else { Some(p) }
379                },
380                method: {
381                    let m = get_str(&map, &m_sec, "method");
382                    if m.is_empty() { None } else { Some(m.to_uppercase()) }
383                },
384                content_type: {
385                    let c = get_str(&map, &m_sec, "content_type");
386                    if c.is_empty() { None } else { Some(c) }
387                },
388            };
389
390            // action
391            let a_sec = format!("{}.action", sec);
392            let action_type = get_str(&map, &a_sec, "type");
393            merge_action_keys(&mut map, &a_sec, &action_type);
394            let action = match action_type.as_str() {
395                "proxy" => {
396                    let p_sec = format!("{}.action.proxy", sec);
397                    ActionConfig::Proxy {
398                        upstream: get_str(&map, &p_sec, "upstream"),
399                        connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
400                        read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
401                        strip_path_prefix: {
402                            let v = get_str(&map, &p_sec, "strip_path_prefix");
403                            if v.is_empty() { None } else { Some(v) }
404                        },
405                        add_path_prefix: {
406                            let v = get_str(&map, &p_sec, "add_path_prefix");
407                            if v.is_empty() { None } else { Some(v) }
408                        },
409                    }
410                }
411                "grpc" => {
412                    let p_sec = format!("{}.action.grpc", sec);
413                    ActionConfig::Grpc {
414                        upstream: get_str(&map, &p_sec, "upstream"),
415                        connect_timeout_ms: get_u64(&map, &p_sec, "connect_timeout_ms", 5000),
416                        read_timeout_ms: get_u64(&map, &p_sec, "read_timeout_ms", 30000),
417                    }
418                }
419                "static" => {
420                    let s_sec = format!("{}.action.static", sec);
421                    ActionConfig::Static {
422                        root: get_str(&map, &s_sec, "root"),
423                        index: get_array(&map, &s_sec, "index"),
424                    }
425                }
426                "redirect" => {
427                    let r_sec = format!("{}.action.redirect", sec);
428                    ActionConfig::Redirect {
429                        location: get_str(&map, &r_sec, "location"),
430                        status: get_u64(&map, &r_sec, "status", 301) as u16,
431                    }
432                }
433                "respond" => {
434                    let r_sec = format!("{}.action.respond", sec);
435                    ActionConfig::Respond {
436                        status: get_u64(&map, &r_sec, "status", 200) as u16,
437                        body: get_str(&map, &r_sec, "body"),
438                        content_type: {
439                            let ct = get_str(&map, &r_sec, "content_type");
440                            if ct.is_empty() { "text/plain".to_string() } else { ct }
441                        },
442                    }
443                }
444                "mcp" => ActionConfig::Mcp,
445                other => ActionConfig::Unknown(other.to_string()),
446            };
447
448            // middleware
449            let mw_sec = format!("{}.middleware", sec);
450            let middleware = parse_middleware_config(&map, &mw_sec, i);
451
452            routes.push(RouteConfig { name, match_, action, middleware });
453            i += 1;
454        }
455
456        // ── tcp_proxy ──────────────────────────────────────────────────────────
457        let mut tcp_proxies = Vec::new();
458        let mut i = 0;
459        loop {
460            let sec = format!("tcp_proxy[{}]", i);
461            if !section_exists(&map, &sec) {
462                break;
463            }
464            tcp_proxies.push(TcpProxyConfig {
465                name: get_str(&map, &sec, "name"),
466                listen: get_str(&map, &sec, "listen"),
467                backends: get_array(&map, &sec, "backends"),
468                connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
469            });
470            i += 1;
471        }
472
473        // ── udp_proxy ──────────────────────────────────────────────────────────
474        let mut udp_proxies = Vec::new();
475        let mut i = 0;
476        loop {
477            let sec = format!("udp_proxy[{}]", i);
478            if !section_exists(&map, &sec) {
479                break;
480            }
481            udp_proxies.push(UdpProxyConfig {
482                name: get_str(&map, &sec, "name"),
483                listen: get_str(&map, &sec, "listen"),
484                backends: get_array(&map, &sec, "backends"),
485                reply_timeout_ms: get_u64(&map, &sec, "reply_timeout_ms", 5000),
486                buffer_size: get_u64(&map, &sec, "buffer_size", 65536) as usize,
487            });
488            i += 1;
489        }
490
491        // ── ws_proxy ───────────────────────────────────────────────────────────
492        let mut ws_proxies = Vec::new();
493        let mut i = 0;
494        loop {
495            let sec = format!("ws_proxy[{}]", i);
496            if !section_exists(&map, &sec) {
497                break;
498            }
499            let hc_sec = format!("{}.health_check", sec);
500            let health_check = if section_exists(&map, &hc_sec) {
501                Some(HealthCheckConfig {
502                    path: {
503                        let p = get_str(&map, &hc_sec, "path");
504                        if p.is_empty() { "/health".to_string() } else { p }
505                    },
506                    interval_secs: get_u64(&map, &hc_sec, "interval_secs", 30),
507                    timeout_ms: get_u64(&map, &hc_sec, "timeout_ms", 5000),
508                    healthy_threshold: get_u32(&map, &hc_sec, "healthy_threshold", 2),
509                    unhealthy_threshold: get_u32(&map, &hc_sec, "unhealthy_threshold", 3),
510                })
511            } else {
512                None
513            };
514            ws_proxies.push(WsProxyConfig {
515                name: get_str(&map, &sec, "name"),
516                listen: get_str(&map, &sec, "listen"),
517                backends: get_array(&map, &sec, "backends"),
518                connect_timeout_ms: get_u64(&map, &sec, "connect_timeout_ms", 5000),
519                read_timeout_ms: get_u64(&map, &sec, "read_timeout_ms", 30000),
520                health_check,
521            });
522            i += 1;
523        }
524
525        // ── global middleware ──────────────────────────────────────────────────
526        let global_middleware = parse_middleware_config(&map, "middleware", usize::MAX);
527
528        ProxyConfig {
529            upstreams,
530            routes,
531            tcp_proxies,
532            udp_proxies,
533            ws_proxies,
534            global_middleware,
535        }
536    }
537}
538
539/// Parse a `MiddlewareConfig` from the section map at a given base path.
540/// `route_idx` is used only to build inner-array section paths for rewrite rules.
541fn parse_middleware_config(
542    map: &parser::SectionMap,
543    mw_sec: &str,
544    route_idx: usize,
545) -> MiddlewareConfig {
546    use parser::{get_array, get_str, get_u32, get_u64, section_exists};
547
548    let rl_sec = format!("{}.rate_limit", mw_sec);
549    let rate_limit = if section_exists(map, &rl_sec) {
550        Some(RateLimitConfig {
551            max_requests: get_u32(map, &rl_sec, "max_requests", 1000),
552            window_secs: get_u64(map, &rl_sec, "window_secs", 60),
553        })
554    } else {
555        None
556    };
557
558    let c_sec = format!("{}.cache", mw_sec);
559    let cache = if section_exists(map, &c_sec) {
560        Some(CacheConfig {
561            ttl_secs: get_u64(map, &c_sec, "ttl_secs", 60),
562            vary_by: get_array(map, &c_sec, "vary_by"),
563        })
564    } else {
565        None
566    };
567
568    let a_sec = format!("{}.auth", mw_sec);
569    let auth = if section_exists(map, &a_sec) {
570        let auth_type = get_str(map, &a_sec, "type");
571        match auth_type.as_str() {
572            "bearer" => Some(AuthConfig::Bearer {
573                token_env: get_str(map, &a_sec, "token_env"),
574            }),
575            "jwt" => Some(AuthConfig::Jwt {
576                secret_env: get_str(map, &a_sec, "secret_env"),
577            }),
578            "basic" => Some(AuthConfig::Basic {
579                htpasswd_file: get_str(map, &a_sec, "htpasswd_file"),
580            }),
581            _ => None,
582        }
583    } else {
584        None
585    };
586
587    // Rewrite rules — the section paths use route_idx for route-scoped rules
588    // or a flat path for global middleware. We look for:
589    //   route[N].middleware.rewrite.request[0], [1], …
590    //   route[N].middleware.rewrite.response[0], [1], …
591    // For global: middleware.rewrite.request[0], etc.
592    let rewrite_request = collect_rewrite_rules(map, mw_sec, "request");
593    let rewrite_response = collect_rewrite_rules(map, mw_sec, "response");
594
595    let ip_sec = format!("{}.ip_filter", mw_sec);
596    let ip_allow = if section_exists(map, &ip_sec) {
597        get_array(map, &ip_sec, "allow")
598    } else {
599        vec![]
600    };
601    let ip_deny = if section_exists(map, &ip_sec) {
602        get_array(map, &ip_sec, "deny")
603    } else {
604        vec![]
605    };
606
607    let _ = route_idx; // used implicitly via mw_sec paths
608
609    // Flat scalar directly under [route.middleware] (or the global
610    // [middleware] table), not a nested sub-table like rate_limit/cache —
611    // 0/absent both mean "no timeout configured".
612    let timeout_ms = match get_u64(map, mw_sec, "timeout_ms", 0) {
613        0 => None,
614        ms => Some(ms),
615    };
616
617    // Same flat-scalar convention as timeout_ms — 0/absent means "no
618    // route-specific limit" (the global RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES
619    // still applies regardless).
620    let max_body_size = match get_u64(map, mw_sec, "max_body_size", 0) {
621        0 => None,
622        n => Some(n),
623    };
624
625    MiddlewareConfig {
626        rate_limit, cache, auth, rewrite_request, rewrite_response, ip_allow, ip_deny,
627        timeout_ms, max_body_size,
628    }
629}
630
631/// Collect `[[{mw_sec}.rewrite.{direction}]]` entries.
632fn collect_rewrite_rules(
633    map: &parser::SectionMap,
634    mw_sec: &str,
635    direction: &str,
636) -> Vec<RewriteRuleConfig> {
637    use parser::{get_str, get_u64};
638
639    let mut rules = Vec::new();
640    let mut j = 0;
641    loop {
642        let rsec = format!("{}.rewrite.{}[{}]", mw_sec, direction, j);
643        if !parser::section_exists(map, &rsec) {
644            break;
645        }
646        let code_val = get_u64(map, &rsec, "code", 0);
647        rules.push(RewriteRuleConfig {
648            type_: get_str(map, &rsec, "type"),
649            name: {
650                let v = get_str(map, &rsec, "name");
651                if v.is_empty() { None } else { Some(v) }
652            },
653            value: {
654                let v = get_str(map, &rsec, "value");
655                if v.is_empty() { None } else { Some(v) }
656            },
657            prefix: {
658                let v = get_str(map, &rsec, "prefix");
659                if v.is_empty() { None } else { Some(v) }
660            },
661            from: {
662                let v = get_str(map, &rsec, "from");
663                if v.is_empty() { None } else { Some(v) }
664            },
665            to: {
666                let v = get_str(map, &rsec, "to");
667                if v.is_empty() { None } else { Some(v) }
668            },
669            code: if code_val == 0 { None } else { Some(code_val as u16) },
670            reason: {
671                let v = get_str(map, &rsec, "reason");
672                if v.is_empty() { None } else { Some(v) }
673            },
674        });
675        j += 1;
676    }
677    rules
678}
679
680pub(crate) fn config_file_path() -> String {
681    std::env::var("RWS_CONFIG_FILE").unwrap_or_else(|_| "rws.config.toml".to_string())
682}
683
684// ── ConfigDrivenApp ────────────────────────────────────────────────────────────
685
686/// A compiled route: a matcher paired with a handler application.
687pub(crate) struct CompiledRoute {
688    pub(crate) matcher: RouteMatcher,
689    /// Shared, type-erased handler. `Arc` makes `Clone` cheap (pointer copy).
690    pub(crate) handler: Arc<dyn Application + Send + Sync>,
691    /// The route's proxy when it can be served without blocking a thread
692    /// (plain-HTTP upstream, no route middleware) — see
693    /// [`DynamicProxy::execute_async`].
694    #[cfg_attr(not(feature = "http2"), allow(dead_code))]
695    pub(crate) async_proxy: Option<Arc<DynamicProxy>>,
696}
697
698/// Matching criteria for a single route.
699#[derive(Clone, Default)]
700pub(crate) struct RouteMatcher {
701    /// Optional SNI hostname / `Host` header match.
702    pub(crate) host: Option<String>,
703    /// Path prefix to match (derived from `path = "/v1/*"`).
704    pub(crate) path_prefix: Option<String>,
705    /// Exact path to match (derived from `path = "/v1/ping"`).
706    pub(crate) path_exact: Option<String>,
707    /// Uppercase HTTP method, or `None` for any.
708    pub(crate) method: Option<String>,
709    /// `Content-Type` prefix (e.g. `"application/grpc"`).
710    pub(crate) content_type_prefix: Option<String>,
711}
712
713impl RouteMatcher {
714    pub(crate) fn from_match_config(cfg: &MatchConfig) -> Self {
715        let (path_prefix, path_exact) = match &cfg.path {
716            Some(p) if p.ends_with('*') => {
717                // "/v1/*" → prefix "/v1/"
718                let stripped = p.trim_end_matches('*').to_string();
719                (Some(stripped), None)
720            }
721            Some(p) => (None, Some(p.clone())),
722            None => (None, None),
723        };
724        let content_type_prefix = cfg.content_type.as_ref().map(|ct| {
725            if ct.ends_with('*') {
726                ct.trim_end_matches('*').to_string()
727            } else {
728                ct.clone()
729            }
730        });
731        RouteMatcher {
732            host: cfg.host.clone(),
733            path_prefix,
734            path_exact,
735            method: cfg.method.clone(),
736            content_type_prefix,
737        }
738    }
739
740    /// Returns `true` if `request` and `conn` match all configured criteria.
741    pub(crate) fn matches(&self, request: &Request, conn: &ConnectionInfo) -> bool {
742        // Host matching: SNI first, then Host header
743        if let Some(ref expected_host) = self.host {
744            let actual_host = conn
745                .sni_hostname
746                .as_deref()
747                .or_else(|| {
748                    request
749                        .headers
750                        .iter()
751                        .find(|h| h.name.eq_ignore_ascii_case("host"))
752                        .map(|h| h.value.as_ref())
753                })
754                .unwrap_or("");
755            if !host_matches(expected_host, actual_host) {
756                return false;
757            }
758        }
759
760        // Method matching
761        if let Some(ref m) = self.method {
762            if request.method.to_uppercase() != m.as_str() {
763                return false;
764            }
765        }
766
767        // Path matching: strip query string for comparison
768        let path = request.request_uri.split('?').next().unwrap_or(&request.request_uri);
769        if let Some(ref prefix) = self.path_prefix {
770            if !path.starts_with(prefix.as_str()) {
771                return false;
772            }
773        } else if let Some(ref exact) = self.path_exact {
774            if path != exact.as_str() {
775                return false;
776            }
777        }
778
779        // Content-Type prefix matching
780        if let Some(ref ct_prefix) = self.content_type_prefix {
781            let actual_ct = request
782                .headers
783                .iter()
784                .find(|h| h.name.eq_ignore_ascii_case("content-type"))
785                .map(|h| h.value.as_ref())
786                .unwrap_or("");
787            if !actual_ct.starts_with(ct_prefix.as_str()) {
788                return false;
789            }
790        }
791
792        true
793    }
794}
795
796/// An `Application` that routes requests based on a parsed `ProxyConfig`.
797///
798/// Clones share one live configuration: a hot reload
799/// ([`ConfigDrivenApp::reload_from_str`], run on `SIGHUP` for an app built by
800/// [`build_from_file`]) swaps it for every clone at once. In-flight requests
801/// finish on the configuration they started with.
802#[derive(Clone)]
803pub struct ConfigDrivenApp {
804    current: Arc<RwLock<Arc<Generation>>>,
805    /// Serializes reloads: one SIGHUP reaches every accept loop.
806    reload_lock: Arc<std::sync::Mutex<()>>,
807}
808
809/// One loaded configuration: its compiled routes plus what a reload needs.
810struct Generation {
811    dispatch: RouteDispatch,
812    /// `dispatch` wrapped in the top-level `[middleware]` table; `None` when
813    /// that table is empty.
814    entry: Option<Arc<dyn Application + Send + Sync>>,
815    config: ProxyConfig,
816    /// The file text this generation was built from, to skip no-op reloads.
817    source: Option<String>,
818    /// Set when this generation is replaced: stops its health checkers.
819    health_stop: Arc<AtomicBool>,
820}
821
822impl Generation {
823    fn new(routes: Arc<Vec<CompiledRoute>>, fallback: App, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
824        let dispatch = RouteDispatch { routes, fallback };
825        let entry = if config.global_middleware.is_empty() {
826            None
827        } else {
828            Some(builder::apply_middleware(arc_app(dispatch.clone()), &config.global_middleware))
829        };
830        Generation { dispatch, entry, config, source, health_stop }
831    }
832}
833
834/// First-match routing over the compiled routes, then the fallback.
835#[derive(Clone)]
836struct RouteDispatch {
837    routes: Arc<Vec<CompiledRoute>>,
838    /// Fallback for unmatched requests — handles /healthz, /readyz, /metrics,
839    /// static files, and the 404 controller. Reads `RWS_CONFIG_*` env vars
840    /// per request (`App::new()`'s default) unless pinned via
841    /// [`ConfigDrivenApp::with_config`].
842    fallback: App,
843}
844
845impl Application for RouteDispatch {
846    fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
847        for route in self.routes.iter() {
848            if route.matcher.matches(request, conn) {
849                return route.handler.execute(request, conn);
850            }
851        }
852        self.fallback.execute(request, conn)
853    }
854
855    fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
856        match self.routes.iter().find(|r| r.matcher.matches(request, conn)) {
857            Some(route) => route.handler.streams_body(request, conn),
858            None => self.fallback.streams_body(request, conn),
859        }
860    }
861}
862
863impl MiddlewareConfig {
864    /// `true` when no layer is configured.
865    pub fn is_empty(&self) -> bool {
866        self.rate_limit.is_none() && self.cache.is_none() && self.auth.is_none()
867            && self.rewrite_request.is_empty() && self.rewrite_response.is_empty()
868            && self.ip_allow.is_empty() && self.ip_deny.is_empty()
869            && self.timeout_ms.is_none() && self.max_body_size.is_none()
870    }
871}
872
873/// What a successful [`ConfigDrivenApp::reload_from_str`] changed.
874#[derive(Debug, Clone, PartialEq)]
875pub enum ReloadOutcome {
876    /// The configuration text is unchanged; nothing was rebuilt.
877    Unchanged,
878    /// New routes are live. `warnings` includes `[[tcp_proxy]]` /
879    /// `[[udp_proxy]]` / `[[ws_proxy]]` changes, which need a restart.
880    Reloaded { routes: usize, warnings: Vec<String> },
881}
882
883impl ConfigDrivenApp {
884    pub(crate) fn new(routes: Vec<CompiledRoute>, config: ProxyConfig, health_stop: Arc<AtomicBool>, source: Option<String>) -> Self {
885        use crate::core::New;
886        let generation = Generation::new(Arc::new(routes), App::new(), config, health_stop, source);
887        ConfigDrivenApp {
888            current: Arc::new(RwLock::new(Arc::new(generation))),
889            reload_lock: Arc::new(std::sync::Mutex::new(())),
890        }
891    }
892
893    fn current(&self) -> Arc<Generation> {
894        Arc::clone(&self.current.read().unwrap_or_else(|e| e.into_inner()))
895    }
896
897    /// Pin the fallback [`App`] (used for any request none of the
898    /// config-driven routes match) to an explicit [`ServerConfig`], instead
899    /// of reading `RWS_CONFIG_*` environment variables per request.
900    ///
901    /// Mirrors [`App::with_config`] / [`crate::state::AppWithState::with_config`]
902    /// — same rationale: safe for parallel tests, and lets multiple
903    /// differently-configured proxy instances coexist in one process. The
904    /// pinned fallback is kept across hot reloads.
905    ///
906    /// ```rust,no_run
907    /// use rust_web_server::proxy_config::build_from_file;
908    /// use rust_web_server::server_config::ServerConfig;
909    ///
910    /// let (app, _handles) = build_from_file();
911    /// let app = app.with_config(ServerConfig::default());
912    /// ```
913    pub fn with_config(self, config: ServerConfig) -> Self {
914        {
915            let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
916            let old = Arc::clone(&slot);
917            *slot = Arc::new(Generation::new(
918                Arc::clone(&old.dispatch.routes),
919                App::with_config(config),
920                old.config.clone(),
921                Arc::clone(&old.health_stop),
922                old.source.clone(),
923            ));
924        }
925        self
926    }
927
928    /// Replace the routes, upstreams and `[middleware]` with those in `toml`.
929    ///
930    /// The new configuration is validated first ([`ProxyConfig::validate`]
931    /// plus unknown keys); if it has errors, the running configuration stays
932    /// in place and the errors are returned — a typo never takes the proxy
933    /// down. Otherwise the new routes go live for every clone of this app,
934    /// the previous configuration's health checkers stop, and new ones start.
935    /// `[[tcp_proxy]]` / `[[udp_proxy]]` / `[[ws_proxy]]` listeners are not
936    /// rebound; a change to them is reported as a warning.
937    pub fn reload_from_str(&self, toml: &str) -> Result<ReloadOutcome, crate::config_check::ConfigReport> {
938        let _serialized = self.reload_lock.lock().unwrap_or_else(|e| e.into_inner());
939        let old = self.current();
940        if old.source.as_deref() == Some(toml) {
941            return Ok(ReloadOutcome::Unchanged);
942        }
943        let config = ProxyConfig::from_str(toml);
944        let mut report = crate::config_check::ConfigReport::default();
945        validate::check_unknown_keys(toml, &mut report);
946        config.validate(&mut report);
947        if !report.is_ok() {
948            return Err(report);
949        }
950
951        let mut warnings = report.warnings;
952        let listeners = |c: &ProxyConfig| format!("{:?}{:?}{:?}", c.tcp_proxies, c.udp_proxies, c.ws_proxies);
953        if listeners(&config) != listeners(&old.config) {
954            warnings.push("[[tcp_proxy]] / [[udp_proxy]] / [[ws_proxy]] changes need a restart; the old listeners keep running".to_string());
955        }
956
957        let health_stop = Arc::new(AtomicBool::new(false));
958        let routes = builder::compile_routes(&config, &health_stop);
959        let route_count = routes.len();
960        let generation = Generation::new(Arc::new(routes), old.dispatch.fallback.clone(), config, health_stop, Some(toml.to_string()));
961        let replaced = {
962            let mut slot = self.current.write().unwrap_or_else(|e| e.into_inner());
963            std::mem::replace(&mut *slot, Arc::new(generation))
964        };
965        replaced.health_stop.store(true, Ordering::Relaxed);
966        Ok(ReloadOutcome::Reloaded { routes: route_count, warnings })
967    }
968
969    /// Re-read the configuration file (`RWS_CONFIG_FILE` or
970    /// `rws.config.toml`) and [`reload_from_str`](Self::reload_from_str) it,
971    /// logging the outcome. Registered as a [`crate::config_reload`] hook by
972    /// [`build_from_file`].
973    pub fn reload_from_file(&self) {
974        let path = config_file_path();
975        let toml = match std::fs::read_to_string(&path) {
976            Ok(toml) => toml,
977            Err(e) => {
978                eprintln!("[proxy_config] reload: cannot read {path}: {e} — keeping the running routes");
979                return;
980            }
981        };
982        match self.reload_from_str(&toml) {
983            Ok(ReloadOutcome::Unchanged) => {}
984            Ok(ReloadOutcome::Reloaded { routes, warnings }) => {
985                for w in &warnings {
986                    eprintln!("[proxy_config] Warning: {w}");
987                }
988                println!("[proxy_config] reloaded {path}: {routes} route{}", if routes == 1 { "" } else { "s" });
989            }
990            Err(report) => {
991                for e in &report.errors {
992                    eprintln!("[proxy_config] reload ERROR: {e}");
993                }
994                eprintln!("[proxy_config] {path} has errors — keeping the running routes");
995            }
996        }
997    }
998}
999
1000/// The request with its target normalized ([`normalize_request_target`]),
1001/// borrowed when it already is.
1002fn normalized(request: &Request) -> std::borrow::Cow<'_, Request> {
1003    match normalize_request_target(&request.request_uri) {
1004        std::borrow::Cow::Borrowed(_) => std::borrow::Cow::Borrowed(request),
1005        std::borrow::Cow::Owned(target) => {
1006            let mut owned = request.clone();
1007            owned.request_uri = target;
1008            std::borrow::Cow::Owned(owned)
1009        }
1010    }
1011}
1012
1013/// A request target's path normalized before routes match it, and sent
1014/// upstream that way (#50): unreserved characters percent-decoded (`%61` →
1015/// `a`; other escapes kept, hex upper-cased), repeated slashes collapsed,
1016/// `.` and `..` segments resolved (never above `/`). Otherwise `/%61dmin`,
1017/// `//admin` or `/x/../admin` missed a protected `/admin` route and reached
1018/// the backend through a catch-all route. The query is kept as is; a
1019/// target that isn't a path (`*`) is returned unchanged.
1020pub(crate) fn normalize_request_target(target: &str) -> std::borrow::Cow<'_, str> {
1021    let (path, query) = match target.find('?') {
1022        Some(i) => (&target[..i], &target[i..]),
1023        None => (target, ""),
1024    };
1025    if !path.starts_with('/') || !(path.contains('%') || path.contains("//") || path.contains("/.")) {
1026        return std::borrow::Cow::Borrowed(target);
1027    }
1028    let bytes = path.as_bytes();
1029    let mut decoded = String::with_capacity(path.len());
1030    let mut i = 0;
1031    while i < bytes.len() {
1032        let hex = |b: u8| (b as char).to_digit(16);
1033        if bytes[i] == b'%' && i + 2 < bytes.len() {
1034            if let (Some(h), Some(l)) = (hex(bytes[i + 1]), hex(bytes[i + 2])) {
1035                let value = (h * 16 + l) as u8;
1036                if value.is_ascii_alphanumeric() || matches!(value, b'-' | b'.' | b'_' | b'~') {
1037                    decoded.push(value as char);
1038                } else {
1039                    decoded.push('%');
1040                    decoded.push(bytes[i + 1].to_ascii_uppercase() as char);
1041                    decoded.push(bytes[i + 2].to_ascii_uppercase() as char);
1042                }
1043                i += 3;
1044                continue;
1045            }
1046        }
1047        // A `%` that starts no escape is a literal percent sign: encoded, so
1048        // what follows can't form a new escape with it (normalizing twice
1049        // must change nothing — the fuzzer found `%%611` → `%a1`).
1050        if bytes[i] == b'%' {
1051            decoded.push_str("%25");
1052            i += 1;
1053            continue;
1054        }
1055        // `path` is a &str: copy whole UTF-8 sequences.
1056        let ch = path[i..].chars().next().unwrap_or('\u{FFFD}');
1057        decoded.push(ch);
1058        i += ch.len_utf8().max(1);
1059    }
1060    let mut segments: Vec<&str> = Vec::new();
1061    let parts: Vec<&str> = decoded.split('/').collect();
1062    let last = parts.len().saturating_sub(1);
1063    let mut trailing_slash = false;
1064    for (n, segment) in parts.iter().enumerate() {
1065        match *segment {
1066            "" => trailing_slash = n == last,
1067            "." => trailing_slash = n == last,
1068            ".." => {
1069                segments.pop();
1070                trailing_slash = n == last;
1071            }
1072            s => {
1073                segments.push(s);
1074                trailing_slash = false;
1075            }
1076        }
1077    }
1078    let mut out = String::with_capacity(decoded.len() + query.len());
1079    out.push('/');
1080    out.push_str(&segments.join("/"));
1081    if trailing_slash && !segments.is_empty() {
1082        out.push('/');
1083    }
1084    out.push_str(query);
1085    if out == target {
1086        std::borrow::Cow::Borrowed(target)
1087    } else {
1088        std::borrow::Cow::Owned(out)
1089    }
1090}
1091
1092/// A route's `host` against the request's (SNI or `Host`): ignoring ASCII
1093/// case, and ignoring a port the route doesn't name (#50).
1094fn host_matches(expected: &str, actual: &str) -> bool {
1095    let actual = if expected.contains(':') {
1096        actual
1097    } else if let Some(rest) = actual.strip_prefix('[') {
1098        // [IPv6]:port
1099        rest.split(']').next().unwrap_or(rest)
1100    } else {
1101        actual.rsplit_once(':').map_or(actual, |(host, port)| if port.bytes().all(|b| b.is_ascii_digit()) { host } else { actual })
1102    };
1103    let expected = expected.trim_start_matches('[').trim_end_matches(']');
1104    actual.eq_ignore_ascii_case(expected)
1105}
1106
1107impl Application for ConfigDrivenApp {
1108    fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1109        let request = normalized(request);
1110        let generation = self.current();
1111        match &generation.entry {
1112            Some(entry) => entry.execute(&request, conn),
1113            None => generation.dispatch.execute(&request, conn),
1114        }
1115    }
1116
1117    /// The matched route decides: a `grpc` route streams request bodies
1118    /// (client-streaming and bidirectional calls), the others don't yet.
1119    fn streams_body(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1120        let request = normalized(request);
1121        let generation = self.current();
1122        match &generation.entry {
1123            Some(entry) => entry.streams_body(&request, conn),
1124            None => generation.dispatch.streams_body(&request, conn),
1125        }
1126    }
1127
1128    /// Routes proxy upstream or run configured middleware: only requests
1129    /// that match no route and fall through to the static files run inline,
1130    /// and only without a global `[middleware]` table.
1131    fn runs_inline(&self, request: &Request, conn: &ConnectionInfo) -> bool {
1132        let request = normalized(request);
1133        let generation = self.current();
1134        generation.entry.is_none()
1135            && !generation.dispatch.routes.iter().any(|r| r.matcher.matches(&request, conn))
1136            && App::is_inline_request(&request)
1137    }
1138
1139    /// A request for a proxy route with a plain-HTTP upstream and no
1140    /// middleware (route or global) waits for the backend without holding
1141    /// a thread.
1142    #[cfg(feature = "http2")]
1143    fn execute_async<'a>(&'a self, request: &'a Request, conn: &'a ConnectionInfo) -> Option<crate::application::AsyncResponse<'a>> {
1144        let generation = self.current();
1145        if generation.entry.is_some() {
1146            return None;
1147        }
1148        let request = normalized(request);
1149        let route = generation.dispatch.routes.iter().find(|r| r.matcher.matches(&request, conn))?;
1150        let proxy = Arc::clone(route.async_proxy.as_ref()?);
1151        Some(Box::pin(async move { proxy.execute_async(&request, conn).await }))
1152    }
1153}
1154
1155// ── H2Route ───────────────────────────────────────────────────────────────────
1156
1157/// The `grpc` action: requests forwarded over HTTP/2 to the upstream's live
1158/// backends in turn, bodies streamed both ways, trailers kept; gRPC-web
1159/// requests translated to gRPC on the way ([`GrpcWebLayer`](crate::proxy::GrpcWebLayer)).
1160#[cfg(feature = "http2")]
1161pub(crate) struct H2Route {
1162    live: Arc<RwLock<Vec<String>>>,
1163    tls: bool,
1164    trust: Option<Arc<crate::proxy::UpstreamTls>>,
1165    connect_timeout: std::time::Duration,
1166    read_timeout: std::time::Duration,
1167    counter: std::sync::atomic::AtomicUsize,
1168}
1169
1170#[cfg(feature = "http2")]
1171impl H2Route {
1172    pub(crate) fn new(
1173        live: Arc<RwLock<Vec<String>>>,
1174        tls: bool,
1175        trust: Option<Arc<crate::proxy::UpstreamTls>>,
1176        connect_timeout_ms: u64,
1177        read_timeout_ms: u64,
1178    ) -> H2Route {
1179        H2Route {
1180            live,
1181            tls,
1182            trust,
1183            connect_timeout: std::time::Duration::from_millis(connect_timeout_ms),
1184            read_timeout: std::time::Duration::from_millis(read_timeout_ms),
1185            counter: std::sync::atomic::AtomicUsize::new(0),
1186        }
1187    }
1188}
1189
1190#[cfg(feature = "http2")]
1191impl Application for H2Route {
1192    fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1193        use crate::middleware::Middleware;
1194        crate::proxy::GrpcWebLayer::new().handle(request, conn, &H2Forward(self))
1195    }
1196
1197    fn streams_body(&self, _request: &Request, _conn: &ConnectionInfo) -> bool {
1198        true
1199    }
1200}
1201
1202/// [`H2Route`] after the gRPC-web translation.
1203#[cfg(feature = "http2")]
1204struct H2Forward<'a>(&'a H2Route);
1205
1206#[cfg(feature = "http2")]
1207impl Application for H2Forward<'_> {
1208    fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1209        let route = self.0;
1210        let backends = route.live.read().unwrap_or_else(|e| e.into_inner()).clone();
1211        if backends.is_empty() {
1212            return Ok(crate::proxy::bad_gateway());
1213        }
1214        // A streamed request body can be sent once: no second backend then.
1215        let attempts = if matches!(request.body, crate::request::Body::Stream(_)) { 1 } else { backends.len() };
1216        let start = route.counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1217        for attempt in 0..attempts {
1218            let url = &backends[(start + attempt) % backends.len()];
1219            let result = crate::proxy::forward_h2(
1220                request, &conn.client.ip, url, route.tls, route.trust.as_deref(), route.connect_timeout, route.read_timeout,
1221            );
1222            match result {
1223                Ok(response) => return Ok(response),
1224                Err(e) => eprintln!("grpc route: {e}"),
1225            }
1226        }
1227        Ok(crate::proxy::bad_gateway())
1228    }
1229}
1230
1231// ── NullApp ────────────────────────────────────────────────────────────────────
1232
1233/// A dead-end `Application` that always returns 404.
1234/// Used as the `next` parameter when calling `Middleware::handle` directly.
1235#[derive(Clone, Copy)]
1236pub(crate) struct NullApp;
1237
1238impl Application for NullApp {
1239    fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1240        let mut r = Response::new();
1241        r.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
1242        r.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
1243        Ok(r)
1244    }
1245}
1246
1247// ── DynamicProxy ──────────────────────────────────────────────────────────────
1248
1249use std::collections::HashMap;
1250use std::collections::hash_map::DefaultHasher;
1251use std::hash::{Hash, Hasher};
1252use std::sync::atomic::AtomicUsize;
1253use std::time::{Duration, SystemTime, UNIX_EPOCH};
1254
1255/// Backend-selection strategy for `DynamicProxy`, configured via the
1256/// `strategy` field on `[[upstream]]` in `rws.config.toml`. Unknown or empty
1257/// values fall back to `RoundRobin`, matching the parser's own default.
1258#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1259pub(crate) enum LoadBalanceStrategy {
1260    RoundRobin,
1261    Random,
1262    IpHash,
1263    LeastConnections,
1264}
1265
1266impl LoadBalanceStrategy {
1267    fn parse(s: &str) -> Self {
1268        match s {
1269            "random" => LoadBalanceStrategy::Random,
1270            "ip_hash" => LoadBalanceStrategy::IpHash,
1271            "least_connections" => LoadBalanceStrategy::LeastConnections,
1272            _ => LoadBalanceStrategy::RoundRobin,
1273        }
1274    }
1275}
1276
1277/// A proxy adapter that reads its backend list from a shared, health-checker-
1278/// maintained live list at request time. Supports dynamic removal/restoration
1279/// of backends without restarting, and — for a declared `[[upstream]]` — its
1280/// [`upstream::UpstreamPolicy`]: weights, stickiness, the circuit breaker,
1281/// retries and metrics.
1282///
1283/// This type is `Clone + Send + Sync` and implements `Application`.
1284#[derive(Clone)]
1285pub(crate) struct DynamicProxy {
1286    live: Arc<RwLock<Vec<String>>>,
1287    counter: Arc<AtomicUsize>,
1288    connect_timeout: Duration,
1289    read_timeout: Duration,
1290    strip_prefix: Option<Arc<String>>,
1291    add_prefix: Option<Arc<String>>,
1292    tls: bool,
1293    strategy: LoadBalanceStrategy,
1294    connections: Arc<RwLock<HashMap<String, Arc<AtomicUsize>>>>,
1295    /// The upstream's `tls_ca_file` trust settings; `None` = WebPKI roots.
1296    trust: Option<Arc<crate::proxy::UpstreamTls>>,
1297    /// The upstream's name, for metrics.
1298    upstream: Arc<String>,
1299    /// The declared upstream's policy; `None` for an inline backend.
1300    policy: Option<Arc<upstream::UpstreamPolicy>>,
1301}
1302
1303impl DynamicProxy {
1304    /// Trust the given CA for TLS backends instead of the WebPKI roots.
1305    pub(crate) fn with_trust(mut self, trust: Option<Arc<crate::proxy::UpstreamTls>>) -> Self {
1306        self.trust = trust;
1307        self
1308    }
1309
1310    /// Apply the upstream's weights, stickiness, circuit breaker, retries
1311    /// and metrics.
1312    pub(crate) fn with_policy(mut self, upstream: &str, policy: Option<Arc<upstream::UpstreamPolicy>>) -> Self {
1313        self.upstream = Arc::new(upstream.to_string());
1314        self.policy = policy;
1315        self
1316    }
1317
1318    pub(crate) fn new(
1319        live: Arc<RwLock<Vec<String>>>,
1320        connect_timeout_ms: u64,
1321        read_timeout_ms: u64,
1322        strip_prefix: Option<String>,
1323        add_prefix: Option<String>,
1324        tls: bool,
1325        strategy: String,
1326    ) -> Self {
1327        DynamicProxy {
1328            live,
1329            counter: Arc::new(AtomicUsize::new(0)),
1330            connect_timeout: Duration::from_millis(connect_timeout_ms),
1331            read_timeout: Duration::from_millis(read_timeout_ms),
1332            strip_prefix: strip_prefix.map(Arc::new),
1333            add_prefix: add_prefix.map(Arc::new),
1334            tls,
1335            strategy: LoadBalanceStrategy::parse(&strategy),
1336            connections: Arc::new(RwLock::new(HashMap::new())),
1337            trust: None,
1338            upstream: Arc::new(String::new()),
1339            policy: None,
1340        }
1341    }
1342
1343    /// Pick by the strategy among `candidates` (not empty).
1344    fn pick(&self, candidates: &[String], client_ip: &str) -> String {
1345        let policy = self.policy.as_deref().filter(|p| p.weighted());
1346        let idx = match self.strategy {
1347            LoadBalanceStrategy::RoundRobin => match policy {
1348                Some(policy) => return policy.weighted_round_robin(candidates).unwrap_or_else(|| candidates[0].clone()),
1349                None => self.counter.fetch_add(1, Ordering::Relaxed) % candidates.len(),
1350            },
1351            LoadBalanceStrategy::Random => {
1352                let nanos = SystemTime::now()
1353                    .duration_since(UNIX_EPOCH)
1354                    .map(|d| d.subsec_nanos())
1355                    .unwrap_or(0) as usize;
1356                let salt = self.counter.fetch_add(1, Ordering::Relaxed);
1357                let roll = nanos.wrapping_add(salt.wrapping_mul(2_654_435_761));
1358                match policy {
1359                    Some(policy) => policy.weighted_index(candidates, roll),
1360                    None => roll % candidates.len(),
1361                }
1362            }
1363            LoadBalanceStrategy::IpHash => {
1364                let mut hasher = DefaultHasher::new();
1365                client_ip.hash(&mut hasher);
1366                (hasher.finish() as usize) % candidates.len()
1367            }
1368            LoadBalanceStrategy::LeastConnections => {
1369                let connections = self.connections.read().unwrap();
1370                let load = |backend: &String| connections.get(backend).map(|c| c.load(Ordering::Relaxed)).unwrap_or(0) as u64;
1371                let weight = |backend: &String| self.policy.as_deref().map_or(1, |p| p.weight(backend)) as u64;
1372                // Fewest in flight per unit of weight (compared crosswise).
1373                candidates
1374                    .iter()
1375                    .enumerate()
1376                    .min_by(|(_, a), (_, b)| (load(a) * weight(b)).cmp(&(load(b) * weight(a))))
1377                    .map(|(i, _)| i)
1378                    .unwrap_or(0)
1379            }
1380        };
1381        candidates[idx].clone()
1382    }
1383
1384    /// The backend for this attempt: the sticky one if it's still in
1385    /// rotation, else one by the strategy, skipping `tried` and backends
1386    /// whose circuit breaker is open. `Err(true)`: none, because the breaker
1387    /// holds the rest open; `Err(false)`: none at all.
1388    fn next_backend(&self, request: &Request, client_ip: &str, tried: &[String]) -> Result<String, bool> {
1389        let mut candidates: Vec<String> = self.live.read().unwrap().iter().filter(|b| !tried.contains(b)).cloned().collect();
1390        let Some(policy) = self.policy.as_deref() else {
1391            return if candidates.is_empty() { Err(false) } else { Ok(self.pick(&candidates, client_ip)) };
1392        };
1393        if let Some(sticky) = &policy.sticky {
1394            if let Some(id) = upstream::cookie_value(request, &sticky.cookie) {
1395                if let Some(backend) = candidates.iter().find(|b| upstream::backend_id(b) == id).cloned() {
1396                    if policy.admits(&backend) {
1397                        return Ok(backend);
1398                    }
1399                    candidates.retain(|b| *b != backend);
1400                }
1401            }
1402        }
1403        let ejected = policy.any_open(&candidates);
1404        while !candidates.is_empty() {
1405            let backend = self.pick(&candidates, client_ip);
1406            if policy.admits(&backend) {
1407                return Ok(backend);
1408            }
1409            candidates.retain(|b| *b != backend);
1410        }
1411        Err(ejected)
1412    }
1413
1414    /// Returns the shared in-flight connection counter for `backend`,
1415    /// creating it on first use. Only consulted under `LeastConnections`.
1416    fn connection_counter(&self, backend: &str) -> Arc<AtomicUsize> {
1417        if let Some(counter) = self.connections.read().unwrap().get(backend) {
1418            return Arc::clone(counter);
1419        }
1420        let mut connections = self.connections.write().unwrap();
1421        Arc::clone(
1422            connections
1423                .entry(backend.to_string())
1424                .or_insert_with(|| Arc::new(AtomicUsize::new(0))),
1425        )
1426    }
1427}
1428
1429/// Decrements a backend's in-flight connection count when the request
1430/// finishes (including early returns), keeping `least_connections` accurate.
1431struct ConnectionGuard {
1432    counter: Arc<AtomicUsize>,
1433}
1434
1435impl Drop for ConnectionGuard {
1436    fn drop(&mut self) {
1437        self.counter.fetch_sub(1, Ordering::Relaxed);
1438    }
1439}
1440
1441/// A request ready to send: the chosen backend and the (possibly rewritten)
1442/// request. Holding it keeps the backend's least-connections count raised.
1443struct Prepared<'a> {
1444    backend: String,
1445    host: String,
1446    port: u16,
1447    request: std::borrow::Cow<'a, Request>,
1448    _guard: Option<ConnectionGuard>,
1449}
1450
1451/// How one attempt went.
1452enum Attempt {
1453    /// A response to pass on.
1454    Done(Response),
1455    /// A failure, retried when the policy allows: the backend's response (a
1456    /// retried status) or none (no response).
1457    Failed(Option<Response>),
1458}
1459
1460impl DynamicProxy {
1461    /// Pick a backend (not one of `tried`) and apply the path rewriting;
1462    /// `Err` is the response to send instead (no backend in rotation — `503`
1463    /// when the circuit breaker took them out, `502` otherwise — or an
1464    /// unparsable backend address).
1465    fn prepare<'a>(&self, request: &'a Request, conn: &ConnectionInfo, tried: &[String]) -> Result<Prepared<'a>, Response> {
1466        let backend = self.next_backend(request, &conn.client.ip, tried).map_err(|ejected| if ejected { service_unavailable() } else { bad_gateway() })?;
1467
1468        let guard = if self.strategy == LoadBalanceStrategy::LeastConnections {
1469            let counter = self.connection_counter(&backend);
1470            counter.fetch_add(1, Ordering::Relaxed);
1471            Some(ConnectionGuard { counter })
1472        } else {
1473            None
1474        };
1475
1476        let (host, port, _) = crate::proxy_config::health::parse_backend_url(&backend).ok_or_else(bad_gateway)?;
1477
1478        // Apply path rewriting if configured
1479        let request = if self.strip_prefix.is_some() || self.add_prefix.is_some() {
1480            let mut req_clone = request.clone();
1481            if let Some(ref sp) = self.strip_prefix {
1482                if let Some(stripped) = req_clone.request_uri.strip_prefix(sp.as_str()) {
1483                    req_clone.request_uri = if stripped.is_empty() || !stripped.starts_with('/') {
1484                        format!("/{}", stripped)
1485                    } else {
1486                        stripped.to_string()
1487                    };
1488                }
1489            }
1490            if let Some(ref ap) = self.add_prefix {
1491                req_clone.request_uri = format!("{}{}", ap, req_clone.request_uri);
1492            }
1493            std::borrow::Cow::Owned(req_clone)
1494        } else {
1495            std::borrow::Cow::Borrowed(request)
1496        };
1497
1498        Ok(Prepared { backend, host, port, request, _guard: guard })
1499    }
1500
1501    /// Record an attempt's outcome (breaker, metrics) and classify it.
1502    fn settle(&self, backend: &str, result: Result<Response, ()>, started: std::time::Instant) -> Attempt {
1503        let status = result.as_ref().ok().map(|r| r.status_code);
1504        upstream::observe(&self.upstream, backend, status, started);
1505        let failed = status.is_none_or(|s| matches!(s, 502..=504));
1506        if let Some(policy) = &self.policy {
1507            policy.record(backend, failed);
1508        }
1509        match result {
1510            Ok(response) if self.policy.as_deref().is_some_and(|p| p.retries_status(response.status_code)) => Attempt::Failed(Some(response)),
1511            Ok(response) => Attempt::Done(response),
1512            Err(()) => Attempt::Failed(None),
1513        }
1514    }
1515
1516    /// How many attempts `request` may get.
1517    fn attempts(&self, request: &Request) -> u32 {
1518        match self.policy.as_deref().and_then(|p| p.retry.as_ref()) {
1519            Some(retry) if upstream::replayable(request) => 1 + retry.attempts,
1520            _ => 1,
1521        }
1522    }
1523
1524    /// Set the sticky cookie when the client doesn't hold this backend's.
1525    fn stick(&self, request: &Request, backend: &str, response: &mut Response) {
1526        let Some(sticky) = self.policy.as_deref().and_then(|p| p.sticky.as_ref()) else { return };
1527        let id = upstream::backend_id(backend);
1528        if upstream::cookie_value(request, &sticky.cookie).as_deref() == Some(id.as_str()) {
1529            return;
1530        }
1531        let max_age = sticky.max_age_secs.map(|s| format!("; Max-Age={s}")).unwrap_or_default();
1532        response.headers.push(crate::header::Header {
1533            name: "Set-Cookie".into(),
1534            value: format!("{}={id}; Path=/; HttpOnly; SameSite=Lax{max_age}", sticky.cookie).into(),
1535        });
1536    }
1537
1538    /// The same as [`Application::execute`], without blocking a thread while
1539    /// the backend answers, for HTTP and HTTPS backends alike.
1540    #[cfg(feature = "http2")]
1541    pub(crate) async fn execute_async(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1542        let mut tried: Vec<String> = Vec::new();
1543        let mut last: Option<Response> = None;
1544        for attempt in 0..self.attempts(request) {
1545            let prepared = match self.prepare(request, conn, &tried) {
1546                Ok(p) => p,
1547                Err(response) => return Ok(last.unwrap_or(response)),
1548            };
1549            if attempt > 0 {
1550                upstream::observe_retry(&self.upstream);
1551            }
1552            let tls = self.tls.then(|| match &self.trust {
1553                Some(trust) => trust.async_tls(),
1554                None => crate::proxy::async_upstream::AsyncTls {
1555                    config: crate::proxy::tls_client_config(),
1556                    pool: crate::proxy::async_upstream::default_tls_pool(),
1557                },
1558            });
1559            let started = std::time::Instant::now();
1560            let result = crate::proxy::async_upstream::forward(
1561                &prepared.request,
1562                &conn.client.ip,
1563                &prepared.host,
1564                prepared.port,
1565                tls,
1566                self.connect_timeout,
1567                self.read_timeout,
1568            )
1569            .await
1570            .map_err(|_| ());
1571            match self.settle(&prepared.backend, result, started) {
1572                Attempt::Done(mut response) => {
1573                    self.stick(request, &prepared.backend, &mut response);
1574                    return Ok(response);
1575                }
1576                Attempt::Failed(response) => last = response,
1577            }
1578            tried.push(prepared.backend.clone());
1579        }
1580        Ok(last.unwrap_or_else(bad_gateway))
1581    }
1582
1583    /// Whether [`execute_async`](Self::execute_async) can serve this proxy.
1584    pub(crate) fn supports_async(&self) -> bool {
1585        cfg!(feature = "http2")
1586    }
1587}
1588
1589impl Application for DynamicProxy {
1590    fn execute(&self, request: &Request, conn: &ConnectionInfo) -> Result<Response, String> {
1591        let mut tried: Vec<String> = Vec::new();
1592        let mut last: Option<Response> = None;
1593        for attempt in 0..self.attempts(request) {
1594            let prepared = match self.prepare(request, conn, &tried) {
1595                Ok(p) => p,
1596                Err(response) => return Ok(last.unwrap_or(response)),
1597            };
1598            if attempt > 0 {
1599                upstream::observe_retry(&self.upstream);
1600            }
1601            // Pooled keep-alive connections to the backend, TLS or not (a
1602            // fresh connection per request halved throughput vs. Caddy/nginx).
1603            let started = std::time::Instant::now();
1604            let result = crate::proxy::forward_upstream_trusting(
1605                crate::proxy::shared_pool(),
1606                &prepared.request,
1607                &conn.client.ip,
1608                &prepared.host,
1609                prepared.port,
1610                self.tls,
1611                self.trust.as_deref(),
1612                self.connect_timeout,
1613                self.read_timeout,
1614            )
1615            .map_err(|_| ());
1616            match self.settle(&prepared.backend, result, started) {
1617                Attempt::Done(mut response) => {
1618                    self.stick(request, &prepared.backend, &mut response);
1619                    return Ok(response);
1620                }
1621                Attempt::Failed(response) => last = response,
1622            }
1623            tried.push(prepared.backend.clone());
1624        }
1625        Ok(last.unwrap_or_else(bad_gateway))
1626    }
1627}
1628
1629fn service_unavailable() -> Response {
1630    use crate::mime_type::MimeType;
1631    use crate::range::Range;
1632    let mut r = Response::new();
1633    r.status_code = *STATUS_CODE_REASON_PHRASE.n503_service_unavailable.status_code;
1634    r.reason_phrase = STATUS_CODE_REASON_PHRASE.n503_service_unavailable.reason_phrase.to_string();
1635    r.headers.push(crate::header::Header { name: "Retry-After".into(), value: "5".into() });
1636    r.content_range_list = vec![Range::get_content_range(b"503 Service Unavailable: every backend is out of rotation".to_vec(), MimeType::TEXT_PLAIN.to_string())];
1637    r
1638}
1639
1640fn bad_gateway() -> Response {
1641    use crate::mime_type::MimeType;
1642    use crate::range::Range;
1643    let cr = Range::get_content_range(
1644        b"502 Bad Gateway".to_vec(),
1645        MimeType::TEXT_PLAIN.to_string(),
1646    );
1647    let mut r = Response::new();
1648    r.status_code = *STATUS_CODE_REASON_PHRASE.n502_bad_gateway.status_code;
1649    r.reason_phrase = STATUS_CODE_REASON_PHRASE.n502_bad_gateway.reason_phrase.to_string();
1650    r.content_range_list = vec![cr];
1651    r
1652}
1653
1654// ── RedirectAdapter ────────────────────────────────────────────────────────────
1655
1656/// Action adapter that issues HTTP redirects.
1657///
1658/// `$path` in `location_template` is replaced with the request URI at runtime.
1659#[derive(Clone)]
1660pub(crate) struct RedirectAdapter {
1661    location_template: Arc<String>,
1662    status: i16,
1663    reason: Arc<String>,
1664}
1665
1666impl RedirectAdapter {
1667    pub(crate) fn new(location: String, status: u16) -> Self {
1668        let (code, reason) = redirect_status(status);
1669        RedirectAdapter {
1670            location_template: Arc::new(location),
1671            status: code,
1672            reason: Arc::new(reason),
1673        }
1674    }
1675}
1676
1677fn redirect_status(code: u16) -> (i16, String) {
1678    let phrase = match code {
1679        301 => STATUS_CODE_REASON_PHRASE.n301_moved_permanently.reason_phrase,
1680        302 => STATUS_CODE_REASON_PHRASE.n302_found.reason_phrase,
1681        307 => STATUS_CODE_REASON_PHRASE.n307_temporary_redirect.reason_phrase,
1682        308 => STATUS_CODE_REASON_PHRASE.n308_permanent_redirect.reason_phrase,
1683        _ => "Redirect",
1684    };
1685    (code as i16, phrase.to_string())
1686}
1687
1688impl Application for RedirectAdapter {
1689    fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1690        let location = self
1691            .location_template
1692            .replace("$path", &request.request_uri);
1693        use crate::header::Header;
1694        let mut r = Response::new();
1695        r.status_code = self.status;
1696        r.reason_phrase = self.reason.as_ref().clone();
1697        r.headers.push(Header { name: "Location".into(), value: location.into() });
1698        Ok(r)
1699    }
1700}
1701
1702// ── RespondAdapter ─────────────────────────────────────────────────────────────
1703
1704/// Action adapter that returns a fixed response body.
1705#[derive(Clone)]
1706pub(crate) struct RespondAdapter {
1707    status: i16,
1708    reason: Arc<String>,
1709    body: Arc<Vec<u8>>,
1710    content_type: Arc<String>,
1711}
1712
1713impl RespondAdapter {
1714    pub(crate) fn new(status: u16, body: String, content_type: String) -> Self {
1715        use crate::response::STATUS_CODE_REASON_PHRASE;
1716        let reason = match status {
1717            200 => STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string(),
1718            201 => STATUS_CODE_REASON_PHRASE.n201_created.reason_phrase.to_string(),
1719            204 => STATUS_CODE_REASON_PHRASE.n204_no_content.reason_phrase.to_string(),
1720            400 => STATUS_CODE_REASON_PHRASE.n400_bad_request.reason_phrase.to_string(),
1721            401 => STATUS_CODE_REASON_PHRASE.n401_unauthorized.reason_phrase.to_string(),
1722            403 => STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string(),
1723            404 => STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string(),
1724            500 => STATUS_CODE_REASON_PHRASE.n500_internal_server_error.reason_phrase.to_string(),
1725            _ => "OK".to_string(),
1726        };
1727        RespondAdapter {
1728            status: status as i16,
1729            reason: Arc::new(reason),
1730            body: Arc::new(body.into_bytes()),
1731            content_type: Arc::new(content_type),
1732        }
1733    }
1734}
1735
1736impl Application for RespondAdapter {
1737    fn execute(&self, _request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1738        use crate::range::Range;
1739        let mut r = Response::new();
1740        r.status_code = self.status;
1741        r.reason_phrase = self.reason.as_ref().clone();
1742        if !self.body.is_empty() {
1743            r.content_range_list = vec![Range::get_content_range(
1744                self.body.as_ref().clone(),
1745                self.content_type.as_ref().clone(),
1746            )];
1747        }
1748        Ok(r)
1749    }
1750}
1751
1752// ── StaticAdapter ──────────────────────────────────────────────────────────────
1753
1754/// Action adapter that serves static files from a configured `root` directory.
1755///
1756/// Unlike `StaticResourceController` (which always resolves paths relative to
1757/// the process's current working directory), this adapter is parameterized
1758/// per-route by `ActionConfig::Static { root, index }` from `rws.config.toml`,
1759/// so a config-driven proxy can serve an arbitrary directory without Rust code.
1760#[derive(Clone)]
1761pub(crate) struct StaticAdapter {
1762    root: Arc<std::path::PathBuf>,
1763    index: Arc<Vec<String>>,
1764}
1765
1766impl StaticAdapter {
1767    pub(crate) fn new(root: String, index: Vec<String>) -> Self {
1768        let index = if index.is_empty() { vec!["index.html".to_string()] } else { index };
1769        StaticAdapter {
1770            root: Arc::new(std::path::PathBuf::from(root)),
1771            index: Arc::new(index),
1772        }
1773    }
1774
1775    /// Resolves `request_uri` against `root`. Returns `None` if the decoded
1776    /// path contains a `..` segment, which would otherwise let a request
1777    /// escape the configured root directory.
1778    fn resolve(&self, request_uri: &str) -> Option<std::path::PathBuf> {
1779        let raw_path = request_uri.split('?').next().unwrap_or(request_uri);
1780        let decoded = crate::url::URL::percent_decode(raw_path);
1781
1782        if decoded.split('/').any(|segment| segment == "..") {
1783            return None;
1784        }
1785
1786        let relative = decoded.trim_start_matches('/');
1787        Some(self.root.join(relative))
1788    }
1789}
1790
1791impl Application for StaticAdapter {
1792    fn execute(&self, request: &Request, _conn: &ConnectionInfo) -> Result<Response, String> {
1793        let mut response = Response::new();
1794
1795        let not_found = |mut response: Response| {
1796            response.status_code = *STATUS_CODE_REASON_PHRASE.n404_not_found.status_code;
1797            response.reason_phrase = STATUS_CODE_REASON_PHRASE.n404_not_found.reason_phrase.to_string();
1798            response
1799        };
1800
1801        let candidate = match self.resolve(&request.request_uri) {
1802            Some(p) => p,
1803            None => {
1804                response.status_code = *STATUS_CODE_REASON_PHRASE.n403_forbidden.status_code;
1805                response.reason_phrase = STATUS_CODE_REASON_PHRASE.n403_forbidden.reason_phrase.to_string();
1806                return Ok(response);
1807            }
1808        };
1809
1810        let mut file_path = candidate;
1811        if file_path.is_dir() {
1812            let indexed = self
1813                .index
1814                .iter()
1815                .map(|name| file_path.join(name))
1816                .find(|p| p.is_file());
1817
1818            file_path = match indexed {
1819                Some(p) => p,
1820                None => return Ok(not_found(response)),
1821            };
1822        }
1823
1824        if !file_path.is_file() {
1825            return Ok(not_found(response));
1826        }
1827
1828        // Defense-in-depth against symlinks inside `root` that point outside it —
1829        // the `..`-segment check above only catches traversal in the request URI.
1830        if let (Ok(root_canon), Ok(file_canon)) =
1831            (self.root.canonicalize(), file_path.canonicalize())
1832        {
1833            if !file_canon.starts_with(&root_canon) {
1834                return Ok(not_found(response));
1835            }
1836        }
1837
1838        let path_str = match file_path.to_str() {
1839            Some(s) => s,
1840            None => return Ok(not_found(response)),
1841        };
1842
1843        match crate::range::Range::get_content_range_of_a_file(path_str) {
1844            Ok(content_range) => {
1845                response.status_code = *STATUS_CODE_REASON_PHRASE.n200_ok.status_code;
1846                response.reason_phrase = STATUS_CODE_REASON_PHRASE.n200_ok.reason_phrase.to_string();
1847                response.content_range_list = vec![content_range];
1848                Ok(response)
1849            }
1850            Err(_) => Ok(not_found(response)),
1851        }
1852    }
1853}
1854
1855// ── PerRouteRateLimit middleware ───────────────────────────────────────────────
1856
1857/// A per-route rate limiter middleware backed by a shared `RateLimiter`.
1858pub(crate) struct PerRouteRateLimit(pub(crate) Arc<crate::rate_limit::RateLimiter>);
1859
1860impl crate::middleware::Middleware for PerRouteRateLimit {
1861    fn handle(
1862        &self,
1863        request: &Request,
1864        conn: &ConnectionInfo,
1865        next: &dyn Application,
1866    ) -> Result<Response, String> {
1867        use crate::error::{AppError, IntoResponse};
1868        if self.0.check(&conn.client.ip) {
1869            next.execute(request, conn)
1870        } else {
1871            let mut response = AppError::TooManyRequests.into_response();
1872            let retry_after = self.0.retry_after(&conn.client.ip);
1873            response.headers.push(Header {
1874                name: Header::_RETRY_AFTER.into(),
1875                value: retry_after.to_string().into(),
1876            });
1877            Ok(response)
1878        }
1879    }
1880}
1881
1882// ── PerRouteMaxBodySize ───────────────────────────────────────────────────────
1883
1884/// A route-specific request body size limit, from `[route.middleware]
1885/// max_body_size`.
1886///
1887/// Rejects with `413 Payload Too Large` if `request.body` is bigger than the
1888/// configured ceiling. This runs *after* the request has already been fully
1889/// read off the socket — route middleware only ever sees an already-built
1890/// `Request` — so, unlike the global `RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES`
1891/// (checked against the declared `Content-Length` before any of an
1892/// oversized body is buffered), this can't prevent the memory from being
1893/// spent for a single oversized request. What it protects is this route's
1894/// own downstream logic/upstream backend from a body that's within the
1895/// global limit but still too big for this specific route.
1896pub(crate) struct PerRouteMaxBodySize(pub(crate) u64);
1897
1898impl crate::middleware::Middleware for PerRouteMaxBodySize {
1899    fn handle(
1900        &self,
1901        request: &Request,
1902        conn: &ConnectionInfo,
1903        next: &dyn Application,
1904    ) -> Result<Response, String> {
1905        use crate::error::{AppError, IntoResponse};
1906        if request.body.len() as u64 > self.0 {
1907            Ok(AppError::PayloadTooLarge(format!(
1908                "request body of {} bytes exceeds this route's {}-byte limit",
1909                request.body.len(),
1910                self.0,
1911            ))
1912            .into_response())
1913        } else {
1914            next.execute(request, conn)
1915        }
1916    }
1917}
1918
1919// ── BearerAuthMiddleware ───────────────────────────────────────────────────────
1920
1921/// Bearer token authentication middleware.
1922pub(crate) struct BearerAuthMiddleware {
1923    pub(crate) token: Arc<String>,
1924}
1925
1926impl crate::middleware::Middleware for BearerAuthMiddleware {
1927    fn handle(
1928        &self,
1929        request: &Request,
1930        conn: &ConnectionInfo,
1931        next: &dyn Application,
1932    ) -> Result<Response, String> {
1933        use crate::error::{AppError, IntoResponse};
1934        let expected = format!("Bearer {}", self.token);
1935        let authorized = request
1936            .headers
1937            .iter()
1938            .any(|h| h.name.eq_ignore_ascii_case("authorization") && constant_time_eq(h.value.as_bytes(), expected.as_bytes()));
1939        if authorized {
1940            next.execute(request, conn)
1941        } else {
1942            Ok(AppError::Unauthorized.into_response())
1943        }
1944    }
1945}
1946
1947/// Compares in time independent of where the inputs first differ, so response
1948/// timing doesn't reveal how much of a guessed token was right.
1949pub(crate) fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
1950    if a.len() != b.len() {
1951        return false;
1952    }
1953    a.iter().zip(b).fold(0u8, |diff, (x, y)| diff | (x ^ y)) == 0
1954}
1955
1956/// Stands in for an auth layer whose configuration can't be honoured (see
1957/// `builder::apply_middleware`): every request gets `503`, never the route.
1958pub(crate) struct AuthMisconfigured;
1959
1960impl crate::middleware::Middleware for AuthMisconfigured {
1961    fn handle(
1962        &self,
1963        _request: &Request,
1964        _conn: &ConnectionInfo,
1965        _next: &dyn Application,
1966    ) -> Result<Response, String> {
1967        Ok(Response::text(
1968            STATUS_CODE_REASON_PHRASE.n503_service_unavailable,
1969            "503 Service Unavailable: authentication for this route is misconfigured",
1970        ))
1971    }
1972}
1973
1974// ── arc_app helper ─────────────────────────────────────────────────────────────
1975
1976/// Box any `Application + Send + Sync + 'static` into an `Arc<dyn …>`.
1977pub(crate) fn arc_app<A: Application + Send + Sync + 'static>(
1978    a: A,
1979) -> Arc<dyn Application + Send + Sync> {
1980    Arc::new(a)
1981}
1982
1983// ── Public entry points ────────────────────────────────────────────────────────
1984
1985/// Build a `ConfigDrivenApp` from `rws.config.toml` and spawn L4/WS proxy
1986/// threads. Returns the app and a list of thread handles.
1987pub fn build_from_file() -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
1988    builder::build_from_file()
1989}