Skip to main content

rust_web_server/proxy_config/
builder.rs

1//! Builder: converts a `ProxyConfig` into a live `ConfigDrivenApp` with
2//! health checkers and L4/WS proxy threads.
3
4use std::collections::HashMap;
5use std::sync::atomic::AtomicBool;
6use std::sync::{Arc, RwLock};
7
8use crate::middleware::WithMiddleware;
9use crate::proxy_config::{
10    ActionConfig, AuthConfig, CompiledRoute, ConfigDrivenApp, DynamicProxy, MiddlewareConfig,
11    ProxyConfig, RedirectAdapter, RespondAdapter, RouteMatcher, StaticAdapter, arc_app,
12    AuthMisconfigured, BearerAuthMiddleware, PerRouteMaxBodySize, PerRouteRateLimit,
13};
14
15/// Build a `ConfigDrivenApp` from `rws.config.toml` and spawn L4/WS proxy
16/// threads. Returns the app and a list of background thread handles.
17/// Problems `ProxyConfig::validate` finds are printed first (the same
18/// findings `rws --check` reports). The returned app reloads its routes when
19/// the configuration is reloaded (`SIGHUP`, `POST /admin/config/reload`) —
20/// see [`ConfigDrivenApp::reload_from_file`].
21pub fn build_from_file() -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
22    let source = std::fs::read_to_string(crate::proxy_config::config_file_path()).unwrap_or_default();
23    let config = ProxyConfig::from_str(&source);
24    let mut report = crate::config_check::ConfigReport::default();
25    config.validate(&mut report);
26    for e in &report.errors {
27        eprintln!("[proxy_config] ERROR: {e}");
28    }
29    for w in &report.warnings {
30        eprintln!("[proxy_config] Warning: {w}");
31    }
32    let (app, handles) = build_with_source(config, Some(source));
33    let reloadable = app.clone();
34    crate::config_reload::on_reload(move || reloadable.reload_from_file());
35    (app, handles)
36}
37
38/// Build a `ConfigDrivenApp` from a `ProxyConfig` struct.
39pub fn build(config: ProxyConfig) -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
40    build_with_source(config, None)
41}
42
43fn build_with_source(config: ProxyConfig, source: Option<String>) -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
44    let health_stop = Arc::new(AtomicBool::new(false));
45    let compiled = compile_routes(&config, &health_stop);
46    let handles = spawn_listeners(&config);
47    (ConfigDrivenApp::new(compiled, config, health_stop, source), handles)
48}
49
50/// Compile `config`'s routes and start health checkers for its upstreams;
51/// the checkers stop when `health_stop` is set. Used at startup and on every
52/// hot reload.
53pub(crate) fn compile_routes(config: &ProxyConfig, health_stop: &Arc<AtomicBool>) -> Vec<CompiledRoute> {
54    // ── 0. The [authz] authorizer, for routes' authz middleware ─────────────
55    let authorizer = config_authorizer(config);
56    if let Some(Err(e)) = &authorizer {
57        eprintln!("[proxy_config] ERROR: [authz]: {e} — routes with authz answer 503 until it's fixed.");
58    }
59    crate::authz::config::set_current(authorizer);
60    // ── 1. Build upstream name → live backends map ─────────────────────────
61    let mut upstream_lives: HashMap<String, Arc<RwLock<Vec<String>>>> = HashMap::new();
62    // tls_ca_file trust settings per upstream, loaded once per compile (so a
63    // reload picks up a rotated CA). A CA file that can't be loaded fails
64    // closed: every handshake fails (502) instead of trusting public CAs.
65    let mut upstream_trust: HashMap<String, Arc<crate::proxy::UpstreamTls>> = HashMap::new();
66    // Weights, stickiness, circuit breaker, retries: one state per
67    // upstream, shared by its routes.
68    let mut upstream_policies: HashMap<String, Arc<crate::proxy_config::upstream::UpstreamPolicy>> = HashMap::new();
69
70    for upstream in &config.upstreams {
71        let live = Arc::new(RwLock::new(upstream.backends.clone()));
72        upstream_lives.insert(upstream.name.clone(), Arc::clone(&live));
73        upstream_policies.insert(
74            upstream.name.clone(),
75            Arc::new(crate::proxy_config::upstream::UpstreamPolicy::new(
76                &upstream.name,
77                Arc::clone(&live),
78                &upstream.backends,
79                &upstream.weights,
80                upstream.sticky.clone(),
81                upstream.circuit_breaker.as_ref(),
82                upstream.retry.clone(),
83            )),
84        );
85
86        let client = upstream.tls_client_cert_file.as_deref().zip(upstream.tls_client_key_file.as_deref());
87        let configured = upstream.tls_ca_file.is_some() || upstream.tls_client_cert_file.is_some() || upstream.tls_client_key_file.is_some();
88        let trust = configured.then(|| {
89            Arc::new(crate::proxy::UpstreamTls::new(upstream.tls_ca_file.as_deref(), client).unwrap_or_else(|e| {
90                eprintln!("[proxy_config] ERROR: upstream '{}': {e} — its TLS backends answer 502 until this is fixed", upstream.name);
91                crate::proxy::UpstreamTls::reject_all(&upstream.name)
92            }))
93        });
94        if let Some(trust) = &trust {
95            upstream_trust.insert(upstream.name.clone(), Arc::clone(trust));
96        }
97
98        if let Some(ref hc) = upstream.health_check {
99            crate::proxy_config::health::start_health_checker(
100                upstream.name.clone(),
101                upstream.backends.clone(),
102                Arc::clone(&live),
103                hc.clone(),
104                Arc::clone(health_stop),
105                trust,
106            );
107        }
108    }
109
110    crate::proxy_config::upstream::register(upstream_policies.values().cloned().collect());
111
112    // ── 2. Build compiled routes ───────────────────────────────────────────
113    let mut caches: Vec<(String, crate::cache::CacheLayer)> = Vec::new();
114    let mut compiled: Vec<CompiledRoute> = Vec::new();
115
116    for route in &config.routes {
117        let matcher = RouteMatcher::from_match_config(&route.match_);
118
119        let mut async_proxy: Option<Arc<DynamicProxy>> = None;
120        let cache_name = format!("route:{}", if route.name.is_empty() { format!("#{}", compiled.len() + 1) } else { route.name.clone() });
121        let mirror = route.middleware.mirror.as_ref().map(|m| build_mirror(&route.name, m, &upstream_lives));
122
123        // Build the action handler (base application)
124        let base_handler: Arc<dyn crate::application::Application + Send + Sync> =
125            match &route.action {
126                ActionConfig::Proxy {
127                    upstream,
128                    connect_timeout_ms,
129                    read_timeout_ms,
130                    strip_path_prefix,
131                    add_path_prefix,
132                } => {
133                    let live = upstream_lives
134                        .get(upstream.as_str())
135                        .cloned()
136                        .unwrap_or_else(|| {
137                            // Fallback: upstream not declared, use name as single backend
138                            Arc::new(RwLock::new(vec![upstream.clone()]))
139                        });
140                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
141                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
142                    let upstream_strategy = upstream_cfg
143                        .map(|u| u.strategy.clone())
144                        .unwrap_or_else(|| "round_robin".to_string());
145                    let proxy = Arc::new(DynamicProxy::new(
146                        live,
147                        *connect_timeout_ms,
148                        *read_timeout_ms,
149                        strip_path_prefix.clone(),
150                        add_path_prefix.clone(),
151                        upstream_tls,
152                        upstream_strategy,
153                    )
154                    .with_trust(upstream_trust.get(upstream.as_str()).cloned())
155                    .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned())
156                    .with_mirror(mirror.clone())
157                    .with_early_hints(&route.middleware.early_hints)
158                    .with_h3(upstream_cfg.is_some_and(|u| u.protocol.eq_ignore_ascii_case("h3"))));
159                    if proxy.supports_async() && route.middleware.is_empty() {
160                        async_proxy = Some(Arc::clone(&proxy));
161                    }
162                    proxy
163                }
164
165                ActionConfig::Redirect { location, status } => {
166                    arc_app(RedirectAdapter::new(location.clone(), *status))
167                }
168
169                ActionConfig::Respond { status, body, content_type } => {
170                    arc_app(RespondAdapter::new(*status, body.clone(), content_type.clone()))
171                }
172
173                ActionConfig::Static { root, index } => {
174                    arc_app(StaticAdapter::new(root.clone(), index.clone()))
175                }
176
177                // Grpc action: HTTP/2 to the upstream (HTTP/1.1 only in builds
178                // without the http2 feature, where gRPC can't work anyway).
179                ActionConfig::Grpc { upstream, connect_timeout_ms, read_timeout_ms } => {
180                    let live = upstream_lives
181                        .get(upstream.as_str())
182                        .cloned()
183                        .unwrap_or_else(|| Arc::new(RwLock::new(vec![upstream.clone()])));
184                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
185                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
186                    #[cfg(feature = "http2")]
187                    {
188                        let _ = upstream_cfg;
189                        arc_app(crate::proxy_config::H2Route::new(
190                            live,
191                            upstream_tls,
192                            upstream_trust.get(upstream.as_str()).cloned(),
193                            *connect_timeout_ms,
194                            *read_timeout_ms,
195                        ))
196                    }
197                    #[cfg(not(feature = "http2"))]
198                    {
199                        let upstream_strategy = upstream_cfg
200                            .map(|u| u.strategy.clone())
201                            .unwrap_or_else(|| "round_robin".to_string());
202                        arc_app(
203                            DynamicProxy::new(live, *connect_timeout_ms, *read_timeout_ms, None, None, upstream_tls, upstream_strategy)
204                                .with_trust(upstream_trust.get(upstream.as_str()).cloned())
205                                .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned()),
206                        )
207                    }
208                }
209
210                ActionConfig::Mcp(gateway) => arc_app(crate::proxy_config::mcp_gateway::GatewayAdapter::start(
211                    (**gateway).clone(),
212                    route.match_.path.clone().unwrap_or_else(|| "/mcp".to_string()),
213                )),
214                ActionConfig::Llm(llm) => arc_app(crate::proxy_config::llm::build(llm)),
215                ActionConfig::A2a(gateway) => arc_app(crate::proxy_config::a2a_gateway::A2aGateway::new(
216                    (**gateway).clone(),
217                    route.match_.path.as_deref().unwrap_or("/"),
218                )),
219                ActionConfig::Unknown(_) => {
220                    // No-op: these fall through to fallback
221                    arc_app(crate::proxy_config::NullApp)
222                }
223            };
224
225        // Wrap the base handler with middleware layers. A proxy route sends
226        // its mirror copies and early hints itself (so it keeps the async
227        // forwarder); any other action gets a MirrorLayer / EarlyHintsLayer.
228        let mut handler = apply_middleware_named(base_handler, &route.middleware, &cache_name, &mut caches);
229        if !matches!(route.action, ActionConfig::Proxy { .. }) {
230            if let Some(mirror) = &mirror {
231                handler = arc_app(WithMiddleware::new(ArcApp(handler)).wrap(crate::proxy::mirror::MirrorLayer::new(mirror.clone())));
232            }
233            if !route.middleware.early_hints.is_empty() {
234                let layer = crate::early_hints::EarlyHintsLayer::new(route.middleware.early_hints.iter().cloned());
235                handler = arc_app(WithMiddleware::new(ArcApp(handler)).wrap(layer));
236            }
237        }
238
239        compiled.push(CompiledRoute { matcher, handler, async_proxy });
240    }
241    crate::cache::registry::replace_group("route:", caches);
242    compiled
243}
244
245/// Start the `[[tcp_proxy]]` / `[[udp_proxy]]` / `[[ws_proxy]]` listeners.
246/// Startup only: a hot reload can't move bound sockets, so changes to these
247/// sections need a restart.
248fn spawn_listeners(config: &ProxyConfig) -> Vec<std::thread::JoinHandle<()>> {
249    let mut handles: Vec<std::thread::JoinHandle<()>> = Vec::new();
250
251    for tcp_cfg in &config.tcp_proxies {
252        let listen = tcp_cfg.listen.clone();
253        let backends = tcp_cfg.backends.clone();
254        let timeout_ms = tcp_cfg.connect_timeout_ms;
255        let name = tcp_cfg.name.clone();
256        let h = std::thread::Builder::new()
257            .name(format!("tcp-proxy-{}", name))
258            .spawn(move || {
259                let proxy = crate::tcp_proxy::TcpProxy::new(backends)
260                    .connect_timeout_ms(timeout_ms);
261                if let Err(e) = proxy.bind(&listen) {
262                    eprintln!("[tcp_proxy:{}] {}", name, e);
263                }
264            })
265            .expect("failed to spawn tcp proxy thread");
266        handles.push(h);
267    }
268
269    for udp_cfg in &config.udp_proxies {
270        let listen = udp_cfg.listen.clone();
271        let backends = udp_cfg.backends.clone();
272        let reply_timeout_ms = udp_cfg.reply_timeout_ms;
273        let buffer_size = udp_cfg.buffer_size;
274        let name = udp_cfg.name.clone();
275        let h = std::thread::Builder::new()
276            .name(format!("udp-proxy-{}", name))
277            .spawn(move || {
278                let proxy = crate::udp_proxy::UdpProxy::new(backends)
279                    .reply_timeout_ms(reply_timeout_ms)
280                    .buffer_size(buffer_size);
281                if let Err(e) = proxy.bind(&listen) {
282                    eprintln!("[udp_proxy:{}] {}", name, e);
283                }
284            })
285            .expect("failed to spawn udp proxy thread");
286        handles.push(h);
287    }
288
289    for ws_cfg in &config.ws_proxies {
290        let listen = ws_cfg.listen.clone();
291        let backends = ws_cfg.backends.clone();
292        let connect_timeout_ms = ws_cfg.connect_timeout_ms;
293        let read_timeout_ms = ws_cfg.read_timeout_ms;
294        let name = ws_cfg.name.clone();
295
296        // With a health check configured, share a live-backend list with a
297        // background checker thread (same mechanism `[[upstream]]` pools
298        // use) instead of treating every configured backend as always live.
299        let live = Arc::new(RwLock::new(backends.clone()));
300        if let Some(ref hc) = ws_cfg.health_check {
301            crate::proxy_config::health::start_health_checker(
302                name.clone(),
303                backends.clone(),
304                Arc::clone(&live),
305                hc.clone(),
306                Arc::new(AtomicBool::new(false)),
307                None,
308            );
309        }
310
311        let h = std::thread::Builder::new()
312            .name(format!("ws-proxy-{}", name))
313            .spawn(move || {
314                let proxy = crate::ws_proxy::WsProxy::with_live_backends(backends, live)
315                    .connect_timeout_ms(connect_timeout_ms)
316                    .read_timeout_ms(read_timeout_ms);
317                if let Err(e) = proxy.bind(&listen) {
318                    eprintln!("[ws_proxy:{}] {}", name, e);
319                }
320            })
321            .expect("failed to spawn ws proxy thread");
322        handles.push(h);
323    }
324
325    handles
326}
327
328/// Wrap `handler` with the middleware layers described by `mw`, innermost
329/// (closest to the handler) to outermost (runs first, on every request):
330///
331///   1. Cache (innermost)
332///   2. Rewrite (request + response rules combined)
333///   3. Auth
334///   4. Rate limit
335///   5. IP filter
336///   6. Timeout — bounds this route's total time, including every layer above
337///   7. Max body size (outermost) — a fast, non-blocking length check, so a
338///      request this route will reject outright doesn't pay for a timeout
339///      wrapper (which runs its target on a background thread) or engage
340///      any other layer at all
341/// A route's mirror: a declared upstream's live backends, or its own list.
342fn build_mirror(route: &str, config: &crate::proxy_config::MirrorConfig, lives: &HashMap<String, Arc<RwLock<Vec<String>>>>) -> crate::proxy::mirror::Mirror {
343    let name = if config.upstream.is_empty() { format!("{route}-mirror") } else { config.upstream.clone() };
344    let mirror = match lives.get(&config.upstream) {
345        Some(live) if !config.upstream.is_empty() => crate::proxy::mirror::Mirror::with_live_backends(name, Arc::clone(live)),
346        _ if !config.upstream.is_empty() => crate::proxy::mirror::Mirror::new(name, vec![config.upstream.clone()]),
347        _ => crate::proxy::mirror::Mirror::new(name, config.backends.clone()),
348    };
349    mirror.percent(if config.percent.is_finite() { config.percent } else { 0.0 })
350}
351
352pub(crate) fn apply_middleware(
353    handler: Arc<dyn crate::application::Application + Send + Sync>,
354    mw: &MiddlewareConfig,
355) -> Arc<dyn crate::application::Application + Send + Sync> {
356    apply_middleware_named(handler, mw, "middleware", &mut Vec::new())
357}
358
359/// [`apply_middleware`], recording the route's cache (if any) as `name`
360/// in `caches` for the admin API's registry.
361pub(crate) fn apply_middleware_named(
362    handler: Arc<dyn crate::application::Application + Send + Sync>,
363    mw: &MiddlewareConfig,
364    name: &str,
365    caches: &mut Vec<(String, crate::cache::CacheLayer)>,
366) -> Arc<dyn crate::application::Application + Send + Sync> {
367    // Wrap the Arc<dyn Application> in an ArcApp adapter so it implements
368    // Application itself (needed for WithMiddleware::new).
369    let mut app: Arc<dyn crate::application::Application + Send + Sync> = handler;
370
371    // ── Cache (applied first so it's the innermost middleware) ─────────────
372    if let Some(ref cache_cfg) = mw.cache {
373        let mut layer = if cache_cfg.dir.is_empty() {
374            crate::cache::CacheLayer::memory(1000)
375        } else {
376            let max = crate::proxy_config::cache_max_bytes(&cache_cfg.max_size).unwrap_or(256 * 1024 * 1024);
377            crate::cache::CacheLayer::disk(&cache_cfg.dir, max).unwrap_or_else(|e| {
378                eprintln!("[proxy_config] ERROR: cache dir {}: {e} — this route caches in memory instead", cache_cfg.dir);
379                crate::cache::CacheLayer::memory(1000)
380            })
381        }
382        .ttl(cache_cfg.ttl_secs);
383        if !cache_cfg.purge_token_env.is_empty() {
384            layer = layer.purge_token(std::env::var(&cache_cfg.purge_token_env).unwrap_or_default().trim());
385        }
386        caches.push((name.to_string(), layer.clone()));
387        for vh in &cache_cfg.vary_by {
388            layer = layer.vary_by_header(vh.as_str());
389        }
390        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
391    }
392
393    // ── Rewrite ────────────────────────────────────────────────────────────
394    if !mw.rewrite_request.is_empty() || !mw.rewrite_response.is_empty() {
395        let mut layer = crate::rewrite::RewriteLayer::new();
396        for rule in &mw.rewrite_request {
397            layer = apply_request_rewrite_rule(layer, rule);
398        }
399        for rule in &mw.rewrite_response {
400            layer = apply_response_rewrite_rule(layer, rule);
401        }
402        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
403    }
404
405    // ── WebAssembly plugin (inside auth: it sees authenticated requests) ──
406    if let Some(ref wasm) = mw.wasm {
407        #[cfg(feature = "wasm-plugins")]
408        let layer = crate::wasm_plugin::WasmLayer::from_file(&wasm.module)
409            .map(|l| l.fuel(wasm.fuel).memory_mb(wasm.memory_mb).config(wasm.config.clone()));
410        #[cfg(not(feature = "wasm-plugins"))]
411        let layer: Result<crate::proxy_config::PluginMisconfigured, String> =
412            Err(format!("plugin {}: this rws binary was built without the wasm-plugins feature", wasm.module));
413        app = match layer {
414            Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
415            Err(e) => {
416                eprintln!("[proxy_config] ERROR: {e} — this route answers 503 until the plugin is fixed.");
417                arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(crate::proxy_config::PluginMisconfigured))
418            }
419        };
420    }
421
422    // ── Authorization (inside auth, which sets the caller) ─────────────────
423    if let Some(ref rule) = mw.authz {
424        let layer = match crate::authz::config::current() {
425            Some(Ok(authorizer)) => crate::authz::AuthzLayer::new(authorizer, &rule.relation, &rule.object),
426            Some(Err(e)) => Err(format!("[authz] couldn't be built: {e}")),
427            None => Err("the route has [route.middleware.authz] but there's no [authz] section".to_string()),
428        };
429        app = match layer {
430            Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
431            Err(e) => {
432                eprintln!("[proxy_config] ERROR: authz: {e} — this route answers 503 until it's fixed.");
433                arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(AuthMisconfigured))
434            }
435        };
436    }
437
438    // ── Auth ───────────────────────────────────────────────────────────────
439    // Fails closed: an auth setting that can't be honoured (secret unset,
440    // htpasswd missing, feature not compiled in) puts AuthMisconfigured in
441    // front of the route instead of silently serving it unauthenticated.
442    if let Some(ref auth_cfg) = mw.auth {
443        let deny = |reason: String| -> Arc<dyn crate::application::Application + Send + Sync> {
444            eprintln!(
445                "[proxy_config] ERROR: {reason} — this route answers 503 until the auth configuration is fixed."
446            );
447            arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(AuthMisconfigured))
448        };
449        app = match auth_cfg {
450            AuthConfig::Bearer { token_env } => {
451                let token = std::env::var(token_env).unwrap_or_default();
452                if token.is_empty() {
453                    deny(format!("bearer auth: env var '{token_env}' is unset or empty"))
454                } else {
455                    arc_app(
456                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
457                            .wrap(BearerAuthMiddleware { token: Arc::new(token) }),
458                    )
459                }
460            }
461            #[cfg(feature = "auth")]
462            AuthConfig::Jwt { secret_env } => {
463                let secret = std::env::var(secret_env).unwrap_or_default();
464                if secret.is_empty() {
465                    deny(format!("JWT auth: env var '{secret_env}' is unset or empty"))
466                } else {
467                    arc_app(
468                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
469                            .wrap(crate::auth::JwtLayer::new(secret)),
470                    )
471                }
472            }
473            #[cfg(not(feature = "auth"))]
474            AuthConfig::Jwt { .. } => {
475                deny("JWT auth requires the 'auth' feature, which this build doesn't include".to_string())
476            }
477            #[cfg(feature = "auth")]
478            AuthConfig::Basic { htpasswd_file } => {
479                match crate::auth::BasicAuthLayer::from_htpasswd_file(htpasswd_file) {
480                    Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
481                    Err(e) => deny(format!("basic auth: {e}")),
482                }
483            }
484            #[cfg(not(feature = "auth"))]
485            AuthConfig::Basic { .. } => {
486                deny("basic auth requires the 'auth' feature, which this build doesn't include".to_string())
487            }
488        };
489    }
490
491    // ── Rate limit ─────────────────────────────────────────────────────────
492    if let Some(ref rl_cfg) = mw.rate_limit {
493        let limiter = Arc::new(crate::rate_limit::RateLimiter::new(
494            rl_cfg.max_requests,
495            rl_cfg.window_secs,
496        ));
497        app = arc_app(
498            WithMiddleware::new(ArcApp(Arc::clone(&app)))
499                .wrap(PerRouteRateLimit(limiter)),
500        );
501    }
502
503    // ── IP filter ────────────────────────────────────────────────────────────
504    if !mw.ip_allow.is_empty() {
505        let filter = crate::ip_filter::IpFilter::allow(mw.ip_allow.iter().map(|s| s.as_str()));
506        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
507    } else if !mw.ip_deny.is_empty() {
508        let filter = crate::ip_filter::IpFilter::deny(mw.ip_deny.iter().map(|s| s.as_str()));
509        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
510    }
511
512    // ── Timeout (bounds this route's total time, including every other
513    // middleware above) ─────────────────────────────────────────────────────
514    if let Some(timeout_ms) = mw.timeout_ms {
515        app = arc_app(crate::timeout::TimeoutLayer::from_arc(
516            app,
517            std::time::Duration::from_millis(timeout_ms),
518        ));
519    }
520
521    // ── Max body size (outermost — see PerRouteMaxBodySize's docs for why
522    // this is not a substitute for RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES) ──────
523    if let Some(max_body_size) = mw.max_body_size {
524        app = arc_app(
525            WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(PerRouteMaxBodySize(max_body_size)),
526        );
527    }
528
529    app
530}
531
532fn apply_request_rewrite_rule(
533    layer: crate::rewrite::RewriteLayer,
534    rule: &crate::proxy_config::RewriteRuleConfig,
535) -> crate::rewrite::RewriteLayer {
536    match rule.type_.as_str() {
537        "header_set" => {
538            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
539                return layer.request_header_set(name, value);
540            }
541        }
542        "header_remove" => {
543            if let Some(name) = &rule.name {
544                return layer.request_header_remove(name);
545            }
546        }
547        "uri_set" => {
548            if let Some(value) = &rule.value {
549                return layer.request_uri_set(value);
550            }
551        }
552        "uri_strip_prefix" | "strip_prefix" => {
553            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
554                return layer.request_uri_strip_prefix(prefix);
555            }
556        }
557        "uri_add_prefix" | "add_prefix" => {
558            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
559                return layer.request_uri_add_prefix(prefix);
560            }
561        }
562        _ => {}
563    }
564    layer
565}
566
567fn apply_response_rewrite_rule(
568    layer: crate::rewrite::RewriteLayer,
569    rule: &crate::proxy_config::RewriteRuleConfig,
570) -> crate::rewrite::RewriteLayer {
571    match rule.type_.as_str() {
572        "header_set" => {
573            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
574                return layer.response_header_set(name, value);
575            }
576        }
577        "header_remove" => {
578            if let Some(name) = &rule.name {
579                return layer.response_header_remove(name);
580            }
581        }
582        "status" => {
583            if let (Some(code), Some(reason)) = (&rule.code, &rule.reason) {
584                return layer.response_status(*code as i16, reason);
585            }
586        }
587        "body_replace" => {
588            if let (Some(from), Some(to)) = (&rule.from, &rule.to) {
589                return layer.response_body_replace(from, to);
590            }
591        }
592        _ => {}
593    }
594    layer
595}
596
597// ── ArcApp adapter ─────────────────────────────────────────────────────────────
598
599/// Wraps `Arc<dyn Application + Send + Sync>` to implement `Application`
600/// (needed because you can't implement a foreign trait on a foreign type).
601pub(crate) struct ArcApp(pub(crate) Arc<dyn crate::application::Application + Send + Sync>);
602
603impl crate::application::Application for ArcApp {
604    fn execute(
605        &self,
606        request: &crate::request::Request,
607        conn: &crate::server::ConnectionInfo,
608    ) -> Result<crate::response::Response, String> {
609        self.0.execute(request, conn)
610    }
611
612    fn streams_body(&self, request: &crate::request::Request, conn: &crate::server::ConnectionInfo) -> bool {
613        self.0.streams_body(request, conn)
614    }
615}
616
617impl Clone for ArcApp {
618    fn clone(&self) -> Self {
619        ArcApp(Arc::clone(&self.0))
620    }
621}
622
623/// The authorizer `[authz]` describes: its schema, the process-wide tuple
624/// store, the seed tuples written. `None` without `[authz]`.
625pub(crate) fn config_authorizer(config: &ProxyConfig) -> Option<Result<crate::authz::Authorizer, String>> {
626    config_authorizer_with(config, None)
627}
628
629/// [`config_authorizer`], or over `check_store` whatever `store` says — a
630/// throwaway store for `rws --check`, which must not connect or write.
631pub(crate) fn config_authorizer_with(
632    config: &ProxyConfig,
633    check_store: Option<Arc<dyn crate::authz::TupleStore>>,
634) -> Option<Result<crate::authz::Authorizer, String>> {
635    let authz = config.authz.as_ref()?;
636    Some((|| {
637        let mut schema = crate::authz::Schema::new();
638        for (kind, relations) in &authz.types {
639            schema = if relations.is_empty() { schema.object_type(kind) } else { Ok(schema) }.map_err(|e| e.to_string())?;
640            for (relation, definition) in relations {
641                schema = schema.relation(kind, relation, definition).map_err(|e| format!("type {kind}, relation {relation}: {e}"))?;
642            }
643        }
644        let store = match (authz.store.as_str(), check_store) {
645            ("" | "memory" | "sql", Some(store)) => store,
646            ("" | "memory", None) => crate::authz::config::memory_store(),
647            #[cfg(any(feature = "model-sqlite", feature = "model-postgres", feature = "model-mysql"))]
648            ("sql", None) => crate::authz::sql::config_store()?,
649            #[cfg(not(any(feature = "model-sqlite", feature = "model-postgres", feature = "model-mysql")))]
650            ("sql", None) => return Err("store = \"sql\" needs a build with a model-sqlite / model-postgres / model-mysql feature".to_string()),
651            (other, _) => return Err(format!("store = \"{other}\" isn't supported (memory or sql)")),
652        };
653        let authorizer = crate::authz::Authorizer::with_store(schema, store).map_err(|e| e.to_string())?;
654        for tuple in &authz.tuples {
655            authorizer.write_str(tuple).map_err(|e| format!("tuple {tuple:?}: {e}"))?;
656        }
657        Ok(authorizer)
658    })())
659}