lodestone_client/

compatibility.rs

//! Exact, bounded Elasticsearch compatibility execution through a validated
//! Collection session.
//!
//! The executor owns Core endpoint, credential, Collection, and TLS material.
//! One execution request supplies only the exact index incarnation and the
//! bounded external HTTP envelope. It never retries a request.

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

use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header};
use reqwest::{Certificate, Method, Url, redirect::Policy};

use crate::transport::{Client, authorization_value};
use crate::{ClientError, CredentialProviderError};

pub const MAX_ELASTICSEARCH_COMPATIBILITY_REQUEST_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_ELASTICSEARCH_COMPATIBILITY_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_ELASTICSEARCH_COMPATIBILITY_HEADERS: usize = 64;
pub const MAX_ELASTICSEARCH_COMPATIBILITY_HEADER_BYTES: usize = 32 * 1024;

const COLLECTION_ID_HEADER: &str = "x-lodestone-collection-id";
const INDEX_NAME_HEADER: &str = "x-lodestone-index-name";
const INDEX_ID_HEADER: &str = "x-lodestone-index-id";
const EFFECT_HEADER: &str = "x-lodestone-compatibility-effect";

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

impl ElasticsearchCompatibilityEndpointV1 {
    pub fn parse(value: &str) -> Result<Self, ElasticsearchCompatibilityErrorV1> {
        let origin = Url::parse(value).map_err(|_| ElasticsearchCompatibilityErrorV1::InvalidRequest)?;
        if origin.scheme() != "https"
            || origin.cannot_be_a_base()
            || origin.username() != ""
            || origin.password().is_some()
            || origin.query().is_some()
            || origin.fragment().is_some()
            || origin.host_str().is_none()
            || !matches!(origin.path(), "" | "/")
        {
            return Err(ElasticsearchCompatibilityErrorV1::InvalidRequest);
        }
        Ok(Self { origin })
    }

    fn request_url(&self, path: &str, query: Option<&str>) -> Result<Url, ElasticsearchCompatibilityErrorV1> {
        let mut url = self
            .origin
            .join(path.trim_start_matches('/'))
            .map_err(|_| ElasticsearchCompatibilityErrorV1::InvalidRequest)?;
        url.set_query(query);
        Ok(url)
    }
}

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

impl ElasticsearchCompatibilityMethodV1 {
    fn reqwest(self) -> Method {
        match self {
            Self::Delete => Method::DELETE,
            Self::Get => Method::GET,
            Self::Head => Method::HEAD,
            Self::Post => Method::POST,
            Self::Put => Method::PUT,
        }
    }
}

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

pub struct ElasticsearchCompatibilityRequestV1 {
    pub operation_id: String,
    pub index_name: String,
    pub expected_index_id: String,
    pub method: ElasticsearchCompatibilityMethodV1,
    pub path: String,
    pub query: Option<String>,
    pub headers: Vec<ElasticsearchCompatibilityHeaderV1>,
    pub body: Vec<u8>,
    pub timeout: Duration,
}

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

pub struct ElasticsearchCompatibilityResponseV1 {
    pub status: StatusCode,
    pub headers: Vec<ElasticsearchCompatibilityHeaderV1>,
    pub body: Vec<u8>,
    pub effect: ElasticsearchCompatibilityEffectV1,
}

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

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

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

#[derive(Clone)]
pub struct ElasticsearchCompatibilityExecutorV1 {
    client: Client,
    endpoint: ElasticsearchCompatibilityEndpointV1,
}

impl ElasticsearchCompatibilityExecutorV1 {
    pub fn new(
        client: Client,
        endpoint: ElasticsearchCompatibilityEndpointV1,
    ) -> Result<Self, ElasticsearchCompatibilityErrorV1> {
        if endpoint.origin.host_str() != Some(client.transport_config().identity.tls_server_name()) {
            return Err(ElasticsearchCompatibilityErrorV1::InvalidRequest);
        }
        Ok(Self { client, endpoint })
    }

    pub async fn execute(
        &self,
        request: ElasticsearchCompatibilityRequestV1,
    ) -> Result<ElasticsearchCompatibilityResponseV1, ElasticsearchCompatibilityErrorV1> {
        validate_request(&request)?;
        let mutation = request_is_mutation(request.method, &request.path);
        self.execute_validated(request, mutation).await
    }

    pub(crate) async fn execute_validated(
        &self,
        request: ElasticsearchCompatibilityRequestV1,
        mutation: bool,
    ) -> Result<ElasticsearchCompatibilityResponseV1, ElasticsearchCompatibilityErrorV1> {
        let material = self.client.credential().await.map_err(map_client_error)?;
        let authorization = authorization_value(&material).map_err(map_client_error)?;
        let trust = self
            .client
            .transport_config()
            .trust
            .certificates_der()
            .iter()
            .map(|certificate| {
                Certificate::from_der(certificate).map_err(|_| ElasticsearchCompatibilityErrorV1::InvalidRequest)
            })
            .collect::<Result<Vec<_>, _>>()?;
        let client = reqwest::Client::builder()
            .https_only(true)
            .no_proxy()
            .redirect(Policy::none())
            .connect_timeout(self.client.pool_config().connect_timeout)
            .timeout(request.timeout)
            .tls_certs_only(trust)
            .build()
            .map_err(|_| ElasticsearchCompatibilityErrorV1::DependencyUnavailable)?;
        let url = self.endpoint.request_url(&request.path, request.query.as_deref())?;
        let mut outbound = client
            .request(request.method.reqwest(), url)
            .header(header::AUTHORIZATION, sensitive_header(authorization.as_slice())?)
            .header(COLLECTION_ID_HEADER, self.client.expected_collection_id())
            .header(INDEX_NAME_HEADER, &request.index_name)
            .header(INDEX_ID_HEADER, &request.expected_index_id)
            .header("x-opaque-id", &request.operation_id)
            .body(request.body);
        for field in request.headers {
            let name = HeaderName::from_bytes(field.name.as_bytes())
                .map_err(|_| ElasticsearchCompatibilityErrorV1::InvalidRequest)?;
            let value =
                HeaderValue::from_bytes(&field.value).map_err(|_| ElasticsearchCompatibilityErrorV1::InvalidRequest)?;
            outbound = outbound.header(name, value);
        }
        let mut response = outbound
            .send()
            .await
            .map_err(|error| map_reqwest_error(error, mutation))?;
        if response.content_length().is_some_and(|length| {
            length > u64::try_from(MAX_ELASTICSEARCH_COMPATIBILITY_RESPONSE_BYTES).unwrap_or(u64::MAX)
        }) {
            return Err(ElasticsearchCompatibilityErrorV1::InvalidResponse);
        }
        let status = response.status();
        let effect = parse_effect(response.headers(), mutation)?;
        let headers = response_headers(response.headers())?;
        let mut body = Vec::new();
        while let Some(chunk) = response.chunk().await.map_err(|_| {
            if mutation {
                ElasticsearchCompatibilityErrorV1::Indeterminate
            } else {
                ElasticsearchCompatibilityErrorV1::DependencyUnavailable
            }
        })? {
            if body.len().saturating_add(chunk.len()) > MAX_ELASTICSEARCH_COMPATIBILITY_RESPONSE_BYTES {
                return Err(ElasticsearchCompatibilityErrorV1::InvalidResponse);
            }
            body.extend_from_slice(&chunk);
        }
        Ok(ElasticsearchCompatibilityResponseV1 {
            status,
            headers,
            body,
            effect,
        })
    }
}

fn validate_request(request: &ElasticsearchCompatibilityRequestV1) -> Result<(), ElasticsearchCompatibilityErrorV1> {
    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(ElasticsearchCompatibilityErrorV1::InvalidRequest);
        }
        total
            .checked_add(field.name.len())
            .and_then(|value| value.checked_add(field.value.len()))
            .ok_or(ElasticsearchCompatibilityErrorV1::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(ElasticsearchCompatibilityErrorV1::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_ELASTICSEARCH_COMPATIBILITY_HEADERS
        || header_bytes > MAX_ELASTICSEARCH_COMPATIBILITY_HEADER_BYTES
        || request_bytes > MAX_ELASTICSEARCH_COMPATIBILITY_REQUEST_BYTES
        || request.timeout.is_zero()
        || request.timeout > Duration::from_secs(60)
        || !supported_route(request.method, &request.path, &request.index_name)
    {
        return Err(ElasticsearchCompatibilityErrorV1::InvalidRequest);
    }
    Ok(())
}

fn supported_route(method: ElasticsearchCompatibilityMethodV1, 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()) {
        (ElasticsearchCompatibilityMethodV1::Put | ElasticsearchCompatibilityMethodV1::Post, ["_bulk"])
        | (ElasticsearchCompatibilityMethodV1::Post, ["_mget" | "_msearch" | "_search"])
        | (
            ElasticsearchCompatibilityMethodV1::Get | ElasticsearchCompatibilityMethodV1::Post,
            ["_validate", "query"],
        )
        | (ElasticsearchCompatibilityMethodV1::Get, ["_search"]) => true,
        (ElasticsearchCompatibilityMethodV1::Put | ElasticsearchCompatibilityMethodV1::Post, [index, "_bulk"])
            if exact(index) =>
        {
            true
        }
        (ElasticsearchCompatibilityMethodV1::Put | ElasticsearchCompatibilityMethodV1::Post, [index, "_doc", _])
            if exact(index) =>
        {
            true
        }
        (ElasticsearchCompatibilityMethodV1::Put, [index, "_create", _]) if exact(index) => true,
        (ElasticsearchCompatibilityMethodV1::Post, [index, "_update", _]) if exact(index) => true,
        (
            ElasticsearchCompatibilityMethodV1::Get
            | ElasticsearchCompatibilityMethodV1::Head
            | ElasticsearchCompatibilityMethodV1::Delete,
            [index, "_doc", _],
        ) if exact(index) => true,
        (
            ElasticsearchCompatibilityMethodV1::Get | ElasticsearchCompatibilityMethodV1::Post,
            [index, "_search" | "_count" | "_terms_enum" | "_field_caps"],
        ) if exact(index) => true,
        (ElasticsearchCompatibilityMethodV1::Post, [index, "_mget" | "_msearch"]) if exact(index) => true,
        (
            ElasticsearchCompatibilityMethodV1::Get | ElasticsearchCompatibilityMethodV1::Post,
            [index, "_validate", "query"],
        ) if exact(index) => true,
        _ => false,
    }
}

fn request_is_mutation(method: ElasticsearchCompatibilityMethodV1, path: &str) -> bool {
    match method {
        ElasticsearchCompatibilityMethodV1::Get | ElasticsearchCompatibilityMethodV1::Head => false,
        ElasticsearchCompatibilityMethodV1::Put | ElasticsearchCompatibilityMethodV1::Delete => true,
        ElasticsearchCompatibilityMethodV1::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 parse_effect(
    headers: &HeaderMap,
    mutation: bool,
) -> Result<ElasticsearchCompatibilityEffectV1, ElasticsearchCompatibilityErrorV1> {
    let mut values = headers.get_all(EFFECT_HEADER).iter();
    let value = values.next().and_then(|value| value.to_str().ok());
    if values.next().is_some() {
        return Err(ElasticsearchCompatibilityErrorV1::InvalidResponse);
    }
    match value {
        Some("read_only") if !mutation => Ok(ElasticsearchCompatibilityEffectV1::ReadOnly),
        Some("mutation_committed") if mutation => Ok(ElasticsearchCompatibilityEffectV1::MutationCommitted),
        Some("mutation_no_effect") if mutation => Ok(ElasticsearchCompatibilityEffectV1::MutationNoEffect),
        Some("mutation_indeterminate") if mutation => Err(ElasticsearchCompatibilityErrorV1::Indeterminate),
        _ => Err(ElasticsearchCompatibilityErrorV1::InvalidResponse),
    }
}

fn response_headers(
    headers: &HeaderMap,
) -> Result<Vec<ElasticsearchCompatibilityHeaderV1>, ElasticsearchCompatibilityErrorV1> {
    let mut output = Vec::new();
    let mut bytes = 0usize;
    for (name, value) in headers {
        let name = name.as_str();
        if name == EFFECT_HEADER || forbidden_response_header(name) {
            continue;
        }
        bytes = bytes
            .checked_add(name.len())
            .and_then(|total| total.checked_add(value.as_bytes().len()))
            .ok_or(ElasticsearchCompatibilityErrorV1::InvalidResponse)?;
        if output.len() >= MAX_ELASTICSEARCH_COMPATIBILITY_HEADERS
            || bytes > MAX_ELASTICSEARCH_COMPATIBILITY_HEADER_BYTES
        {
            return Err(ElasticsearchCompatibilityErrorV1::InvalidResponse);
        }
        output.push(ElasticsearchCompatibilityHeaderV1 {
            name: name.to_owned(),
            value: value.as_bytes().to_vec(),
        });
    }
    Ok(output)
}

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 forbidden_response_header(name: &str) -> bool {
    name.starts_with("x-lodestone-")
        || matches!(
            name,
            "content-length"
                | "connection"
                | "keep-alive"
                | "proxy-connection"
                | "te"
                | "trailer"
                | "transfer-encoding"
                | "upgrade"
                | "set-cookie"
                | "proxy-authenticate"
                | "www-authenticate"
        )
}

fn sensitive_header(value: &[u8]) -> Result<HeaderValue, ElasticsearchCompatibilityErrorV1> {
    let mut header =
        HeaderValue::from_bytes(value).map_err(|_| ElasticsearchCompatibilityErrorV1::CredentialUnavailable)?;
    header.set_sensitive(true);
    Ok(header)
}

fn map_client_error(error: ClientError) -> ElasticsearchCompatibilityErrorV1 {
    match error {
        ClientError::Credential(
            CredentialProviderError::Unavailable
            | CredentialProviderError::NotFound
            | CredentialProviderError::InvalidMaterial
            | CredentialProviderError::RandomUnavailable,
        ) => ElasticsearchCompatibilityErrorV1::CredentialUnavailable,
        _ => ElasticsearchCompatibilityErrorV1::DependencyUnavailable,
    }
}

fn map_reqwest_error(error: reqwest::Error, mutation: bool) -> ElasticsearchCompatibilityErrorV1 {
    if error.is_builder() {
        ElasticsearchCompatibilityErrorV1::InvalidRequest
    } else if error.is_connect() {
        ElasticsearchCompatibilityErrorV1::NotSubmitted
    } else if mutation {
        ElasticsearchCompatibilityErrorV1::Indeterminate
    } else if error.is_timeout() {
        ElasticsearchCompatibilityErrorV1::DeadlineExceeded
    } else {
        ElasticsearchCompatibilityErrorV1::DependencyUnavailable
    }
}

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)
        })
}

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

    #[test]
    fn endpoint_and_exact_data_matrix_are_closed() {
        assert!(ElasticsearchCompatibilityEndpointV1::parse("https://core.example:8443").is_ok());
        assert!(ElasticsearchCompatibilityEndpointV1::parse("http://core.example:8443").is_err());
        assert!(forbidden_header("x-lodestone-private-authority"));
        assert!(forbidden_response_header("x-lodestone-private-evidence"));
        assert!(supported_route(
            ElasticsearchCompatibilityMethodV1::Post,
            "/logs/_search",
            "logs"
        ));
        assert!(supported_route(
            ElasticsearchCompatibilityMethodV1::Post,
            "/_bulk",
            "logs"
        ));
        assert!(!supported_route(
            ElasticsearchCompatibilityMethodV1::Post,
            "/other/_search",
            "logs"
        ));
        assert!(!supported_route(
            ElasticsearchCompatibilityMethodV1::Post,
            "/_reindex",
            "logs"
        ));
    }

    #[test]
    fn response_effect_requires_protocol_and_mutation_agreement() {
        let mut headers = HeaderMap::new();
        headers.insert(EFFECT_HEADER, HeaderValue::from_static("mutation_committed"));
        assert_eq!(
            parse_effect(&headers, true),
            Ok(ElasticsearchCompatibilityEffectV1::MutationCommitted)
        );
        assert_eq!(
            parse_effect(&headers, false),
            Err(ElasticsearchCompatibilityErrorV1::InvalidResponse)
        );
        headers.insert(EFFECT_HEADER, HeaderValue::from_static("mutation_indeterminate"));
        assert_eq!(
            parse_effect(&headers, true),
            Err(ElasticsearchCompatibilityErrorV1::Indeterminate)
        );
        headers.remove(EFFECT_HEADER);
        assert_eq!(
            parse_effect(&headers, true),
            Err(ElasticsearchCompatibilityErrorV1::InvalidResponse)
        );
        assert_eq!(
            parse_effect(&headers, false),
            Err(ElasticsearchCompatibilityErrorV1::InvalidResponse)
        );
    }

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