lodestone-client 0.0.33

//! Exact, bounded OpenSearch-compatible execution through a validated
//! Collection session.
//!
//! OpenSearch has an independent public identity and admitted route matrix.
//! The implementation shares only the already-bounded HTTPS transport with
//! the Elasticsearch leaf; it does not select, detect, or fall back between
//! protocol identities at runtime.

use std::collections::BTreeSet;
use std::fmt::{self, Display, Formatter};
use std::time::Duration;

use http::StatusCode;

use crate::{
    Client, ElasticsearchCompatibilityEffectV1, ElasticsearchCompatibilityEndpointV1,
    ElasticsearchCompatibilityErrorV1, ElasticsearchCompatibilityExecutorV1, ElasticsearchCompatibilityHeaderV1,
    ElasticsearchCompatibilityMethodV1, ElasticsearchCompatibilityRequestV1,
};

pub const MAX_OPENSEARCH_COMPATIBILITY_REQUEST_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_OPENSEARCH_COMPATIBILITY_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_OPENSEARCH_COMPATIBILITY_HEADERS: usize = 64;
pub const MAX_OPENSEARCH_COMPATIBILITY_HEADER_BYTES: usize = 32 * 1024;

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OpenSearchCompatibilityEndpointV1 {
    transport: ElasticsearchCompatibilityEndpointV1,
}

impl OpenSearchCompatibilityEndpointV1 {
    pub fn parse(value: &str) -> Result<Self, OpenSearchCompatibilityErrorV1> {
        let transport = ElasticsearchCompatibilityEndpointV1::parse(value)
            .map_err(|_| OpenSearchCompatibilityErrorV1::InvalidRequest)?;
        Ok(Self { transport })
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OpenSearchCompatibilityMethodV1 {
    Delete,
    Get,
    Head,
    Post,
    Put,
}

impl From<OpenSearchCompatibilityMethodV1> for ElasticsearchCompatibilityMethodV1 {
    fn from(value: OpenSearchCompatibilityMethodV1) -> Self {
        match value {
            OpenSearchCompatibilityMethodV1::Delete => Self::Delete,
            OpenSearchCompatibilityMethodV1::Get => Self::Get,
            OpenSearchCompatibilityMethodV1::Head => Self::Head,
            OpenSearchCompatibilityMethodV1::Post => Self::Post,
            OpenSearchCompatibilityMethodV1::Put => Self::Put,
        }
    }
}

#[derive(Clone, PartialEq, Eq)]
pub struct OpenSearchCompatibilityHeaderV1 {
    pub name: String,
    pub value: Vec<u8>,
}

pub struct OpenSearchCompatibilityRequestV1 {
    pub operation_id: String,
    pub index_name: String,
    pub expected_index_id: String,
    pub method: OpenSearchCompatibilityMethodV1,
    pub path: String,
    pub query: Option<String>,
    pub headers: Vec<OpenSearchCompatibilityHeaderV1>,
    pub body: Vec<u8>,
    pub timeout: Duration,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OpenSearchCompatibilityEffectV1 {
    ReadOnly,
    MutationCommitted,
    MutationNoEffect,
}

pub struct OpenSearchCompatibilityResponseV1 {
    pub status: StatusCode,
    pub headers: Vec<OpenSearchCompatibilityHeaderV1>,
    pub body: Vec<u8>,
    pub effect: OpenSearchCompatibilityEffectV1,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum OpenSearchCompatibilityErrorV1 {
    InvalidRequest,
    CredentialUnavailable,
    NotSubmitted,
    Indeterminate,
    DeadlineExceeded,
    DependencyUnavailable,
    InvalidResponse,
}

impl Display for OpenSearchCompatibilityErrorV1 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidRequest => "OpenSearch compatibility request is invalid",
            Self::CredentialUnavailable => "OpenSearch compatibility credential is unavailable",
            Self::NotSubmitted => "OpenSearch compatibility request was not submitted",
            Self::Indeterminate => "OpenSearch compatibility mutation outcome is indeterminate",
            Self::DeadlineExceeded => "OpenSearch compatibility deadline was exceeded",
            Self::DependencyUnavailable => "OpenSearch compatibility dependency is unavailable",
            Self::InvalidResponse => "OpenSearch compatibility response is invalid",
        })
    }
}

impl std::error::Error for OpenSearchCompatibilityErrorV1 {}

#[derive(Clone)]
pub struct OpenSearchCompatibilityExecutorV1 {
    transport: ElasticsearchCompatibilityExecutorV1,
}

impl OpenSearchCompatibilityExecutorV1 {
    pub fn new(
        client: Client,
        endpoint: OpenSearchCompatibilityEndpointV1,
    ) -> Result<Self, OpenSearchCompatibilityErrorV1> {
        let transport = ElasticsearchCompatibilityExecutorV1::new(client, endpoint.transport).map_err(map_error)?;
        Ok(Self { transport })
    }

    pub async fn execute(
        &self,
        request: OpenSearchCompatibilityRequestV1,
    ) -> Result<OpenSearchCompatibilityResponseV1, OpenSearchCompatibilityErrorV1> {
        validate_request(&request)?;
        let mutation = request_is_mutation(request.method, &request.path);
        let response = self
            .transport
            .execute_validated(
                ElasticsearchCompatibilityRequestV1 {
                    operation_id: request.operation_id,
                    index_name: request.index_name,
                    expected_index_id: request.expected_index_id,
                    method: request.method.into(),
                    path: request.path,
                    query: request.query,
                    headers: request
                        .headers
                        .into_iter()
                        .map(|field| ElasticsearchCompatibilityHeaderV1 {
                            name: field.name,
                            value: field.value,
                        })
                        .collect(),
                    body: request.body,
                    timeout: request.timeout,
                },
                mutation,
            )
            .await
            .map_err(map_error)?;
        Ok(OpenSearchCompatibilityResponseV1 {
            status: response.status,
            headers: response
                .headers
                .into_iter()
                .map(|field| OpenSearchCompatibilityHeaderV1 {
                    name: field.name,
                    value: field.value,
                })
                .collect(),
            body: response.body,
            effect: match response.effect {
                ElasticsearchCompatibilityEffectV1::ReadOnly => OpenSearchCompatibilityEffectV1::ReadOnly,
                ElasticsearchCompatibilityEffectV1::MutationCommitted => {
                    OpenSearchCompatibilityEffectV1::MutationCommitted
                }
                ElasticsearchCompatibilityEffectV1::MutationNoEffect => {
                    OpenSearchCompatibilityEffectV1::MutationNoEffect
                }
            },
        })
    }
}

fn validate_request(request: &OpenSearchCompatibilityRequestV1) -> Result<(), OpenSearchCompatibilityErrorV1> {
    let mut header_names = BTreeSet::new();
    let header_bytes = request.headers.iter().try_fold(0usize, |total, field| {
        if !valid_header_name(&field.name)
            || !header_names.insert(field.name.as_str())
            || forbidden_header(&field.name)
            || field.value.len() > 8 * 1024
            || field.value.iter().any(|byte| matches!(byte, b'\r' | b'\n' | 0))
        {
            return Err(OpenSearchCompatibilityErrorV1::InvalidRequest);
        }
        total
            .checked_add(field.name.len())
            .and_then(|value| value.checked_add(field.value.len()))
            .ok_or(OpenSearchCompatibilityErrorV1::InvalidRequest)
    })?;
    let request_bytes = request
        .path
        .len()
        .checked_add(request.query.as_deref().map_or(0, str::len))
        .and_then(|value| value.checked_add(header_bytes))
        .and_then(|value| value.checked_add(request.body.len()))
        .ok_or(OpenSearchCompatibilityErrorV1::InvalidRequest)?;
    if !valid_uuid_v7(&request.operation_id)
        || !valid_index_name(&request.index_name)
        || !valid_lower_hex(&request.expected_index_id, 16)
        || !valid_path(&request.path)
        || request.query.as_deref().is_some_and(|value| {
            value.len() > 4_096 || value.starts_with('?') || value.contains('#') || !value.is_ascii()
        })
        || request.headers.len() > MAX_OPENSEARCH_COMPATIBILITY_HEADERS
        || header_bytes > MAX_OPENSEARCH_COMPATIBILITY_HEADER_BYTES
        || request_bytes > MAX_OPENSEARCH_COMPATIBILITY_REQUEST_BYTES
        || request.timeout.is_zero()
        || request.timeout > Duration::from_secs(60)
        || !supported_route(request.method, &request.path, &request.index_name)
    {
        return Err(OpenSearchCompatibilityErrorV1::InvalidRequest);
    }
    Ok(())
}

fn supported_route(method: OpenSearchCompatibilityMethodV1, path: &str, admitted_index: &str) -> bool {
    let segments = path.trim_start_matches('/').split('/').collect::<Vec<_>>();
    let exact = |observed: &str| observed == admitted_index;
    match (method, segments.as_slice()) {
        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, ["_bulk"])
        | (OpenSearchCompatibilityMethodV1::Post, ["_mget" | "_msearch" | "_search"])
        | (OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post, ["_validate", "query"])
        | (OpenSearchCompatibilityMethodV1::Get, ["_search"]) => true,
        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, [index, "_bulk"])
            if exact(index) =>
        {
            true
        }
        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, [index, "_doc", _])
            if exact(index) =>
        {
            true
        }
        (OpenSearchCompatibilityMethodV1::Put, [index, "_create", _]) if exact(index) => true,
        (OpenSearchCompatibilityMethodV1::Post, [index, "_update", _]) if exact(index) => true,
        (
            OpenSearchCompatibilityMethodV1::Get
            | OpenSearchCompatibilityMethodV1::Head
            | OpenSearchCompatibilityMethodV1::Delete,
            [index, "_doc", _],
        ) if exact(index) => true,
        (
            OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post,
            [index, "_search" | "_count" | "_terms_enum" | "_field_caps"],
        ) if exact(index) => true,
        (OpenSearchCompatibilityMethodV1::Post, [index, "_mget" | "_msearch"]) if exact(index) => true,
        (
            OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post,
            [index, "_validate", "query"],
        ) if exact(index) => true,
        _ => false,
    }
}

fn request_is_mutation(method: OpenSearchCompatibilityMethodV1, path: &str) -> bool {
    match method {
        OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Head => false,
        OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Delete => true,
        OpenSearchCompatibilityMethodV1::Post => {
            !(matches!(path, "/_search" | "/_msearch" | "/_mget")
                || path.ends_with("/_search")
                || path.ends_with("/_msearch")
                || path.ends_with("/_mget")
                || path.ends_with("/_count")
                || path.ends_with("/_terms_enum")
                || path.ends_with("/_field_caps")
                || path.ends_with("/_validate/query"))
        }
    }
}

fn valid_path(value: &str) -> bool {
    value.starts_with('/')
        && !value.starts_with("//")
        && value.len() <= 2_048
        && !value.contains(['?', '#', '\\'])
        && !value.split('/').any(|segment| segment == "..")
        && value.bytes().all(|byte| byte.is_ascii_graphic())
}

fn valid_index_name(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 255
        && value
            .bytes()
            .all(|byte| byte.is_ascii_graphic() && byte != b'/' && byte != b'\\')
}

fn valid_header_name(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 128
        && value
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
}

fn forbidden_header(name: &str) -> bool {
    name.starts_with("x-lodestone-")
        || matches!(
            name,
            "authorization"
                | "proxy-authorization"
                | "cookie"
                | "host"
                | "connection"
                | "content-length"
                | "keep-alive"
                | "proxy-connection"
                | "te"
                | "trailer"
                | "transfer-encoding"
                | "upgrade"
                | "forwarded"
                | "x-forwarded-for"
                | "x-forwarded-host"
                | "x-forwarded-proto"
                | "x-opaque-id"
        )
}

fn valid_lower_hex(value: &str, bytes: usize) -> bool {
    value.len() == bytes.saturating_mul(2)
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
        && value.bytes().any(|byte| byte != b'0')
}

fn valid_uuid_v7(value: &str) -> bool {
    value.len() == 36
        && value.as_bytes().get(8) == Some(&b'-')
        && value.as_bytes().get(13) == Some(&b'-')
        && value.as_bytes().get(18) == Some(&b'-')
        && value.as_bytes().get(23) == Some(&b'-')
        && value.as_bytes().get(14) == Some(&b'7')
        && value
            .as_bytes()
            .get(19)
            .is_some_and(|value| matches!(value, b'8' | b'9' | b'a' | b'b'))
        && value.bytes().enumerate().all(|(index, byte)| {
            matches!(index, 8 | 13 | 18 | 23) || byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)
        })
}

fn map_error(error: ElasticsearchCompatibilityErrorV1) -> OpenSearchCompatibilityErrorV1 {
    match error {
        ElasticsearchCompatibilityErrorV1::InvalidRequest => OpenSearchCompatibilityErrorV1::InvalidRequest,
        ElasticsearchCompatibilityErrorV1::CredentialUnavailable => {
            OpenSearchCompatibilityErrorV1::CredentialUnavailable
        }
        ElasticsearchCompatibilityErrorV1::NotSubmitted => OpenSearchCompatibilityErrorV1::NotSubmitted,
        ElasticsearchCompatibilityErrorV1::Indeterminate => OpenSearchCompatibilityErrorV1::Indeterminate,
        ElasticsearchCompatibilityErrorV1::DeadlineExceeded => OpenSearchCompatibilityErrorV1::DeadlineExceeded,
        ElasticsearchCompatibilityErrorV1::DependencyUnavailable => {
            OpenSearchCompatibilityErrorV1::DependencyUnavailable
        }
        ElasticsearchCompatibilityErrorV1::InvalidResponse => OpenSearchCompatibilityErrorV1::InvalidResponse,
    }
}

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

    #[test]
    fn endpoint_and_opensearch_matrix_are_independently_closed() {
        assert!(OpenSearchCompatibilityEndpointV1::parse("https://core.example:8443").is_ok());
        assert!(OpenSearchCompatibilityEndpointV1::parse("http://core.example:8443").is_err());
        assert!(supported_route(
            OpenSearchCompatibilityMethodV1::Post,
            "/logs/_search",
            "logs"
        ));
        assert!(supported_route(OpenSearchCompatibilityMethodV1::Post, "/_bulk", "logs"));
        assert!(!supported_route(
            OpenSearchCompatibilityMethodV1::Post,
            "/other/_search",
            "logs"
        ));
        assert!(!supported_route(
            OpenSearchCompatibilityMethodV1::Post,
            "/_plugins/_security/api/roles",
            "logs"
        ));
        assert!(!supported_route(
            OpenSearchCompatibilityMethodV1::Post,
            "/_reindex",
            "logs"
        ));
    }

    #[test]
    fn mutation_classification_never_turns_response_loss_into_a_retry() {
        assert!(request_is_mutation(
            OpenSearchCompatibilityMethodV1::Post,
            "/logs/_update/1"
        ));
        assert!(!request_is_mutation(
            OpenSearchCompatibilityMethodV1::Post,
            "/logs/_search"
        ));
    }
}