rust_web_server/proxy_config/

builder.rs

//! Builder: converts a `ProxyConfig` into a live `ConfigDrivenApp` with
//! health checkers and L4/WS proxy threads.

use std::collections::HashMap;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, RwLock};

use crate::middleware::WithMiddleware;
use crate::proxy_config::{
    ActionConfig, AuthConfig, CompiledRoute, ConfigDrivenApp, DynamicProxy, MiddlewareConfig,
    ProxyConfig, RedirectAdapter, RespondAdapter, RouteMatcher, StaticAdapter, arc_app,
    AuthMisconfigured, BearerAuthMiddleware, PerRouteMaxBodySize, PerRouteRateLimit,
};

/// Build a `ConfigDrivenApp` from `rws.config.toml` and spawn L4/WS proxy
/// threads. Returns the app and a list of background thread handles.
/// Problems `ProxyConfig::validate` finds are printed first (the same
/// findings `rws --check` reports). The returned app reloads its routes when
/// the configuration is reloaded (`SIGHUP`, `POST /admin/config/reload`) —
/// see [`ConfigDrivenApp::reload_from_file`].
pub fn build_from_file() -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
    let source = std::fs::read_to_string(crate::proxy_config::config_file_path()).unwrap_or_default();
    let config = ProxyConfig::from_str(&source);
    let mut report = crate::config_check::ConfigReport::default();
    config.validate(&mut report);
    for e in &report.errors {
        eprintln!("[proxy_config] ERROR: {e}");
    }
    for w in &report.warnings {
        eprintln!("[proxy_config] Warning: {w}");
    }
    let (app, handles) = build_with_source(config, Some(source));
    let reloadable = app.clone();
    crate::config_reload::on_reload(move || reloadable.reload_from_file());
    (app, handles)
}

/// Build a `ConfigDrivenApp` from a `ProxyConfig` struct.
pub fn build(config: ProxyConfig) -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
    build_with_source(config, None)
}

fn build_with_source(config: ProxyConfig, source: Option<String>) -> (ConfigDrivenApp, Vec<std::thread::JoinHandle<()>>) {
    let health_stop = Arc::new(AtomicBool::new(false));
    let compiled = compile_routes(&config, &health_stop);
    let handles = spawn_listeners(&config);
    (ConfigDrivenApp::new(compiled, config, health_stop, source), handles)
}

/// Compile `config`'s routes and start health checkers for its upstreams;
/// the checkers stop when `health_stop` is set. Used at startup and on every
/// hot reload.
pub(crate) fn compile_routes(config: &ProxyConfig, health_stop: &Arc<AtomicBool>) -> Vec<CompiledRoute> {
    // ── 1. Build upstream name → live backends map ─────────────────────────
    let mut upstream_lives: HashMap<String, Arc<RwLock<Vec<String>>>> = HashMap::new();
    // tls_ca_file trust settings per upstream, loaded once per compile (so a
    // reload picks up a rotated CA). A CA file that can't be loaded fails
    // closed: every handshake fails (502) instead of trusting public CAs.
    let mut upstream_trust: HashMap<String, Arc<crate::proxy::UpstreamTls>> = HashMap::new();
    // Weights, stickiness, circuit breaker, retries: one state per
    // upstream, shared by its routes.
    let mut upstream_policies: HashMap<String, Arc<crate::proxy_config::upstream::UpstreamPolicy>> = HashMap::new();

    for upstream in &config.upstreams {
        let live = Arc::new(RwLock::new(upstream.backends.clone()));
        upstream_lives.insert(upstream.name.clone(), Arc::clone(&live));
        upstream_policies.insert(
            upstream.name.clone(),
            Arc::new(crate::proxy_config::upstream::UpstreamPolicy::new(
                &upstream.name,
                Arc::clone(&live),
                &upstream.backends,
                &upstream.weights,
                upstream.sticky.clone(),
                upstream.circuit_breaker.as_ref(),
                upstream.retry.clone(),
            )),
        );

        let client = upstream.tls_client_cert_file.as_deref().zip(upstream.tls_client_key_file.as_deref());
        let configured = upstream.tls_ca_file.is_some() || upstream.tls_client_cert_file.is_some() || upstream.tls_client_key_file.is_some();
        let trust = configured.then(|| {
            Arc::new(crate::proxy::UpstreamTls::new(upstream.tls_ca_file.as_deref(), client).unwrap_or_else(|e| {
                eprintln!("[proxy_config] ERROR: upstream '{}': {e} — its TLS backends answer 502 until this is fixed", upstream.name);
                crate::proxy::UpstreamTls::reject_all(&upstream.name)
            }))
        });
        if let Some(trust) = &trust {
            upstream_trust.insert(upstream.name.clone(), Arc::clone(trust));
        }

        if let Some(ref hc) = upstream.health_check {
            crate::proxy_config::health::start_health_checker(
                upstream.name.clone(),
                upstream.backends.clone(),
                Arc::clone(&live),
                hc.clone(),
                Arc::clone(health_stop),
                trust,
            );
        }
    }

    crate::proxy_config::upstream::register(upstream_policies.values().cloned().collect());

    // ── 2. Build compiled routes ───────────────────────────────────────────
    let mut compiled: Vec<CompiledRoute> = Vec::new();

    for route in &config.routes {
        let matcher = RouteMatcher::from_match_config(&route.match_);

        let mut async_proxy: Option<Arc<DynamicProxy>> = None;

        // Build the action handler (base application)
        let base_handler: Arc<dyn crate::application::Application + Send + Sync> =
            match &route.action {
                ActionConfig::Proxy {
                    upstream,
                    connect_timeout_ms,
                    read_timeout_ms,
                    strip_path_prefix,
                    add_path_prefix,
                } => {
                    let live = upstream_lives
                        .get(upstream.as_str())
                        .cloned()
                        .unwrap_or_else(|| {
                            // Fallback: upstream not declared, use name as single backend
                            Arc::new(RwLock::new(vec![upstream.clone()]))
                        });
                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
                    let upstream_strategy = upstream_cfg
                        .map(|u| u.strategy.clone())
                        .unwrap_or_else(|| "round_robin".to_string());
                    let proxy = Arc::new(DynamicProxy::new(
                        live,
                        *connect_timeout_ms,
                        *read_timeout_ms,
                        strip_path_prefix.clone(),
                        add_path_prefix.clone(),
                        upstream_tls,
                        upstream_strategy,
                    )
                    .with_trust(upstream_trust.get(upstream.as_str()).cloned())
                    .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned()));
                    if proxy.supports_async() && route.middleware.is_empty() {
                        async_proxy = Some(Arc::clone(&proxy));
                    }
                    proxy
                }

                ActionConfig::Redirect { location, status } => {
                    arc_app(RedirectAdapter::new(location.clone(), *status))
                }

                ActionConfig::Respond { status, body, content_type } => {
                    arc_app(RespondAdapter::new(*status, body.clone(), content_type.clone()))
                }

                ActionConfig::Static { root, index } => {
                    arc_app(StaticAdapter::new(root.clone(), index.clone()))
                }

                // Grpc action: HTTP/2 to the upstream (HTTP/1.1 only in builds
                // without the http2 feature, where gRPC can't work anyway).
                ActionConfig::Grpc { upstream, connect_timeout_ms, read_timeout_ms } => {
                    let live = upstream_lives
                        .get(upstream.as_str())
                        .cloned()
                        .unwrap_or_else(|| Arc::new(RwLock::new(vec![upstream.clone()])));
                    let upstream_cfg = config.upstreams.iter().find(|u| u.name == *upstream);
                    let upstream_tls = upstream_cfg.map_or_else(|| upstream.starts_with("https://"), |u| u.tls);
                    #[cfg(feature = "http2")]
                    {
                        let _ = upstream_cfg;
                        arc_app(crate::proxy_config::H2Route::new(
                            live,
                            upstream_tls,
                            upstream_trust.get(upstream.as_str()).cloned(),
                            *connect_timeout_ms,
                            *read_timeout_ms,
                        ))
                    }
                    #[cfg(not(feature = "http2"))]
                    {
                        let upstream_strategy = upstream_cfg
                            .map(|u| u.strategy.clone())
                            .unwrap_or_else(|| "round_robin".to_string());
                        arc_app(
                            DynamicProxy::new(live, *connect_timeout_ms, *read_timeout_ms, None, None, upstream_tls, upstream_strategy)
                                .with_trust(upstream_trust.get(upstream.as_str()).cloned())
                                .with_policy(upstream, upstream_policies.get(upstream.as_str()).cloned()),
                        )
                    }
                }

                ActionConfig::Mcp(gateway) => arc_app(crate::proxy_config::mcp_gateway::GatewayAdapter::start(
                    (**gateway).clone(),
                    route.match_.path.clone().unwrap_or_else(|| "/mcp".to_string()),
                )),
                ActionConfig::Llm(llm) => arc_app(crate::proxy_config::llm::build(llm)),
                ActionConfig::Unknown(_) => {
                    // No-op: these fall through to fallback
                    arc_app(crate::proxy_config::NullApp)
                }
            };

        // Wrap the base handler with middleware layers
        let handler = apply_middleware(base_handler, &route.middleware);

        compiled.push(CompiledRoute { matcher, handler, async_proxy });
    }
    compiled
}

/// Start the `[[tcp_proxy]]` / `[[udp_proxy]]` / `[[ws_proxy]]` listeners.
/// Startup only: a hot reload can't move bound sockets, so changes to these
/// sections need a restart.
fn spawn_listeners(config: &ProxyConfig) -> Vec<std::thread::JoinHandle<()>> {
    let mut handles: Vec<std::thread::JoinHandle<()>> = Vec::new();

    for tcp_cfg in &config.tcp_proxies {
        let listen = tcp_cfg.listen.clone();
        let backends = tcp_cfg.backends.clone();
        let timeout_ms = tcp_cfg.connect_timeout_ms;
        let name = tcp_cfg.name.clone();
        let h = std::thread::Builder::new()
            .name(format!("tcp-proxy-{}", name))
            .spawn(move || {
                let proxy = crate::tcp_proxy::TcpProxy::new(backends)
                    .connect_timeout_ms(timeout_ms);
                if let Err(e) = proxy.bind(&listen) {
                    eprintln!("[tcp_proxy:{}] {}", name, e);
                }
            })
            .expect("failed to spawn tcp proxy thread");
        handles.push(h);
    }

    for udp_cfg in &config.udp_proxies {
        let listen = udp_cfg.listen.clone();
        let backends = udp_cfg.backends.clone();
        let reply_timeout_ms = udp_cfg.reply_timeout_ms;
        let buffer_size = udp_cfg.buffer_size;
        let name = udp_cfg.name.clone();
        let h = std::thread::Builder::new()
            .name(format!("udp-proxy-{}", name))
            .spawn(move || {
                let proxy = crate::udp_proxy::UdpProxy::new(backends)
                    .reply_timeout_ms(reply_timeout_ms)
                    .buffer_size(buffer_size);
                if let Err(e) = proxy.bind(&listen) {
                    eprintln!("[udp_proxy:{}] {}", name, e);
                }
            })
            .expect("failed to spawn udp proxy thread");
        handles.push(h);
    }

    for ws_cfg in &config.ws_proxies {
        let listen = ws_cfg.listen.clone();
        let backends = ws_cfg.backends.clone();
        let connect_timeout_ms = ws_cfg.connect_timeout_ms;
        let read_timeout_ms = ws_cfg.read_timeout_ms;
        let name = ws_cfg.name.clone();

        // With a health check configured, share a live-backend list with a
        // background checker thread (same mechanism `[[upstream]]` pools
        // use) instead of treating every configured backend as always live.
        let live = Arc::new(RwLock::new(backends.clone()));
        if let Some(ref hc) = ws_cfg.health_check {
            crate::proxy_config::health::start_health_checker(
                name.clone(),
                backends.clone(),
                Arc::clone(&live),
                hc.clone(),
                Arc::new(AtomicBool::new(false)),
                None,
            );
        }

        let h = std::thread::Builder::new()
            .name(format!("ws-proxy-{}", name))
            .spawn(move || {
                let proxy = crate::ws_proxy::WsProxy::with_live_backends(backends, live)
                    .connect_timeout_ms(connect_timeout_ms)
                    .read_timeout_ms(read_timeout_ms);
                if let Err(e) = proxy.bind(&listen) {
                    eprintln!("[ws_proxy:{}] {}", name, e);
                }
            })
            .expect("failed to spawn ws proxy thread");
        handles.push(h);
    }

    handles
}

/// Wrap `handler` with the middleware layers described by `mw`, innermost
/// (closest to the handler) to outermost (runs first, on every request):
///
///   1. Cache (innermost)
///   2. Rewrite (request + response rules combined)
///   3. Auth
///   4. Rate limit
///   5. IP filter
///   6. Timeout — bounds this route's total time, including every layer above
///   7. Max body size (outermost) — a fast, non-blocking length check, so a
///      request this route will reject outright doesn't pay for a timeout
///      wrapper (which runs its target on a background thread) or engage
///      any other layer at all
pub(crate) fn apply_middleware(
    handler: Arc<dyn crate::application::Application + Send + Sync>,
    mw: &MiddlewareConfig,
) -> Arc<dyn crate::application::Application + Send + Sync> {
    // Wrap the Arc<dyn Application> in an ArcApp adapter so it implements
    // Application itself (needed for WithMiddleware::new).
    let mut app: Arc<dyn crate::application::Application + Send + Sync> = handler;

    // ── Cache (applied first so it's the innermost middleware) ─────────────
    if let Some(ref cache_cfg) = mw.cache {
        let mut layer = crate::cache::CacheLayer::memory(1000).ttl(cache_cfg.ttl_secs);
        for vh in &cache_cfg.vary_by {
            layer = layer.vary_by_header(vh.as_str());
        }
        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
    }

    // ── Rewrite ────────────────────────────────────────────────────────────
    if !mw.rewrite_request.is_empty() || !mw.rewrite_response.is_empty() {
        let mut layer = crate::rewrite::RewriteLayer::new();
        for rule in &mw.rewrite_request {
            layer = apply_request_rewrite_rule(layer, rule);
        }
        for rule in &mw.rewrite_response {
            layer = apply_response_rewrite_rule(layer, rule);
        }
        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer));
    }

    // ── Auth ───────────────────────────────────────────────────────────────
    // Fails closed: an auth setting that can't be honoured (secret unset,
    // htpasswd missing, feature not compiled in) puts AuthMisconfigured in
    // front of the route instead of silently serving it unauthenticated.
    if let Some(ref auth_cfg) = mw.auth {
        let deny = |reason: String| -> Arc<dyn crate::application::Application + Send + Sync> {
            eprintln!(
                "[proxy_config] ERROR: {reason} — this route answers 503 until the auth configuration is fixed."
            );
            arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(AuthMisconfigured))
        };
        app = match auth_cfg {
            AuthConfig::Bearer { token_env } => {
                let token = std::env::var(token_env).unwrap_or_default();
                if token.is_empty() {
                    deny(format!("bearer auth: env var '{token_env}' is unset or empty"))
                } else {
                    arc_app(
                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
                            .wrap(BearerAuthMiddleware { token: Arc::new(token) }),
                    )
                }
            }
            #[cfg(feature = "auth")]
            AuthConfig::Jwt { secret_env } => {
                let secret = std::env::var(secret_env).unwrap_or_default();
                if secret.is_empty() {
                    deny(format!("JWT auth: env var '{secret_env}' is unset or empty"))
                } else {
                    arc_app(
                        WithMiddleware::new(ArcApp(Arc::clone(&app)))
                            .wrap(crate::auth::JwtLayer::new(secret)),
                    )
                }
            }
            #[cfg(not(feature = "auth"))]
            AuthConfig::Jwt { .. } => {
                deny("JWT auth requires the 'auth' feature, which this build doesn't include".to_string())
            }
            #[cfg(feature = "auth")]
            AuthConfig::Basic { htpasswd_file } => {
                match crate::auth::BasicAuthLayer::from_htpasswd_file(htpasswd_file) {
                    Ok(layer) => arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(layer)),
                    Err(e) => deny(format!("basic auth: {e}")),
                }
            }
            #[cfg(not(feature = "auth"))]
            AuthConfig::Basic { .. } => {
                deny("basic auth requires the 'auth' feature, which this build doesn't include".to_string())
            }
        };
    }

    // ── Rate limit ─────────────────────────────────────────────────────────
    if let Some(ref rl_cfg) = mw.rate_limit {
        let limiter = Arc::new(crate::rate_limit::RateLimiter::new(
            rl_cfg.max_requests,
            rl_cfg.window_secs,
        ));
        app = arc_app(
            WithMiddleware::new(ArcApp(Arc::clone(&app)))
                .wrap(PerRouteRateLimit(limiter)),
        );
    }

    // ── IP filter ────────────────────────────────────────────────────────────
    if !mw.ip_allow.is_empty() {
        let filter = crate::ip_filter::IpFilter::allow(mw.ip_allow.iter().map(|s| s.as_str()));
        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
    } else if !mw.ip_deny.is_empty() {
        let filter = crate::ip_filter::IpFilter::deny(mw.ip_deny.iter().map(|s| s.as_str()));
        app = arc_app(WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(filter));
    }

    // ── Timeout (bounds this route's total time, including every other
    // middleware above) ─────────────────────────────────────────────────────
    if let Some(timeout_ms) = mw.timeout_ms {
        app = arc_app(crate::timeout::TimeoutLayer::from_arc(
            app,
            std::time::Duration::from_millis(timeout_ms),
        ));
    }

    // ── Max body size (outermost — see PerRouteMaxBodySize's docs for why
    // this is not a substitute for RWS_CONFIG_MAX_BODY_SIZE_IN_BYTES) ──────
    if let Some(max_body_size) = mw.max_body_size {
        app = arc_app(
            WithMiddleware::new(ArcApp(Arc::clone(&app))).wrap(PerRouteMaxBodySize(max_body_size)),
        );
    }

    app
}

fn apply_request_rewrite_rule(
    layer: crate::rewrite::RewriteLayer,
    rule: &crate::proxy_config::RewriteRuleConfig,
) -> crate::rewrite::RewriteLayer {
    match rule.type_.as_str() {
        "header_set" => {
            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
                return layer.request_header_set(name, value);
            }
        }
        "header_remove" => {
            if let Some(name) = &rule.name {
                return layer.request_header_remove(name);
            }
        }
        "uri_set" => {
            if let Some(value) = &rule.value {
                return layer.request_uri_set(value);
            }
        }
        "uri_strip_prefix" | "strip_prefix" => {
            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
                return layer.request_uri_strip_prefix(prefix);
            }
        }
        "uri_add_prefix" | "add_prefix" => {
            if let Some(prefix) = rule.prefix.as_ref().or(rule.value.as_ref()) {
                return layer.request_uri_add_prefix(prefix);
            }
        }
        _ => {}
    }
    layer
}

fn apply_response_rewrite_rule(
    layer: crate::rewrite::RewriteLayer,
    rule: &crate::proxy_config::RewriteRuleConfig,
) -> crate::rewrite::RewriteLayer {
    match rule.type_.as_str() {
        "header_set" => {
            if let (Some(name), Some(value)) = (&rule.name, &rule.value) {
                return layer.response_header_set(name, value);
            }
        }
        "header_remove" => {
            if let Some(name) = &rule.name {
                return layer.response_header_remove(name);
            }
        }
        "status" => {
            if let (Some(code), Some(reason)) = (&rule.code, &rule.reason) {
                return layer.response_status(*code as i16, reason);
            }
        }
        "body_replace" => {
            if let (Some(from), Some(to)) = (&rule.from, &rule.to) {
                return layer.response_body_replace(from, to);
            }
        }
        _ => {}
    }
    layer
}

// ── ArcApp adapter ─────────────────────────────────────────────────────────────

/// Wraps `Arc<dyn Application + Send + Sync>` to implement `Application`
/// (needed because you can't implement a foreign trait on a foreign type).
struct ArcApp(Arc<dyn crate::application::Application + Send + Sync>);

impl crate::application::Application for ArcApp {
    fn execute(
        &self,
        request: &crate::request::Request,
        conn: &crate::server::ConnectionInfo,
    ) -> Result<crate::response::Response, String> {
        self.0.execute(request, conn)
    }

    fn streams_body(&self, request: &crate::request::Request, conn: &crate::server::ConnectionInfo) -> bool {
        self.0.streams_body(request, conn)
    }
}

impl Clone for ArcApp {
    fn clone(&self) -> Self {
        ArcApp(Arc::clone(&self.0))
    }
}