core/server/src/http.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Shard-0 HTTP/REST listener. This root binds the listener and assembles the
//! router; the rest is split across submodules: the `state` bridge and axum
//! `State`, the bearer `extractor`, the `jwt` issuer and its `jwks` resolver,
//! the route `handlers`, the `reads` gates, the `submit` write paths, `wire`
//! request mapping, partition-write `admission`, committed-reply `reply`
//! decoding, the rejection `error` types, and per-credential `session` state.

mod admission;
mod error;
mod extractor;
mod forward;
mod handlers;
mod jwks;
mod jwt;
mod metrics;
mod reads;
mod reply;
mod session;
mod state;
mod submit;
mod tls;
#[cfg(feature = "iggy-web")]
mod web;
mod wire;

use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::fmt::Display;
use std::net::SocketAddr;
use std::rc::Rc;
use std::str::FromStr;
use std::sync::Arc;

use axum::Router;
use axum::extract::connect_info::Connected;
use axum::extract::{DefaultBodyLimit, Request};
use axum::http::{HeaderName, HeaderValue, Method, StatusCode, Version, header::CONNECTION};
use axum::middleware::{Next, from_fn, from_fn_with_state};
use axum::response::Response;
use axum::routing::{delete, get, post, put};
use compio::net::TcpListener;
use configs::cluster::{ClusterConfig, http_forwarding_key_material};
use configs::http::{HttpConfig, HttpCorsConfig};
use configs::server::ServerSystemConfig;
use iggy_common::IggyError;
use message_bus::client_listener;
use send_wrapper::SendWrapper;
use tower_http::cors::{AllowOrigin, CorsLayer};
use tracing::{error, info, warn};

use crate::cluster_meta::ClusterRoster;
use crate::http::handlers::{
    change_password, create_cg, create_partitions, create_pat, create_stream, create_topic,
    create_user, delete_cg, delete_consumer_offset, delete_partitions, delete_pat, delete_segments,
    delete_stream, delete_topic, delete_user, describe_options, get_cg, get_cgs, get_client,
    get_clients, get_cluster_metadata, get_consumer_offset, get_pats, get_snapshot, get_stats,
    get_stream, get_streams, get_topic, get_topics, get_user, get_users, login_user,
    login_with_personal_access_token, logout_user, ping, poll_messages, purge_stream, purge_topic,
    refresh_token, send_messages, store_consumer_offset, update_permissions, update_stream,
    update_topic, update_user,
};
use crate::http::jwt::JwtManager;
use crate::http::session::RegistrationBarrier;
use crate::http::state::{ForwardState, HttpInner, HttpState, insert_view_header};
use crate::server_error::ServerError;
use crate::shell::ServerShard;

/// The shard-0 HTTP listener before its socket exists: the address to bind
/// and the config products the serve loop needs. Built first at boot, so a
/// bad `[http.*]` section fails before any listener accepts; [`Self::bind`]
/// opens the socket.
pub struct PreparedHttp {
    addr: SocketAddr,
    jwt: JwtManager,
    forward: ForwardState,
    cors: Option<CorsLayer>,
    metrics_endpoint: Option<String>,
    max_request_size: usize,
}

/// The shard-0 HTTP listener between bind and serve, held apart so the bound
/// port can be published (cluster roster, `current_config.toml`) before any
/// request is answered.
pub struct BoundHttp {
    listener: TcpListener,
    pub bound_addr: SocketAddr,
    prepared: PreparedHttp,
}

/// Validate the HTTP config and build what the serve loop needs, without
/// opening the socket; [`PreparedHttp::bind`] binds it and [`start`] serves
/// it. The caller gates this to shard 0 and to `http.enabled`.
///
/// # Errors
///
/// Returns [`ServerError`] if the JWT manager cannot be built from
/// `http_config.jwt`, the `[http.cors]` config is invalid, the `[http.tls]`
/// forwarding credentials cannot be loaded, or the `[http.metrics]` endpoint
/// is not a route path.
pub fn prepare(
    addr: SocketAddr,
    http_config: &HttpConfig,
    cluster: &ClusterConfig,
) -> Result<PreparedHttp, ServerError> {
    // In cluster mode with no configured JWT secret the signing key derives
    // from the cluster PSK, so a bearer minted on any node verifies on every
    // node - the invariant follower-to-primary forwarding depends on.
    let cluster_psk =
        (cluster.enabled && cluster.auth.enabled && !cluster.auth.shared_secret.is_empty())
            .then_some(cluster.auth.shared_secret.as_str());
    let jwt = JwtManager::build(&http_config.jwt, cluster_psk)?;
    // Forwarding needs a bearer every node can verify; without key material it
    // degrades to off (followers answer the transient 503) instead of failing
    // the boot, so keyless clusters still serve HTTP node-locally.
    let forwarding_active = http_forwarding_key_material(&http_config.jwt, cluster);
    if cluster.enabled && !forwarding_active {
        warn!(
            "cluster mode with http enabled but no http.jwt secrets and no cluster.auth: bearers are node-local and follower-to-primary forwarding is disabled - control-plane writes on followers answer a transient 503; configure http.jwt encoding/decoding secrets or enable cluster.auth (identical on every node) to activate forwarding"
        );
    }
    // Saturating: a configured limit past the pointer width (32-bit target,
    // >4 GiB value) clamps to the largest enforceable cap instead of wrapping.
    let max_request_size =
        usize::try_from(http_config.max_request_size.as_bytes_u64()).unwrap_or(usize::MAX);
    let forward =
        forward::build_forward_state(&http_config.tls, max_request_size, forwarding_active)?;
    // Validated here, before the socket opens, so a bad [http.cors] fails boot
    // before the "started" log prints.
    let cors = http_config
        .cors
        .enabled
        .then(|| configure_cors(&http_config.cors))
        .transpose()?;
    // Same early-fail rule for the scrape path: axum panics on a route
    // without a leading '/', so reject it as a config error instead.
    let metrics_endpoint = metrics::validated_endpoint(&http_config.metrics)?;
    Ok(PreparedHttp {
        addr,
        jwt,
        forward,
        cors,
        metrics_endpoint,
        max_request_size,
    })
}

impl PreparedHttp {
    /// Open the socket. Kept apart from [`prepare`] so the caller can order it
    /// after any boot step that may block, and the port never sits listening
    /// with nothing serving it.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError`] if the listener cannot bind to the prepared
    /// address.
    pub fn bind(self) -> Result<BoundHttp, ServerError> {
        let (listener, bound_addr) = client_listener::tcp::bind(self.addr).map_err(|source| {
            error!(addr = %self.addr, error = %source, "failed to bind HTTP listener");
            source
        })?;
        Ok(BoundHttp {
            listener,
            bound_addr,
            prepared: self,
        })
    }
}

/// Assemble the router over a bound listener and spawn the `cyper-axum`
/// serve loop as a background task on shard 0's compio runtime. Serves HTTPS
/// when `http.tls.enabled` (a TLS accept pump feeds handshaken streams to the
/// serve loop, see [`mod@tls`]), plain HTTP otherwise; the listener stops
/// when the bus shutdown token fires.
///
/// `roster` is shard 0's own [`ClusterRoster`], so the HTTP and binary
/// cluster-metadata reads serve one truth.
///
/// # Errors
///
/// Returns [`ServerError`] if the `[http.tls]` server credentials cannot be
/// loaded.
#[allow(clippy::too_many_arguments)]
pub fn start(
    bound: BoundHttp,
    shard: &Rc<ServerShard>,
    http_config: &HttpConfig,
    clients_table_max: usize,
    max_tokens_per_user: u32,
    system_config: Arc<ServerSystemConfig>,
    roster: Rc<ClusterRoster>,
    shard_metrics_all: &[shard::metrics::ShardMetrics],
) -> Result<(), ServerError> {
    let BoundHttp {
        listener,
        bound_addr,
        prepared:
            PreparedHttp {
                jwt,
                forward,
                cors,
                metrics_endpoint,
                max_request_size,
                ..
            },
    } = bound;
    let state: HttpState = SendWrapper::new(Rc::new(HttpInner {
        shard: Rc::clone(shard),
        jwt,
        system_config,
        sessions: RefCell::new(HashMap::new()),
        registrations: RegistrationBarrier::default(),
        roster,
        max_http_sessions: crate::http::session::max_http_sessions(clients_table_max),
        max_tokens_per_user,
        in_flight_writes: Cell::new(0),
        forward,
        metrics: metrics::HttpMetrics::init(shard_metrics_all),
    }));
    let app = router(
        state,
        max_request_size,
        cors,
        metrics_endpoint.as_deref(),
        http_config.web_ui,
    );

    if http_config.tls.enabled {
        let server_config = tls::load_http_tls_server_config(&http_config.tls)?;
        let (connections, pump) = tls::spawn_accept_pump(
            listener,
            server_config,
            shard.bus.config().handshake_grace,
            shard.bus.token(),
        );
        shard.bus.track_background(pump);
        info!(address = %bound_addr, "server HTTPS listener started");
        let handle = compio::runtime::spawn(tls::serve(connections, app, shard.bus.token()));
        shard.bus.track_background(handle);
    } else {
        info!(address = %bound_addr, "server HTTP listener started");
        let shutdown = shard.bus.token();
        let handle = compio::runtime::spawn(async move {
            if let Err(error) = cyper_axum::serve(
                listener,
                app.into_make_service_with_connect_info::<ClientAddr>(),
            )
            .with_graceful_shutdown(async move { shutdown.wait().await })
            .await
            {
                error!(%error, "server HTTP listener terminated with error");
            }
        });
        shard.bus.track_background(handle);
    }

    Ok(())
}

/// Connect-info payload carrying the peer socket address of an HTTP client.
///
/// The plain listener records it through axum's connect-info make-service
/// (the [`Connected`] impl below); the TLS path cannot (its hand-rolled serve
/// loop bypasses the make-service), so `tls::serve` injects the identical
/// `ConnectInfo<ClientAddr>` extension per connection instead. Handlers read
/// it through the `Identity` extractor's `client_ip`, which picks the
/// advertised address a client is told about - never authorization.
#[derive(Debug, Clone, Copy)]
pub struct ClientAddr(pub SocketAddr);

impl Connected<cyper_axum::IncomingStream<'_, TcpListener>> for ClientAddr {
    fn connect_info(stream: cyper_axum::IncomingStream<'_, TcpListener>) -> Self {
        Self(*stream.remote_addr())
    }
}

/// Health-probe path. Public and pre-auth, and the one success route reached
/// without proving a credential, so the `Iggy-View` layer withholds the
/// cluster-internal view number here (see the response layer below).
const PING_PATH: &str = "/ping";

/// Assemble the shard-0 router: unauthenticated health + login routes plus the
/// authenticated REST surface.
///
/// The surface is split by consensus dependency. The control-plane routes -
/// whose writes all commit through the metadata consensus group - carry the
/// `forward_to_primary` route layer, so on a follower they are relayed to the
/// primary instead of failing with a transient 503 (reads under that layer
/// still serve locally unless `?consistency=linearizable`). The local routes
/// never need the metadata primary: health, the login/refresh flows (STM read
/// + JWT mint), node-local reads, and the partition-plane routes.
///
/// `max_request_size` becomes the router-wide `DefaultBodyLimit` (413 past
/// it), exactly like the legacy server: it bounds the per-request term of the
/// admission math - what one body may cost in bytes and decode CPU - while
/// the in-flight caps bound the multiplier. Those 413s reject the body unread,
/// so they are stamped `Connection: close` (see below).
///
/// `cors`, present only when `[http.cors]` is enabled, is applied as the
/// outermost layer. This node authenticates per route (the `Authenticated` /
/// `Identity` extractors), so a preflight `OPTIONS` - which carries no
/// `Authorization` header and matches none of the method routes - would 401 or
/// 405 if it reached the router; the outermost `CorsLayer` answers it first
/// instead, and stamps the CORS response headers over every reply, including
/// the inner layer's `iggy-view`.
///
/// `metrics_endpoint`, present only when `[http.metrics]` is enabled, mounts
/// the auth-only scrape route among the local routes (a scrape must describe
/// the serving node, never a forwarded primary) and switches on the
/// request-counting layer.
fn router(
    state: HttpState,
    max_request_size: usize,
    cors: Option<CorsLayer>,
    metrics_endpoint: Option<&str>,
    web_ui: bool,
) -> Router {
    // Cloned for the response layer so `Iggy-View` reads the live view per
    // response; the original `state` is moved into `with_state` below.
    let view_source = state.clone();
    // Counter handle taken before `state` moves below; the counting layer
    // shares it with the registry the scrape handler encodes.
    let http_requests = metrics_endpoint
        .is_some()
        .then(|| state.metrics.request_counter());
    let forwardable = forwardable_routes(state.clone());
    let partition_writes = partition_write_routes(state.clone());
    let local = Router::new()
        .route(PING_PATH, get(ping))
        .route("/users/login", post(login_user))
        .route("/users/refresh-token", post(refresh_token))
        .route(
            "/personal-access-tokens/login",
            post(login_with_personal_access_token),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/messages",
            get(poll_messages),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
            get(get_consumer_offset),
        )
        .route("/stats", get(get_stats))
        .route("/options/{scope}", get(describe_options))
        .route("/snapshot", post(get_snapshot))
        .route("/cluster/metadata", get(get_cluster_metadata))
        .route("/clients", get(get_clients))
        .route("/clients/{client_id}", get(get_client));
    let local = match metrics_endpoint {
        Some(endpoint) => local.route(endpoint, get(handlers::get_metrics)),
        None => local,
    };
    let router = Router::new()
        .merge(forwardable)
        .merge(partition_writes)
        .merge(local)
        .with_state(state)
        .layer(DefaultBodyLimit::max(max_request_size))
        .layer(from_fn(move |request: Request, next: Next| {
            let view_source = view_source.clone();
            // `/ping` is the one success route reached without proving a
            // credential, so it must not leak the cluster-internal view number
            // (the anon-leak gate). Every other route - the metrics scrape
            // included - authenticates before its handler, so a success or
            // redirect there is an authed flow that may carry the header; the
            // login routes prove credentials on success.
            let suppress_view = request.uri().path() == PING_PATH;
            async move {
                let response = next.run(request).await;
                if suppress_view {
                    response
                } else {
                    insert_view_header(&view_source, response)
                }
            }
        }))
        .layer(from_fn(close_on_payload_too_large));
    let router = match cors {
        Some(cors) => router.layer(cors),
        None => router,
    };
    // Outermost, mirroring the legacy server's wiring: every request is
    // counted, including CORS preflights answered by the layer beneath.
    let router = match http_requests {
        Some(http_requests) => router.layer(from_fn(move |request: Request, next: Next| {
            http_requests.inc();
            async move { next.run(request).await }
        })),
        None => router,
    };

    // `with_state(())` finalizes every route eagerly, once for the whole
    // listener - including the web-ui routes merged after the stateful
    // `with_state` above, which would otherwise stay boxed handlers that
    // axum rebuilds per request. Identity on already-finalized routes, so
    // both the plain and the TLS serve paths share the finalized form.
    merge_web_ui(router, web_ui).with_state(())
}

/// Acknowledged partition writes use a bounded HTTP roster fallback. This is
/// a correctness path for the existing stateless HTTP transport. Direct
/// partition-primary routing remains the scalable long-term design.
fn partition_write_routes(state: HttpState) -> Router<HttpState> {
    Router::new()
        .route(
            "/streams/{stream_id}/topics/{topic_id}/messages",
            post(send_messages),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/consumer-offsets",
            put(store_consumer_offset),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/consumer-offsets/{consumer_id}",
            delete(delete_consumer_offset),
        )
        .route_layer(from_fn_with_state(state, forward::forward_partition_write))
}

/// The control-plane route table: every write here commits through the
/// metadata consensus group, so the shared `forward_to_primary` route layer
/// (holding `state`) relays them on a follower.
fn forwardable_routes(state: HttpState) -> Router<HttpState> {
    Router::new()
        .route("/users", get(get_users).post(create_user))
        .route(
            "/users/{user_id}",
            get(get_user).put(update_user).delete(delete_user),
        )
        .route("/users/{user_id}/password", put(change_password))
        .route("/users/{user_id}/permissions", put(update_permissions))
        // Static `logout` outranks the `{user_id}` capture in axum's matcher, so
        // `DELETE /users/logout` never misroutes to `delete_user`.
        .route("/users/logout", delete(logout_user))
        .route("/streams", get(get_streams).post(create_stream))
        .route(
            "/streams/{stream_id}",
            get(get_stream).put(update_stream).delete(delete_stream),
        )
        .route("/streams/{stream_id}/purge", delete(purge_stream))
        .route(
            "/streams/{stream_id}/topics",
            get(get_topics).post(create_topic),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}",
            get(get_topic).put(update_topic).delete(delete_topic),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/purge",
            delete(purge_topic),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/partitions",
            post(create_partitions).delete(delete_partitions),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}",
            delete(delete_segments),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/consumer-groups",
            get(get_cgs).post(create_cg),
        )
        .route(
            "/streams/{stream_id}/topics/{topic_id}/consumer-groups/{group_id}",
            get(get_cg).delete(delete_cg),
        )
        .route("/personal-access-tokens", get(get_pats).post(create_pat))
        .route("/personal-access-tokens/{name}", delete(delete_pat))
        .route_layer(from_fn_with_state(state, forward::forward_to_primary))
}

/// Stamp `Connection: close` on a 413, which rejects its request body unread.
///
/// hyper decides to close only once it notices the dropped body, and that is
/// after the response head is already encoded: the reply advertises keep-alive,
/// the peer pools a connection the server has stopped reading, and the next
/// request sent on it dies on a clean EOF. A caller-set `Connection` is written
/// through untouched, so setting it here has the peer retire the connection.
///
/// HTTP/2 carries no connection-level headers, so h2 replies are left alone
/// rather than handed a header hyper would strip and warn about.
async fn close_on_payload_too_large(request: Request, next: Next) -> Response {
    let http2 = request.version() == Version::HTTP_2;
    let mut response = next.run(request).await;
    if !http2 && response.status() == StatusCode::PAYLOAD_TOO_LARGE {
        response
            .headers_mut()
            .insert(CONNECTION, HeaderValue::from_static("close"));
    }
    response
}

/// Merge the unauthenticated `/ui` static-asset surface when `web_ui` is set.
///
/// The caller merges this outermost - after `with_state`, the body limit, the
/// view-header layer, and CORS - mirroring the legacy server. Staying past the
/// view-header layer also keeps the cluster-internal view number off these
/// unauthenticated responses, the same anon-leak gate `/ping` gets above.
///
/// Without the `iggy-web` feature the assets are not compiled in, so an enabled
/// flag only warns instead of serving.
fn merge_web_ui(router: Router, web_ui: bool) -> Router {
    #[cfg(feature = "iggy-web")]
    let router = if web_ui {
        info!("Web UI enabled at /ui");
        router.merge(web::router())
    } else {
        router
    };

    #[cfg(not(feature = "iggy-web"))]
    if web_ui {
        tracing::warn!(
            "Web UI is enabled in configuration (http.web_ui = true) but the server \
             was not compiled with 'iggy-web' feature. The Web UI will not be available. \
             To enable it, rebuild the server with: cargo build --features iggy-web"
        );
    }

    router
}

/// Build the [`CorsLayer`] from `[http.cors]`, porting the legacy server's
/// mapping: an empty origin list yields the tower-http default, a leading `*`
/// allows any origin (later entries are ignored), and anything else is an
/// explicit allow-list. Entries are trimmed and blank ones dropped, so a
/// placeholder like `[""]` maps to "none" rather than a parse error. Methods,
/// allowed headers, and exposed headers map the same way; any unparsable value
/// fails the build.
///
/// Combinations tower-http would reject by panicking (a `*` origin past the
/// first position, or `allow_credentials` with a wildcard origin, header, or
/// exposed-header list) are rejected here as configuration errors instead.
///
/// # Errors
///
/// Returns [`IggyError::InvalidConfiguration`] if an origin or header value is
/// not a valid header token, a method is not one of the standard HTTP verbs,
/// `*` appears past the first origin, or `allow_credentials` is combined with
/// a wildcard origin, header, or exposed-header list.
fn configure_cors(config: &HttpCorsConfig) -> Result<CorsLayer, IggyError> {
    let wildcard_origin = config
        .allowed_origins
        .first()
        .is_some_and(|origin| origin.trim() == "*");
    let allowed_origins = match config.allowed_origins.as_slice() {
        [] => AllowOrigin::default(),
        _ if wildcard_origin => AllowOrigin::any(),
        // `AllowOrigin::list` panics on a wildcard entry, so past the first
        // position `*` is a config mistake, not "any origin".
        origins if origins.iter().any(|origin| origin.trim() == "*") => {
            error!("invalid CORS allowed_origins: \"*\" is honored only as the first entry");
            return Err(IggyError::InvalidConfiguration);
        }
        origins => AllowOrigin::list(parse_cors_values::<HeaderValue>(origins, "origin")?),
    };
    let allowed_headers = parse_cors_values::<HeaderName>(&config.allowed_headers, "header")?;
    let exposed_headers =
        parse_cors_values::<HeaderName>(&config.exposed_headers, "exposed header")?;
    let allowed_methods = parse_cors_methods(&config.allowed_methods)?;

    // tower-http rejects credentials combined with any wildcard by panicking
    // once the layer is applied to the router; catch those combinations here
    // so they fail as configuration errors instead.
    if config.allow_credentials {
        let wildcard_field = if wildcard_origin {
            Some("allowed_origins")
        } else if is_wildcard_header_list(&allowed_headers) {
            Some("allowed_headers")
        } else if is_wildcard_header_list(&exposed_headers) {
            Some("exposed_headers")
        } else {
            None
        };
        if let Some(field) = wildcard_field {
            error!(
                "invalid CORS config: allow_credentials cannot be combined with wildcard {field}"
            );
            return Err(IggyError::InvalidConfiguration);
        }
    }

    Ok(CorsLayer::new()
        .allow_methods(allowed_methods)
        .allow_origin(allowed_origins)
        .allow_headers(allowed_headers)
        .expose_headers(exposed_headers)
        .allow_credentials(config.allow_credentials)
        .allow_private_network(config.allow_private_network))
}

/// Parse a CORS string list into header values or names, trimming entries and
/// skipping blank ones so a placeholder like `[""]` yields an empty set.
/// `label` names the field in the diagnostic log for an unparsable entry.
fn parse_cors_values<T>(values: &[String], label: &str) -> Result<Vec<T>, IggyError>
where
    T: FromStr,
    T::Err: Display,
{
    values
        .iter()
        .map(|value| value.trim())
        .filter(|value| !value.is_empty())
        .map(|value| {
            value.parse::<T>().map_err(|error| {
                error!(%value, %error, "invalid CORS {label}");
                IggyError::InvalidConfiguration
            })
        })
        .collect()
}

/// Map the configured method names onto the standard HTTP verbs, trimming
/// entries and skipping blank ones. A name outside the standard set is rejected
/// (rather than accepted as a custom method token) so a typo fails the config
/// loudly.
fn parse_cors_methods(methods: &[String]) -> Result<Vec<Method>, IggyError> {
    methods
        .iter()
        .map(|method| method.trim())
        .filter(|method| !method.is_empty())
        .map(|method| match method.to_uppercase().as_str() {
            "GET" => Ok(Method::GET),
            "POST" => Ok(Method::POST),
            "PUT" => Ok(Method::PUT),
            "DELETE" => Ok(Method::DELETE),
            "HEAD" => Ok(Method::HEAD),
            "OPTIONS" => Ok(Method::OPTIONS),
            "CONNECT" => Ok(Method::CONNECT),
            "PATCH" => Ok(Method::PATCH),
            "TRACE" => Ok(Method::TRACE),
            other => {
                error!(method = %other, "invalid CORS method");
                Err(IggyError::InvalidConfiguration)
            }
        })
        .collect()
}

/// The exact shape tower-http treats as a wildcard header set: a lone `*`.
/// (`["*", "x"]` joins to the literal value `*,x`, which is not a wildcard.)
fn is_wildcard_header_list(headers: &[HeaderName]) -> bool {
    matches!(headers, [only] if only.as_str() == "*")
}

#[cfg(test)]
mod tests {
    use super::*;

    fn cors_config() -> HttpCorsConfig {
        HttpCorsConfig {
            enabled: true,
            allowed_methods: vec!["GET".to_owned(), "POST".to_owned()],
            allowed_origins: vec!["*".to_owned()],
            allowed_headers: vec!["content-type".to_owned(), "authorization".to_owned()],
            exposed_headers: vec!["iggy-view".to_owned()],
            allow_credentials: false,
            allow_private_network: false,
        }
    }

    #[test]
    fn configure_cors_accepts_wildcard_origin() {
        assert!(configure_cors(&cors_config()).is_ok());
    }

    #[test]
    fn configure_cors_accepts_explicit_origins() {
        let config = HttpCorsConfig {
            allowed_origins: vec![
                "http://localhost:3000".to_owned(),
                "https://app.example.com".to_owned(),
            ],
            ..cors_config()
        };
        assert!(configure_cors(&config).is_ok());
    }

    #[test]
    fn configure_cors_accepts_credentials_with_explicit_origin() {
        let config = HttpCorsConfig {
            allowed_origins: vec!["https://app.example.com".to_owned()],
            allow_credentials: true,
            ..cors_config()
        };
        assert!(configure_cors(&config).is_ok());
    }

    #[test]
    fn configure_cors_skips_blank_entries() {
        // The shipped placeholder shape: a single empty string must map to an
        // empty set, not a parse error.
        let config = HttpCorsConfig {
            allowed_origins: vec!["https://app.example.com".to_owned()],
            exposed_headers: vec![String::new()],
            ..cors_config()
        };
        assert!(configure_cors(&config).is_ok());
    }

    #[test]
    fn configure_cors_accepts_wildcard_headers_without_credentials() {
        let config = HttpCorsConfig {
            allowed_headers: vec!["*".to_owned()],
            exposed_headers: vec!["*".to_owned()],
            ..cors_config()
        };
        assert!(configure_cors(&config).is_ok());
    }

    #[test]
    fn configure_cors_trims_entries() {
        let config = HttpCorsConfig {
            allowed_origins: vec![" https://app.example.com ".to_owned()],
            allowed_headers: vec![" content-type ".to_owned()],
            allowed_methods: vec![" get ".to_owned()],
            ..cors_config()
        };
        assert!(configure_cors(&config).is_ok());
    }

    #[test]
    fn configure_cors_rejects_wildcard_origin_after_first() {
        // tower-http's `AllowOrigin::list` panics on a wildcard entry; the
        // guard must turn it into a clean config error.
        let config = HttpCorsConfig {
            allowed_origins: vec!["https://app.example.com".to_owned(), "*".to_owned()],
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_credentials_with_wildcard_origin() {
        // tower-http's `ensure_usable_cors_rules` panics on this combination
        // when the layer is applied; the guard must turn it into a clean
        // config error.
        let config = HttpCorsConfig {
            allow_credentials: true,
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_credentials_with_wildcard_headers() {
        let config = HttpCorsConfig {
            allowed_origins: vec!["https://app.example.com".to_owned()],
            allowed_headers: vec!["*".to_owned()],
            allow_credentials: true,
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_credentials_with_wildcard_exposed_headers() {
        let config = HttpCorsConfig {
            allowed_origins: vec!["https://app.example.com".to_owned()],
            exposed_headers: vec!["*".to_owned()],
            allow_credentials: true,
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_invalid_origin() {
        let config = HttpCorsConfig {
            allowed_origins: vec!["http://bad\norigin".to_owned()],
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_invalid_header() {
        let config = HttpCorsConfig {
            allowed_headers: vec!["invalid header".to_owned()],
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }

    #[test]
    fn configure_cors_rejects_unknown_method() {
        let config = HttpCorsConfig {
            allowed_methods: vec!["FOOBAR".to_owned()],
            ..cors_config()
        };
        assert!(matches!(
            configure_cors(&config),
            Err(IggyError::InvalidConfiguration)
        ));
    }
}