Skip to main content

rws_proxy/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.
53#[doc(hidden)]
54pub fn compile_routes(config: &ProxyConfig, health_stop: &Arc<AtomicBool>) -> Vec<CompiledRoute> {
55    // ── 0. The [authz] authorizer, for routes' authz middleware ─────────────
56    let authorizer = config_authorizer(config);
57    if let Some(Err(e)) = &authorizer {
58        eprintln!("[proxy_config] ERROR: [authz]: {e} — routes with authz answer 503 until it's fixed.");
59    }
60    crate::authz::config::set_current(authorizer.clone());
61    // Routes compiled below take this authorizer, not the process-wide slot,
62    // which a concurrent compile (tests build configs in parallel) may
63    // already have replaced.
64    let _compiling = CompilingAuthorizer::enter(authorizer);
65    // ── 1. Build upstream name → live backends map ─────────────────────────
66    let mut upstream_lives: HashMap<String, Arc<RwLock<Vec<String>>>> = HashMap::new();
67    // tls_ca_file trust settings per upstream, loaded once per compile (so a
68    // reload picks up a rotated CA). A CA file that can't be loaded fails
69    // closed: every handshake fails (502) instead of trusting public CAs.
70    let mut upstream_trust: HashMap<String, Arc<crate::proxy::UpstreamTls>> = HashMap::new();
71    // Weights, stickiness, circuit breaker, retries: one state per
72    // upstream, shared by its routes.
73    let mut upstream_policies: HashMap<String, Arc<crate::proxy_config::upstream::UpstreamPolicy>> = HashMap::new();
74
75    for upstream in &config.upstreams {
76        let live = Arc::new(RwLock::new(upstream.backends.clone()));
77        upstream_lives.insert(upstream.name.clone(), Arc::clone(&live));
78        upstream_policies.insert(
79            upstream.name.clone(),
80            Arc::new(crate::proxy_config::upstream::UpstreamPolicy::new(
81                &upstream.name,
82                Arc::clone(&live),
83                &upstream.backends,
84                &upstream.weights,
85                upstream.sticky.clone(),
86                upstream.circuit_breaker.as_ref(),
87                upstream.retry.clone(),
88            )),
89        );
90
91        let client = upstream.tls_client_cert_file.as_deref().zip(upstream.tls_client_key_file.as_deref());
92        let configured = upstream.tls_ca_file.is_some() || upstream.tls_client_cert_file.is_some() || upstream.tls_client_key_file.is_some();
93        let trust = configured.then(|| {
94            Arc::new(crate::proxy::UpstreamTls::new(upstream.tls_ca_file.as_deref(), client).unwrap_or_else(|e| {
95                eprintln!("[proxy_config] ERROR: upstream '{}': {e} — its TLS backends answer 502 until this is fixed", upstream.name);
96                crate::proxy::UpstreamTls::reject_all(&upstream.name)
97            }))
98        });
99        if let Some(trust) = &trust {
100            upstream_trust.insert(upstream.name.clone(), Arc::clone(trust));
101        }
102
103        if let Some(ref hc) = upstream.health_check {
104            crate::proxy_config::health::start_health_checker(
105                upstream.name.clone(),
106                upstream.backends.clone(),
107                Arc::clone(&live),
108                hc.clone(),
109                Arc::clone(health_stop),
110                trust,
111            );
112        }
113    }
114
115    crate::proxy_config::upstream::register(upstream_policies.values().cloned().collect());
116
117    // ── 2. Build compiled routes ───────────────────────────────────────────
118    let mut caches: Vec<(String, crate::cache::CacheLayer)> = Vec::new();
119    let mut compiled: Vec<CompiledRoute> = Vec::new();
120
121    for route in &config.routes {
122        let matcher = RouteMatcher::from_match_config(&route.match_);
123
124        let mut async_proxy: Option<Arc<DynamicProxy>> = None;
125        let cache_name = format!("route:{}", if route.name.is_empty() { format!("#{}", compiled.len() + 1) } else { route.name.clone() });
126        let mirror = route.middleware.mirror.as_ref().map(|m| build_mirror(&route.name, m, &upstream_lives));
127
128        // Build the action handler (base application)
129        let base_handler: Arc<dyn crate::application::Application + Send + Sync> =
130            match &route.action {
131                ActionConfig::Proxy {
132                    upstream,
133                    connect_timeout_ms,
134                    read_timeout_ms,
135                    strip_path_prefix,
136                    add_path_prefix,
137                    upstream_host,
138                } => {
139                    let live = upstream_lives
140                        .get(upstream.as_str())
141                        .cloned()
142                        .unwrap_or_else(|| {
143                            // Fallback: upstream not declared, use name as single backend
144                            Arc::new(RwLock::new(vec![upstream.clone()]))
145                        });
146                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
147                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
148                    let upstream_strategy = upstream_cfg
149                        .map(|u| u.strategy.clone())
150                        .unwrap_or_else(|| "round_robin".to_string());
151                    let proxy = Arc::new(DynamicProxy::new(
152                        live,
153                        *connect_timeout_ms,
154                        *read_timeout_ms,
155                        strip_path_prefix.clone(),
156                        add_path_prefix.clone(),
157                        upstream_tls,
158                        upstream_strategy,
159                    )
160                    .with_trust(upstream_trust.get(upstream.as_str()).cloned())
161                    .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned())
162                    .with_mirror(mirror.clone())
163                    .with_early_hints(&route.middleware.early_hints)
164                    .with_h3(upstream_cfg.is_some_and(|u| u.protocol.eq_ignore_ascii_case("h3")))
165                    .with_upstream_host(upstream_host.clone()));
166                    if proxy.supports_async() && route.middleware.is_empty() {
167                        async_proxy = Some(Arc::clone(&proxy));
168                    }
169                    proxy
170                }
171
172                ActionConfig::Redirect { location, status } => {
173                    arc_app(RedirectAdapter::new(location.clone(), *status))
174                }
175
176                ActionConfig::Respond { status, body, content_type } => {
177                    arc_app(RespondAdapter::new(*status, body.clone(), content_type.clone()))
178                }
179
180                ActionConfig::Static { root, index } => {
181                    arc_app(StaticAdapter::new(root.clone(), index.clone()))
182                }
183
184                // Grpc action: HTTP/2 to the upstream (HTTP/1.1 only in builds
185                // without the http2 feature, where gRPC can't work anyway).
186                ActionConfig::Grpc { upstream, connect_timeout_ms, read_timeout_ms } => {
187                    let live = upstream_lives
188                        .get(upstream.as_str())
189                        .cloned()
190                        .unwrap_or_else(|| Arc::new(RwLock::new(vec![upstream.clone()])));
191                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
192                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
193                    #[cfg(feature = "http2")]
194                    {
195                        let _ = upstream_cfg;
196                        arc_app(crate::proxy_config::H2Route::new(
197                            live,
198                            upstream_tls,
199                            upstream_trust.get(upstream.as_str()).cloned(),
200                            *connect_timeout_ms,
201                            *read_timeout_ms,
202                        ))
203                    }
204                    #[cfg(not(feature = "http2"))]
205                    {
206                        let upstream_strategy = upstream_cfg
207                            .map(|u| u.strategy.clone())
208                            .unwrap_or_else(|| "round_robin".to_string());
209                        arc_app(
210                            DynamicProxy::new(live, *connect_timeout_ms, *read_timeout_ms, None, None, upstream_tls, upstream_strategy)
211                                .with_trust(upstream_trust.get(upstream.as_str()).cloned())
212                                .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned()),
213                        )
214                    }
215                }
216
217                ActionConfig::FastCgi(f) => {
218                    let mut cfg = crate::fastcgi::FastCgiConfig::new(f.address.clone(), f.root.clone());
219                    cfg.index = f.index.clone();
220                    cfg.split_path = f.split_path.clone();
221                    cfg.front_controller = f.front_controller.clone();
222                    cfg.params = f.params.clone();
223                    cfg.connect_timeout = std::time::Duration::from_millis(f.connect_timeout_ms);
224                    cfg.read_timeout = std::time::Duration::from_millis(f.read_timeout_ms);
225                    cfg.max_idle = f.max_idle as usize;
226                    match crate::fastcgi::FastCgi::new(cfg) {
227                        Ok(app) => arc_app(app.with_static(arc_app(StaticAdapter::new(f.root.clone(), Vec::new())))),
228                        Err(e) => {
229                            eprintln!("[config] route {cache_name}: {e}; it answers 502");
230                            arc_app(RespondAdapter::new(502, "Bad Gateway".to_string(), "text/plain".to_string()))
231                        }
232                    }
233                }
234                ActionConfig::Mcp(gateway) => arc_app(crate::proxy_config::mcp_gateway::GatewayAdapter::start(
235                    (**gateway).clone(),
236                    route.match_.path.clone().unwrap_or_else(|| "/mcp".to_string()),
237                )),
238                ActionConfig::Llm(llm) => arc_app(crate::proxy_config::llm::build(llm)),
239                ActionConfig::A2a(gateway) => arc_app(crate::proxy_config::a2a_gateway::A2aGateway::new(
240                    (**gateway).clone(),
241                    route.match_.path.as_deref().unwrap_or("/"),
242                )),
243                ActionConfig::Unknown(_) => {
244                    // No-op: these fall through to fallback
245                    arc_app(crate::proxy_config::NullApp)
246                }
247            };
248
249        // Wrap the base handler with middleware layers. A proxy route sends
250        // its mirror copies and early hints itself (so it keeps the async
251        // forwarder); any other action gets a MirrorLayer / EarlyHintsLayer.
252        let mut handler = apply_middleware_named(base_handler, &route.middleware, &cache_name, &mut caches);
253        if !matches!(route.action, ActionConfig::Proxy { .. }) {
254            if let Some(mirror) = &mirror {
255                handler = arc_app(WithMiddleware::new(ArcApp(handler)).wrap(crate::proxy::mirror::MirrorLayer::new(mirror.clone())));
256            }
257            if !route.middleware.early_hints.is_empty() {
258                let layer = crate::early_hints::EarlyHintsLayer::new(route.middleware.early_hints.iter().cloned());
259                handler = arc_app(WithMiddleware::new(ArcApp(handler)).wrap(layer));
260            }
261        }
262
263        compiled.push(CompiledRoute { matcher, handler, async_proxy });
264    }
265    crate::cache::registry::replace_group("route:", caches);
266    compiled
267}
268
269/// Start the `[[tcp_proxy]]` / `[[udp_proxy]]` / `[[ws_proxy]]` listeners.
270/// Startup only: a hot reload can't move bound sockets, so changes to these
271/// sections need a restart.
272fn spawn_listeners(config: &ProxyConfig) -> Vec<std::thread::JoinHandle<()>> {
273    let mut handles: Vec<std::thread::JoinHandle<()>> = Vec::new();
274
275    for tcp_cfg in &config.tcp_proxies {
276        let listen = tcp_cfg.listen.clone();
277        let backends = tcp_cfg.backends.clone();
278        let timeout_ms = tcp_cfg.connect_timeout_ms;
279        let name = tcp_cfg.name.clone();
280        let h = std::thread::Builder::new()
281            .name(format!("tcp-proxy-{}", name))
282            .spawn(move || {
283                let proxy = crate::tcp_proxy::TcpProxy::new(backends)
284                    .connect_timeout_ms(timeout_ms);
285                if let Err(e) = proxy.bind(&listen) {
286                    eprintln!("[tcp_proxy:{}] {}", name, e);
287                }
288            })
289            .expect("failed to spawn tcp proxy thread");
290        handles.push(h);
291    }
292
293    for udp_cfg in &config.udp_proxies {
294        let listen = udp_cfg.listen.clone();
295        let backends = udp_cfg.backends.clone();
296        let reply_timeout_ms = udp_cfg.reply_timeout_ms;
297        let buffer_size = udp_cfg.buffer_size;
298        let name = udp_cfg.name.clone();
299        let h = std::thread::Builder::new()
300            .name(format!("udp-proxy-{}", name))
301            .spawn(move || {
302                let proxy = crate::udp_proxy::UdpProxy::new(backends)
303                    .reply_timeout_ms(reply_timeout_ms)
304                    .buffer_size(buffer_size);
305                if let Err(e) = proxy.bind(&listen) {
306                    eprintln!("[udp_proxy:{}] {}", name, e);
307                }
308            })
309            .expect("failed to spawn udp proxy thread");
310        handles.push(h);
311    }
312
313    for ws_cfg in &config.ws_proxies {
314        let listen = ws_cfg.listen.clone();
315        let backends = ws_cfg.backends.clone();
316        let connect_timeout_ms = ws_cfg.connect_timeout_ms;
317        let read_timeout_ms = ws_cfg.read_timeout_ms;
318        let name = ws_cfg.name.clone();
319
320        // With a health check configured, share a live-backend list with a
321        // background checker thread (same mechanism `[[upstream]]` pools
322        // use) instead of treating every configured backend as always live.
323        let live = Arc::new(RwLock::new(backends.clone()));
324        if let Some(ref hc) = ws_cfg.health_check {
325            crate::proxy_config::health::start_health_checker(
326                name.clone(),
327                backends.clone(),
328                Arc::clone(&live),
329                hc.clone(),
330                Arc::new(AtomicBool::new(false)),
331                None,
332            );
333        }
334
335        let h = std::thread::Builder::new()
336            .name(format!("ws-proxy-{}", name))
337            .spawn(move || {
338                let proxy = crate::ws_proxy::WsProxy::with_live_backends(backends, live)
339                    .connect_timeout_ms(connect_timeout_ms)
340                    .read_timeout_ms(read_timeout_ms);
341                if let Err(e) = proxy.bind(&listen) {
342                    eprintln!("[ws_proxy:{}] {}", name, e);
343                }
344            })
345            .expect("failed to spawn ws proxy thread");
346        handles.push(h);
347    }
348
349    handles
350}
351
352/// Wrap `handler` with the middleware layers described by `mw`, innermost
353/// (closest to the handler) to outermost (runs first, on every request):
354///
355///   1. Cache (innermost)
356///   2. Rewrite (request + response rules combined)
357///   3. Auth
358///   4. Rate limit
359///   5. IP filter
360///   6. Timeout — bounds this route's total time, including every layer above
361///   7. Max body size (outermost) — a fast, non-blocking length check, so a
362///      request this route will reject outright doesn't pay for a timeout
363///      wrapper (which runs its target on a background thread) or engage
364///      any other layer at all
365/// A route's mirror: a declared upstream's live backends, or its own list.
366fn build_mirror(route: &str, config: &crate::proxy_config::MirrorConfig, lives: &HashMap<String, Arc<RwLock<Vec<String>>>>) -> crate::proxy::mirror::Mirror {
367    let name = if config.upstream.is_empty() { format!("{route}-mirror") } else { config.upstream.clone() };
368    let mirror = match lives.get(&config.upstream) {
369        Some(live) if !config.upstream.is_empty() => crate::proxy::mirror::Mirror::with_live_backends(name, Arc::clone(live)),
370        _ if !config.upstream.is_empty() => crate::proxy::mirror::Mirror::new(name, vec![config.upstream.clone()]),
371        _ => crate::proxy::mirror::Mirror::new(name, config.backends.clone()),
372    };
373    mirror.percent(if config.percent.is_finite() { config.percent } else { 0.0 })
374}
375
376thread_local! {
377    static COMPILING_AUTHORIZER: std::cell::RefCell<Option<Option<Result<crate::authz::Authorizer, String>>>> =
378        const { std::cell::RefCell::new(None) };
379}
380
381/// The `[authz]` authorizer of the compile running on this thread, for its
382/// routes' authz middleware; outside a compile, the process-wide one.
383struct CompilingAuthorizer(Option<Option<Result<crate::authz::Authorizer, String>>>);
384
385impl CompilingAuthorizer {
386    fn enter(authorizer: Option<Result<crate::authz::Authorizer, String>>) -> Self {
387        CompilingAuthorizer(COMPILING_AUTHORIZER.with(|c| c.replace(Some(authorizer))))
388    }
389
390    fn current() -> Option<Result<crate::authz::Authorizer, String>> {
391        COMPILING_AUTHORIZER.with(|c| c.borrow().clone()).unwrap_or_else(crate::authz::config::current)
392    }
393}
394
395impl Drop for CompilingAuthorizer {
396    fn drop(&mut self) {
397        let previous = self.0.take();
398        COMPILING_AUTHORIZER.with(|c| *c.borrow_mut() = previous);
399    }
400}
401
402#[doc(hidden)]
403pub fn apply_middleware(
404    handler: Arc<dyn crate::application::Application + Send + Sync>,
405    mw: &MiddlewareConfig,
406) -> Arc<dyn crate::application::Application + Send + Sync> {
407    apply_middleware_named(handler, mw, "middleware", &mut Vec::new())
408}
409
410/// [`apply_middleware`], recording the route's cache (if any) as `name`
411/// in `caches` for the admin API's registry.
412#[doc(hidden)]
413pub fn apply_middleware_named(
414    handler: Arc<dyn crate::application::Application + Send + Sync>,
415    mw: &MiddlewareConfig,
416    name: &str,
417    caches: &mut Vec<(String, crate::cache::CacheLayer)>,
418) -> Arc<dyn crate::application::Application + Send + Sync> {
419    // Wrap the Arc<dyn Application> in an ArcApp adapter so it implements
420    // Application itself (needed for WithMiddleware::new).
421    let mut app: Arc<dyn crate::application::Application + Send + Sync> = handler;
422
423    // ── Cache (applied first so it's the innermost middleware) ─────────────
424    if let Some(ref cache_cfg) = mw.cache {
425        let mut layer = if cache_cfg.dir.is_empty() {
426            crate::cache::CacheLayer::memory(1000)
427        } else {
428            let max = crate::proxy_config::cache_max_bytes(&cache_cfg.max_size).unwrap_or(256 * 1024 * 1024);
429            crate::cache::CacheLayer::disk(&cache_cfg.dir, max).unwrap_or_else(|e| {
430                eprintln!("[proxy_config] ERROR: cache dir {}: {e} — this route caches in memory instead", cache_cfg.dir);
431                crate::cache::CacheLayer::memory(1000)
432            })
433        }
434        .ttl(cache_cfg.ttl_secs);
435        if !cache_cfg.purge_token_env.is_empty() {
436            layer = layer.purge_token(std::env::var(&cache_cfg.purge_token_env).unwrap_or_default().trim());
437        }
438        caches.push((name.to_string(), layer.clone()));
439        for vh in &cache_cfg.vary_by {
440            layer = layer.vary_by_header(vh.as_str());
441        }
442        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
443    }
444
445    // ── Rewrite ────────────────────────────────────────────────────────────
446    if !mw.rewrite_request.is_empty() || !mw.rewrite_response.is_empty() {
447        let mut layer = crate::rewrite::RewriteLayer::new();
448        for rule in &mw.rewrite_request {
449            layer = apply_request_rewrite_rule(layer, rule);
450        }
451        for rule in &mw.rewrite_response {
452            layer = apply_response_rewrite_rule(layer, rule);
453        }
454        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
455    }
456
457    // ── WebAssembly plugin (inside auth: it sees authenticated requests) ──
458    if let Some(ref wasm) = mw.wasm {
459        #[cfg(feature = "wasm-plugins")]
460        let layer = crate::wasm_plugin::WasmLayer::from_file(&wasm.module)
461            .map(|l| l.fuel(wasm.fuel).memory_mb(wasm.memory_mb).config(wasm.config.clone()));
462        #[cfg(not(feature = "wasm-plugins"))]
463        let layer: Result<crate::proxy_config::PluginMisconfigured, String> =
464            Err(format!("plugin {}: this rws binary was built without the wasm-plugins feature", wasm.module));
465        app = match layer {
466            Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
467            Err(e) => {
468                eprintln!("[proxy_config] ERROR: {e} — this route answers 503 until the plugin is fixed.");
469                arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(crate::proxy_config::PluginMisconfigured))
470            }
471        };
472    }
473
474    // ── Authorization (inside auth, which sets the caller) ─────────────────
475    if let Some(ref rule) = mw.authz {
476        let layer = match CompilingAuthorizer::current() {
477            Some(Ok(authorizer)) => crate::authz::AuthzLayer::new(authorizer, &rule.relation, &rule.object),
478            Some(Err(e)) => Err(format!("[authz] couldn't be built: {e}")),
479            None => Err("the route has [route.middleware.authz] but there's no [authz] section".to_string()),
480        };
481        app = match layer {
482            Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
483            Err(e) => {
484                eprintln!("[proxy_config] ERROR: authz: {e} — this route answers 503 until it's fixed.");
485                arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(AuthMisconfigured))
486            }
487        };
488    }
489
490    // ── Auth ───────────────────────────────────────────────────────────────
491    // Fails closed: an auth setting that can't be honoured (secret unset,
492    // htpasswd missing, feature not compiled in) puts AuthMisconfigured in
493    // front of the route instead of silently serving it unauthenticated.
494    if let Some(ref auth_cfg) = mw.auth {
495        let deny = |reason: String| -> Arc<dyn crate::application::Application + Send + Sync> {
496            eprintln!(
497                "[proxy_config] ERROR: {reason} — this route answers 503 until the auth configuration is fixed."
498            );
499            arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(AuthMisconfigured))
500        };
501        app = match auth_cfg {
502            AuthConfig::Bearer { token_env } => {
503                let token = std::env::var(token_env).unwrap_or_default();
504                if token.is_empty() {
505                    deny(format!("bearer auth: env var '{token_env}' is unset or empty"))
506                } else {
507                    arc_app(
508                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
509                            .wrap(BearerAuthMiddleware { token: Arc::new(token) }),
510                    )
511                }
512            }
513            #[cfg(feature = "auth")]
514            AuthConfig::Jwt { secret_env, previous_secret_env } => {
515                let secret = std::env::var(secret_env).unwrap_or_default();
516                let previous = (!previous_secret_env.is_empty()).then(|| std::env::var(previous_secret_env).unwrap_or_default()).filter(|s| !s.is_empty());
517                if secret.is_empty() {
518                    deny(format!("JWT auth: env var '{secret_env}' is unset or empty"))
519                } else {
520                    arc_app(
521                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
522                            .wrap(crate::auth::JwtLayer::secrets(std::iter::once(secret).chain(previous))),
523                    )
524                }
525            }
526            #[cfg(not(feature = "auth"))]
527            AuthConfig::Jwt { .. } => {
528                deny("JWT auth requires the 'auth' feature, which this build doesn't include".to_string())
529            }
530            #[cfg(feature = "auth")]
531            AuthConfig::Basic { htpasswd_file } => {
532                match crate::auth::BasicAuthLayer::from_htpasswd_file(htpasswd_file) {
533                    Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
534                    Err(e) => deny(format!("basic auth: {e}")),
535                }
536            }
537            #[cfg(not(feature = "auth"))]
538            AuthConfig::Basic { .. } => {
539                deny("basic auth requires the 'auth' feature, which this build doesn't include".to_string())
540            }
541            #[cfg(feature = "sso")]
542            AuthConfig::Oidc(oidc) => match crate::proxy_config::oidc_route::LazyOidc::new((**oidc).clone()) {
543                Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
544                Err(e) => deny(e),
545            },
546            #[cfg(not(feature = "sso"))]
547            AuthConfig::Oidc(_) => deny("oidc auth requires the 'sso' feature, which this build doesn't include".to_string()),
548            #[cfg(feature = "auth")]
549            AuthConfig::Forward { url, copy_headers, timeout_ms } => {
550                if url.starts_with("http://") || url.starts_with("https://") {
551                    let layer = copy_headers.iter().fold(crate::auth::forward::ForwardAuthLayer::new(url.clone()).timeout_ms(*timeout_ms), |l, h| l.copy_header(h.clone()));
552                    arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer))
553                } else {
554                    deny(format!("forward auth: url {url:?} is not an http(s) URL"))
555                }
556            }
557            AuthConfig::Unknown(kind) => deny(format!("auth type {kind:?} is not one of bearer, jwt, basic, oidc, forward")),
558            #[cfg(not(feature = "auth"))]
559            AuthConfig::Forward { .. } => deny("forward auth requires the 'auth' feature, which this build doesn't include".to_string()),
560        };
561    }
562
563    // ── Web application firewall (outside auth: login pages are covered;
564    //    inside the rate limit and IP filter, which are cheaper) ─────────
565    if let Some(ref waf) = mw.waf {
566        if waf.config.mode != crate::waf::Mode::Off {
567            app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(crate::waf::WafLayer::new(waf.config.clone())));
568        }
569    }
570
571    // ── Rate limit ─────────────────────────────────────────────────────────
572    if let Some(ref rl_cfg) = mw.rate_limit {
573        let limiter = Arc::new(crate::rate_limit::RateLimiter::new(
574            rl_cfg.max_requests,
575            rl_cfg.window_secs,
576        ));
577        app = arc_app(
578            WithMiddleware::new(ArcApp(Arc::clone(&app)))
579                .wrap(PerRouteRateLimit(limiter)),
580        );
581    }
582
583    // ── IP filter ────────────────────────────────────────────────────────────
584    if !mw.ip_allow.is_empty() {
585        let filter = crate::ip_filter::IpFilter::allow(mw.ip_allow.iter().map(|s| s.as_str()));
586        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
587    } else if !mw.ip_deny.is_empty() {
588        let filter = crate::ip_filter::IpFilter::deny(mw.ip_deny.iter().map(|s| s.as_str()));
589        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
590    }
591
592    // ── Timeout (bounds this route's total time, including every other
593    // middleware above) ─────────────────────────────────────────────────────
594    if let Some(timeout_ms) = mw.timeout_ms {
595        app = arc_app(crate::timeout::TimeoutLayer::from_arc(
596            app,
597            std::time::Duration::from_millis(timeout_ms),
598        ));
599    }
600
601    // ── Max body size (outermost — see PerRouteMaxBodySize's docs for why
602    // this is not a substitute for RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES) ──────
603    if let Some(max_body_size) = mw.max_body_size {
604        app = arc_app(
605            WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(PerRouteMaxBodySize(max_body_size)),
606        );
607    }
608
609    app
610}
611
612fn apply_request_rewrite_rule(
613    layer: crate::rewrite::RewriteLayer,
614    rule: &crate::proxy_config::RewriteRuleConfig,
615) -> crate::rewrite::RewriteLayer {
616    match rule.type_.as_str() {
617        "header_set" => {
618            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
619                return layer.request_header_set(name, value);
620            }
621        }
622        "header_remove" => {
623            if let Some(name) = &rule.name {
624                return layer.request_header_remove(name);
625            }
626        }
627        "uri_set" => {
628            if let Some(value) = &rule.value {
629                return layer.request_uri_set(value);
630            }
631        }
632        "uri_strip_prefix" | "strip_prefix" => {
633            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
634                return layer.request_uri_strip_prefix(prefix);
635            }
636        }
637        "uri_add_prefix" | "add_prefix" => {
638            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
639                return layer.request_uri_add_prefix(prefix);
640            }
641        }
642        _ => {}
643    }
644    layer
645}
646
647fn apply_response_rewrite_rule(
648    layer: crate::rewrite::RewriteLayer,
649    rule: &crate::proxy_config::RewriteRuleConfig,
650) -> crate::rewrite::RewriteLayer {
651    match rule.type_.as_str() {
652        "header_set" => {
653            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
654                return layer.response_header_set(name, value);
655            }
656        }
657        "header_remove" => {
658            if let Some(name) = &rule.name {
659                return layer.response_header_remove(name);
660            }
661        }
662        "status" => {
663            if let (Some(code), Some(reason)) = (&rule.code, &rule.reason) {
664                return layer.response_status(*code as i16, reason);
665            }
666        }
667        "body_replace" => {
668            if let (Some(from), Some(to)) = (&rule.from, &rule.to) {
669                return layer.response_body_replace(from, to);
670            }
671        }
672        _ => {}
673    }
674    layer
675}
676
677// ── ArcApp adapter ─────────────────────────────────────────────────────────────
678
679/// Wraps `Arc<dyn Application + Send + Sync>` to implement `Application`
680/// (needed because you can't implement a foreign trait on a foreign type).
681#[doc(hidden)]
682pub struct ArcApp(pub(crate) Arc<dyn crate::application::Application + Send + Sync>);
683
684impl crate::application::Application for ArcApp {
685    fn execute(
686        &self,
687        request: &crate::request::Request,
688        conn: &crate::server::ConnectionInfo,
689    ) -> Result<crate::response::Response, String> {
690        self.0.execute(request, conn)
691    }
692
693    fn streams_body(&self, request: &crate::request::Request, conn: &crate::server::ConnectionInfo) -> bool {
694        self.0.streams_body(request, conn)
695    }
696}
697
698impl Clone for ArcApp {
699    fn clone(&self) -> Self {
700        ArcApp(Arc::clone(&self.0))
701    }
702}
703
704/// The authorizer `[authz]` describes: its schema, the process-wide tuple
705/// store, the seed tuples written. `None` without `[authz]`.
706#[doc(hidden)]
707pub fn config_authorizer(config: &ProxyConfig) -> Option<Result<crate::authz::Authorizer, String>> {
708    config_authorizer_with(config, None)
709}
710
711/// [`config_authorizer`], or over `check_store` whatever `store` says — a
712/// throwaway store for `rws --check`, which must not connect or write.
713#[doc(hidden)]
714pub fn config_authorizer_with(
715    config: &ProxyConfig,
716    check_store: Option<Arc<dyn crate::authz::TupleStore>>,
717) -> Option<Result<crate::authz::Authorizer, String>> {
718    let authz = config.authz.as_ref()?;
719    Some((|| {
720        let mut schema = crate::authz::Schema::new();
721        for (kind, relations) in &authz.types {
722            schema = if relations.is_empty() { schema.object_type(kind) } else { Ok(schema) }.map_err(|e| e.to_string())?;
723            for (relation, definition) in relations {
724                schema = schema.relation(kind, relation, definition).map_err(|e| format!("type {kind}, relation {relation}: {e}"))?;
725            }
726        }
727        let store = match (authz.store.as_str(), check_store) {
728            ("" | "memory" | "sql", Some(store)) => store,
729            ("" | "memory", None) => crate::authz::config::memory_store(),
730            #[cfg(any(feature = "model-sqlite", feature = "model-postgres", feature = "model-mysql"))]
731            ("sql", None) => crate::authz::sql::config_store()?,
732            #[cfg(not(any(feature = "model-sqlite", feature = "model-postgres", feature = "model-mysql")))]
733            ("sql", None) => return Err("store = \"sql\" needs a build with a model-sqlite / model-postgres / model-mysql feature".to_string()),
734            (other, _) => return Err(format!("store = \"{other}\" isn't supported (memory or sql)")),
735        };
736        let authorizer = crate::authz::Authorizer::with_store(schema, store).map_err(|e| e.to_string())?;
737        for tuple in &authz.tuples {
738            authorizer.write_str(tuple).map_err(|e| format!("tuple {tuple:?}: {e}"))?;
739        }
740        Ok(authorizer)
741    })())
742}
743
744#[cfg(test)]
745mod compiling_authorizer_tests {
746    use super::CompilingAuthorizer;
747
748    fn marker(current: Option<Result<crate::authz::Authorizer, String>>) -> Option<String> {
749        match current {
750            Some(Err(e)) => Some(e),
751            _ => None,
752        }
753    }
754
755    #[test]
756    fn a_compile_sees_its_own_authorizer_and_restores_the_outer_one() {
757        let outer = CompilingAuthorizer::enter(Some(Err("outer".to_string())));
758        assert_eq!(marker(CompilingAuthorizer::current()), Some("outer".to_string()));
759        {
760            let _inner = CompilingAuthorizer::enter(Some(Err("inner".to_string())));
761            assert_eq!(marker(CompilingAuthorizer::current()), Some("inner".to_string()));
762        }
763        assert_eq!(marker(CompilingAuthorizer::current()), Some("outer".to_string()));
764        // `None` (no [authz]) also wins over the process-wide slot.
765        let _none = CompilingAuthorizer::enter(None);
766        assert!(CompilingAuthorizer::current().is_none());
767        drop(outer);
768    }
769
770    #[test]
771    fn another_thread_does_not_see_this_threads_compile() {
772        let _compiling = CompilingAuthorizer::enter(Some(Err("this thread".to_string())));
773        let seen = std::thread::spawn(|| marker(CompilingAuthorizer::current())).join().unwrap();
774        assert_ne!(seen, Some("this thread".to_string()));
775    }
776}