lenso-web-ingress-plugin 0.4.12

use std::{collections::BTreeMap, net::SocketAddr, time::Duration};

use http::HeaderName;
use serde::{Deserialize, Serialize};

const MAX_TRANSFER_BYTES: usize = 64 * 1024 * 1024;
const MAX_HEAD_BYTES: usize = 1024 * 1024;
const MAX_CONCURRENT_REQUESTS: usize = 4_096;
const MAX_CONNECTIONS: usize = 65_536;
const MAX_TIMEOUT_MILLIS: u64 = 300_000;
const MAX_PROTOCOL_NAME_BYTES: usize = 128;

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct SelectedRouteSessionCookieConfig {
    route: crate::WebIngressRoute,
    policy: SessionCookieConfig,
}

/// Cookie-based session and double-submit CSRF policy selected by an Ingress Instance.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SessionCookieConfig {
    name: String,
    csrf_cookie_name: String,
    csrf_header_name: String,
}

impl SessionCookieConfig {
    /// Creates one cookie authentication policy with a fixed `session` credential scheme.
    pub fn new(
        name: impl Into<String>,
        csrf_cookie_name: impl Into<String>,
        csrf_header_name: impl Into<String>,
    ) -> Result<Self, String> {
        let value = Self {
            name: name.into(),
            csrf_cookie_name: csrf_cookie_name.into(),
            csrf_header_name: csrf_header_name.into(),
        };
        value.validate()?;
        Ok(value)
    }

    fn validate(&self) -> Result<(), String> {
        if !host_cookie_name(&self.name)
            || !host_cookie_name(&self.csrf_cookie_name)
            || self.name == self.csrf_cookie_name
        {
            return Err(
                "Web Ingress requires distinct __Host- session and CSRF cookie names".to_owned(),
            );
        }
        if self.csrf_header_name.is_empty() || self.csrf_header_name.len() > MAX_PROTOCOL_NAME_BYTES
        {
            return Err("Web Ingress CSRF header name is invalid".to_owned());
        }
        let header = HeaderName::from_bytes(self.csrf_header_name.as_bytes())
            .map_err(|_| "Web Ingress CSRF header name is invalid".to_owned())?;
        if [
            "authorization",
            "connection",
            "content-length",
            "cookie",
            "host",
            "keep-alive",
            "proxy-connection",
            "te",
            "trailer",
            "transfer-encoding",
            "upgrade",
            "x-request-id",
        ]
        .contains(&header.as_str())
        {
            return Err("Web Ingress CSRF header must be a dedicated request header".to_owned());
        }
        Ok(())
    }

    pub(crate) fn name(&self) -> &str {
        &self.name
    }

    pub(crate) fn csrf_cookie_name(&self) -> &str {
        &self.csrf_cookie_name
    }

    pub(crate) fn csrf_header_name(&self) -> &str {
        &self.csrf_header_name
    }
}

/// Explicit WebSocket transport policy. Browser origins are exact matches.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WebSocketConfig {
    allowed_origins: Vec<String>,
    max_message_bytes: usize,
    max_session_bytes: usize,
}
impl WebSocketConfig {
    pub fn new(allowed_origins: Vec<String>) -> Result<Self, String> {
        let policy = Self {
            allowed_origins,
            max_message_bytes: 65_536,
            max_session_bytes: 1_048_576,
        };
        policy.validate()?;
        Ok(policy)
    }
    pub fn with_limits(
        mut self,
        message_bytes: usize,
        session_bytes: usize,
    ) -> Result<Self, String> {
        self.max_message_bytes = message_bytes;
        self.max_session_bytes = session_bytes;
        self.validate()?;
        Ok(self)
    }
    fn validate(&self) -> Result<(), String> {
        if self.max_message_bytes == 0
            || self.max_message_bytes > 1_048_576
            || self.max_session_bytes < self.max_message_bytes
            || self.max_session_bytes > MAX_TRANSFER_BYTES
            || self.allowed_origins.len() > 64
        {
            return Err("Invalid WebSocket bounds".into());
        }
        for origin in &self.allowed_origins {
            let uri: http::Uri = origin.parse().map_err(|_| "Invalid WebSocket origin")?;
            if !matches!(uri.scheme_str(), Some("http" | "https"))
                || uri.authority().is_none()
                || uri
                    .authority()
                    .is_some_and(|authority| authority.as_str().contains('@'))
                || uri
                    .path_and_query()
                    .is_some_and(|path| path.as_str() != "/")
                || origin.ends_with('/')
                || origin.len() > 2048
            {
                return Err("WebSocket origins must be exact serialized HTTP origins".into());
            }
        }
        Ok(())
    }
    pub(crate) fn allowed_origins(&self) -> &[String] {
        &self.allowed_origins
    }
    pub const fn max_message_bytes(&self) -> usize {
        self.max_message_bytes
    }
    pub const fn max_session_bytes(&self) -> usize {
        self.max_session_bytes
    }
}

/// Immutable HTTP policy for one Web Ingress Plugin Instance.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct WebIngressConfig {
    #[serde(default = "default_bind_address")]
    bind_address: SocketAddr,
    #[serde(default = "default_max_request_body_bytes")]
    max_request_body_bytes: usize,
    #[serde(default = "default_max_request_head_bytes")]
    max_request_head_bytes: usize,
    #[serde(default = "default_max_concurrent_requests")]
    max_concurrent_requests: usize,
    #[serde(default = "default_max_connections")]
    max_connections: usize,
    #[serde(default = "default_request_head_timeout_millis")]
    request_head_timeout_millis: u64,
    #[serde(default = "default_request_body_timeout_millis")]
    request_body_timeout_millis: u64,
    #[serde(default = "default_connection_idle_timeout_millis")]
    connection_idle_timeout_millis: u64,
    #[serde(default = "default_shutdown_grace_timeout_millis")]
    shutdown_grace_timeout_millis: u64,
    #[serde(default = "default_request_timeout_millis")]
    request_timeout_millis: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    session_cookie: Option<SessionCookieConfig>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    route_session_cookies: BTreeMap<String, SessionCookieConfig>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    selected_route_session_cookies: Vec<SelectedRouteSessionCookieConfig>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    websocket: Option<WebSocketConfig>,
}

impl Default for WebIngressConfig {
    fn default() -> Self {
        Self {
            bind_address: default_bind_address(),
            max_request_body_bytes: default_max_request_body_bytes(),
            max_request_head_bytes: default_max_request_head_bytes(),
            max_concurrent_requests: default_max_concurrent_requests(),
            max_connections: default_max_connections(),
            request_head_timeout_millis: default_request_head_timeout_millis(),
            request_body_timeout_millis: default_request_body_timeout_millis(),
            connection_idle_timeout_millis: default_connection_idle_timeout_millis(),
            shutdown_grace_timeout_millis: default_shutdown_grace_timeout_millis(),
            request_timeout_millis: default_request_timeout_millis(),
            session_cookie: None,
            route_session_cookies: BTreeMap::new(),
            selected_route_session_cookies: Vec::new(),
            websocket: None,
        }
    }
}

impl WebIngressConfig {
    pub fn with_websocket(mut self, policy: WebSocketConfig) -> Result<Self, String> {
        policy.validate()?;
        self.websocket = Some(policy);
        Ok(self)
    }
    pub fn websocket(&self) -> Option<&WebSocketConfig> {
        self.websocket.as_ref()
    }

    /// Replaces the listener address. Loopback with port zero remains the default.
    pub fn with_bind_address(mut self, address: SocketAddr) -> Result<Self, String> {
        self.bind_address = address;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the maximum accepted request body and canonical request head sizes.
    pub fn with_request_limits(
        mut self,
        max_request_body_bytes: usize,
        max_request_head_bytes: usize,
    ) -> Result<Self, String> {
        self.max_request_body_bytes = max_request_body_bytes;
        self.max_request_head_bytes = max_request_head_bytes;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the number of HTTP requests admitted concurrently by this Instance.
    pub fn with_max_concurrent_requests(mut self, maximum: usize) -> Result<Self, String> {
        self.max_concurrent_requests = maximum;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the maximum live connections and keep-alive idle deadline.
    pub fn with_connection_limits(
        mut self,
        maximum: usize,
        idle_timeout: Duration,
    ) -> Result<Self, String> {
        self.max_connections = maximum;
        self.connection_idle_timeout_millis = duration_millis(idle_timeout)?;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the total request-head and request-body read deadlines.
    pub fn with_request_read_timeouts(
        mut self,
        head_timeout: Duration,
        body_timeout: Duration,
    ) -> Result<Self, String> {
        self.request_head_timeout_millis = duration_millis(head_timeout)?;
        self.request_body_timeout_millis = duration_millis(body_timeout)?;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the deadline applied to one Endpoint Capability invocation.
    pub fn with_request_timeout(mut self, timeout: Duration) -> Result<Self, String> {
        self.request_timeout_millis = duration_millis(timeout)?;
        self.validate()?;
        Ok(self)
    }

    /// Replaces the bounded drain window used after graceful App shutdown begins.
    pub fn with_shutdown_grace_timeout(mut self, timeout: Duration) -> Result<Self, String> {
        self.shutdown_grace_timeout_millis = duration_millis(timeout)?;
        self.validate()?;
        Ok(self)
    }

    /// Enables one named session Cookie and its double-submit CSRF policy.
    pub fn with_session_cookie(mut self, policy: SessionCookieConfig) -> Result<Self, String> {
        self.session_cookie = Some(policy);
        self.validate()?;
        Ok(self)
    }

    /// Selects a fixed Cookie and CSRF policy for a unique resolved Endpoint route ID.
    ///
    /// Routes without an entry use `session_cookie`, if configured. Selection happens before
    /// credential extraction and middleware; request headers and query parameters cannot select
    /// a policy. Unknown or ambiguous IDs fail activation. Use an exact route selector when
    /// multiple Endpoint Instances declare the same route ID.
    pub fn with_route_session_cookie(
        mut self,
        route_id: impl Into<String>,
        policy: SessionCookieConfig,
    ) -> Result<Self, String> {
        let route_id = route_id.into();
        if self.route_session_cookies.contains_key(&route_id) {
            return Err("Web Ingress route session Cookie policy is duplicated".to_owned());
        }
        self.route_session_cookies.insert(route_id, policy);
        self.validate()?;
        Ok(self)
    }

    /// Selects a policy by the exact method, declared path pattern and ID in the route manifest.
    ///
    /// This distinguishes routes from different Endpoint Instances that reuse a route ID.
    /// Unknown, duplicate or overlapping selectors fail before Ready. This is resolved route
    /// identity, not a second matcher against the request URI.
    pub fn with_selected_route_session_cookie(
        mut self,
        mut route: crate::WebIngressRoute,
        policy: SessionCookieConfig,
    ) -> Result<Self, String> {
        route.method = route.method.trim().to_ascii_uppercase();
        self.selected_route_session_cookies
            .push(SelectedRouteSessionCookieConfig { route, policy });
        self.validate()?;
        Ok(self)
    }

    pub(crate) fn validate(&self) -> Result<(), String> {
        if !(1..=MAX_TRANSFER_BYTES).contains(&self.max_request_body_bytes)
            || !(1..=MAX_HEAD_BYTES).contains(&self.max_request_head_bytes)
            || !(1..=MAX_CONCURRENT_REQUESTS).contains(&self.max_concurrent_requests)
            || !(1..=MAX_CONNECTIONS).contains(&self.max_connections)
            || !(1..=MAX_TIMEOUT_MILLIS).contains(&self.request_head_timeout_millis)
            || !(1..=MAX_TIMEOUT_MILLIS).contains(&self.request_body_timeout_millis)
            || !(1..=MAX_TIMEOUT_MILLIS).contains(&self.connection_idle_timeout_millis)
            || !(1..=MAX_TIMEOUT_MILLIS).contains(&self.shutdown_grace_timeout_millis)
            || !(1..=MAX_TIMEOUT_MILLIS).contains(&self.request_timeout_millis)
        {
            return Err("Web Ingress limits or timeout are invalid".to_owned());
        }
        if let Some(policy) = &self.websocket {
            policy.validate()?;
        }
        if let Some(policy) = &self.session_cookie {
            policy.validate()?;
        }
        for (route_id, policy) in &self.route_session_cookies {
            if route_id.trim().is_empty() {
                return Err(
                    "Web Ingress route session Cookie policy requires a route ID".to_owned(),
                );
            }
            policy.validate()?;
        }
        let mut selected = std::collections::BTreeSet::new();
        for entry in &self.selected_route_session_cookies {
            if entry.route.route_id.trim().is_empty()
                || http::Method::from_bytes(entry.route.method.as_bytes()).is_err()
                || entry.route.method != entry.route.method.trim().to_ascii_uppercase()
                || !entry.route.path.starts_with('/')
                || entry.route.path.contains(['?', '#'])
                || !selected.insert(&entry.route)
            {
                return Err(
                    "Web Ingress selected route Cookie policy is invalid or duplicated".to_owned(),
                );
            }
            entry.policy.validate()?;
        }
        Ok(())
    }

    #[cfg(feature = "native")]
    pub(crate) const fn bind_address(&self) -> SocketAddr {
        self.bind_address
    }

    /// Maximum bytes an HTTP host may collect before invoking Ingress.
    pub const fn max_request_body_bytes(&self) -> usize {
        self.max_request_body_bytes
    }

    /// Maximum canonical request-head size accepted by either transport.
    pub const fn max_request_head_bytes(&self) -> usize {
        self.max_request_head_bytes
    }

    pub(crate) const fn max_concurrent_requests(&self) -> usize {
        self.max_concurrent_requests
    }

    /// Returns the resolved-Plan limits Hosts should apply to each HTTP Endpoint binding.
    ///
    /// The returned copyable tuple is `(queue_capacity, max_concurrency)`. Ingress already queues
    /// at its own semaphore, so the Endpoint binding needs the same execution ceiling but no
    /// second queue. A Host may deliberately choose a stricter binding policy.
    pub const fn endpoint_admission_limits(&self) -> (usize, usize) {
        (0, self.max_concurrent_requests)
    }

    #[cfg(feature = "native")]
    pub(crate) const fn max_connections(&self) -> usize {
        self.max_connections
    }

    #[cfg(feature = "native")]
    pub(crate) fn request_head_timeout(&self) -> Duration {
        Duration::from_millis(self.request_head_timeout_millis)
    }

    /// Total body-read deadline the transport host must enforce.
    pub fn request_body_timeout(&self) -> Duration {
        Duration::from_millis(self.request_body_timeout_millis)
    }

    #[cfg(feature = "native")]
    pub(crate) fn connection_idle_timeout(&self) -> Duration {
        Duration::from_millis(self.connection_idle_timeout_millis)
    }

    pub(crate) fn request_timeout(&self) -> Duration {
        Duration::from_millis(self.request_timeout_millis)
    }

    #[cfg(feature = "native")]
    pub(crate) fn shutdown_grace_timeout(&self) -> Duration {
        Duration::from_millis(self.shutdown_grace_timeout_millis)
    }

    pub(crate) const fn session_cookie(&self) -> Option<&SessionCookieConfig> {
        self.session_cookie.as_ref()
    }

    pub(crate) fn resolved_route_session_cookies(
        &self,
        manifest: &crate::WebIngressRouteManifest,
    ) -> Result<BTreeMap<crate::WebIngressRoute, &SessionCookieConfig>, String> {
        let mut resolved = BTreeMap::new();
        for (route_id, policy) in &self.route_session_cookies {
            let mut matches = manifest
                .routes()
                .iter()
                .filter(|route| &route.route_id == route_id);
            let Some(route) = matches.next() else {
                return Err(format!(
                    "Web Ingress session Cookie policy selects unknown route {route_id}"
                ));
            };
            if matches.next().is_some() {
                return Err(format!(
                    "Web Ingress session Cookie policy selects ambiguous route ID {route_id}"
                ));
            }
            resolved.insert(route.clone(), policy);
        }
        for entry in &self.selected_route_session_cookies {
            if !manifest.routes().contains(&entry.route) {
                return Err(
                    "Web Ingress session Cookie policy selects unknown exact route".to_owned(),
                );
            }
            if resolved
                .insert(entry.route.clone(), &entry.policy)
                .is_some()
            {
                return Err(
                    "Web Ingress session Cookie policies select the same resolved route".to_owned(),
                );
            }
        }
        Ok(resolved)
    }
}

fn valid_http_token(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= MAX_PROTOCOL_NAME_BYTES
        && value.bytes().all(|byte| {
            byte.is_ascii_alphanumeric()
                || matches!(
                    byte,
                    b'!' | b'#'
                        | b'$'
                        | b'%'
                        | b'&'
                        | b'\''
                        | b'*'
                        | b'+'
                        | b'-'
                        | b'.'
                        | b'^'
                        | b'_'
                        | b'`'
                        | b'|'
                        | b'~'
                )
        })
}

fn host_cookie_name(value: &str) -> bool {
    value.starts_with("__Host-") && valid_http_token(value)
}

const fn default_bind_address() -> SocketAddr {
    SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), 0)
}

const fn default_max_request_body_bytes() -> usize {
    1024 * 1024
}

const fn default_max_request_head_bytes() -> usize {
    16 * 1024
}

const fn default_max_concurrent_requests() -> usize {
    128
}

const fn default_max_connections() -> usize {
    1_024
}

const fn default_request_head_timeout_millis() -> u64 {
    10_000
}

const fn default_request_body_timeout_millis() -> u64 {
    30_000
}

const fn default_connection_idle_timeout_millis() -> u64 {
    60_000
}

const fn default_shutdown_grace_timeout_millis() -> u64 {
    default_request_timeout_millis()
}

const fn default_request_timeout_millis() -> u64 {
    30_000
}

fn duration_millis(duration: Duration) -> Result<u64, String> {
    u64::try_from(duration.as_millis())
        .map_err(|_| "Web Ingress timeout does not fit the Plan format".to_owned())
}

#[cfg(test)]
mod tests;