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