lodestone_client/
opensearch_compatibility.rs

1//! Exact, bounded OpenSearch-compatible execution through a validated
2//! Collection session.
3//!
4//! OpenSearch has an independent public identity and admitted route matrix.
5//! The implementation shares only the already-bounded HTTPS transport with
6//! the Elasticsearch leaf; it does not select, detect, or fall back between
7//! protocol identities at runtime.
8
9use std::collections::BTreeSet;
10use std::fmt::{self, Display, Formatter};
11use std::time::Duration;
12
13use http::StatusCode;
14
15use crate::{
16    Client, ElasticsearchCompatibilityEffectV1, ElasticsearchCompatibilityEndpointV1,
17    ElasticsearchCompatibilityErrorV1, ElasticsearchCompatibilityExecutorV1, ElasticsearchCompatibilityHeaderV1,
18    ElasticsearchCompatibilityMethodV1, ElasticsearchCompatibilityRequestV1,
19};
20
21pub const MAX_OPENSEARCH_COMPATIBILITY_REQUEST_BYTES: usize = 2 * 1024 * 1024;
22pub const MAX_OPENSEARCH_COMPATIBILITY_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
23pub const MAX_OPENSEARCH_COMPATIBILITY_HEADERS: usize = 64;
24pub const MAX_OPENSEARCH_COMPATIBILITY_HEADER_BYTES: usize = 32 * 1024;
25
26#[derive(Clone, Debug, Eq, PartialEq)]
27pub struct OpenSearchCompatibilityEndpointV1 {
28    transport: ElasticsearchCompatibilityEndpointV1,
29}
30
31impl OpenSearchCompatibilityEndpointV1 {
32    pub fn parse(value: &str) -> Result<Self, OpenSearchCompatibilityErrorV1> {
33        let transport = ElasticsearchCompatibilityEndpointV1::parse(value)
34            .map_err(|_| OpenSearchCompatibilityErrorV1::InvalidRequest)?;
35        Ok(Self { transport })
36    }
37}
38
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum OpenSearchCompatibilityMethodV1 {
41    Delete,
42    Get,
43    Head,
44    Post,
45    Put,
46}
47
48impl From<OpenSearchCompatibilityMethodV1> for ElasticsearchCompatibilityMethodV1 {
49    fn from(value: OpenSearchCompatibilityMethodV1) -> Self {
50        match value {
51            OpenSearchCompatibilityMethodV1::Delete => Self::Delete,
52            OpenSearchCompatibilityMethodV1::Get => Self::Get,
53            OpenSearchCompatibilityMethodV1::Head => Self::Head,
54            OpenSearchCompatibilityMethodV1::Post => Self::Post,
55            OpenSearchCompatibilityMethodV1::Put => Self::Put,
56        }
57    }
58}
59
60#[derive(Clone, PartialEq, Eq)]
61pub struct OpenSearchCompatibilityHeaderV1 {
62    pub name: String,
63    pub value: Vec<u8>,
64}
65
66pub struct OpenSearchCompatibilityRequestV1 {
67    pub operation_id: String,
68    pub index_name: String,
69    pub expected_index_id: String,
70    pub method: OpenSearchCompatibilityMethodV1,
71    pub path: String,
72    pub query: Option<String>,
73    pub headers: Vec<OpenSearchCompatibilityHeaderV1>,
74    pub body: Vec<u8>,
75    pub timeout: Duration,
76}
77
78#[derive(Clone, Copy, Debug, Eq, PartialEq)]
79pub enum OpenSearchCompatibilityEffectV1 {
80    ReadOnly,
81    MutationCommitted,
82    MutationNoEffect,
83}
84
85pub struct OpenSearchCompatibilityResponseV1 {
86    pub status: StatusCode,
87    pub headers: Vec<OpenSearchCompatibilityHeaderV1>,
88    pub body: Vec<u8>,
89    pub effect: OpenSearchCompatibilityEffectV1,
90}
91
92#[derive(Clone, Copy, Debug, Eq, PartialEq)]
93pub enum OpenSearchCompatibilityErrorV1 {
94    InvalidRequest,
95    CredentialUnavailable,
96    NotSubmitted,
97    Indeterminate,
98    DeadlineExceeded,
99    DependencyUnavailable,
100    InvalidResponse,
101}
102
103impl Display for OpenSearchCompatibilityErrorV1 {
104    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
105        formatter.write_str(match self {
106            Self::InvalidRequest => "OpenSearch compatibility request is invalid",
107            Self::CredentialUnavailable => "OpenSearch compatibility credential is unavailable",
108            Self::NotSubmitted => "OpenSearch compatibility request was not submitted",
109            Self::Indeterminate => "OpenSearch compatibility mutation outcome is indeterminate",
110            Self::DeadlineExceeded => "OpenSearch compatibility deadline was exceeded",
111            Self::DependencyUnavailable => "OpenSearch compatibility dependency is unavailable",
112            Self::InvalidResponse => "OpenSearch compatibility response is invalid",
113        })
114    }
115}
116
117impl std::error::Error for OpenSearchCompatibilityErrorV1 {}
118
119#[derive(Clone)]
120pub struct OpenSearchCompatibilityExecutorV1 {
121    transport: ElasticsearchCompatibilityExecutorV1,
122}
123
124impl OpenSearchCompatibilityExecutorV1 {
125    pub fn new(
126        client: Client,
127        endpoint: OpenSearchCompatibilityEndpointV1,
128    ) -> Result<Self, OpenSearchCompatibilityErrorV1> {
129        let transport = ElasticsearchCompatibilityExecutorV1::new(client, endpoint.transport).map_err(map_error)?;
130        Ok(Self { transport })
131    }
132
133    pub async fn execute(
134        &self,
135        request: OpenSearchCompatibilityRequestV1,
136    ) -> Result<OpenSearchCompatibilityResponseV1, OpenSearchCompatibilityErrorV1> {
137        validate_request(&request)?;
138        let mutation = request_is_mutation(request.method, &request.path);
139        let response = self
140            .transport
141            .execute_validated(
142                ElasticsearchCompatibilityRequestV1 {
143                    operation_id: request.operation_id,
144                    index_name: request.index_name,
145                    expected_index_id: request.expected_index_id,
146                    method: request.method.into(),
147                    path: request.path,
148                    query: request.query,
149                    headers: request
150                        .headers
151                        .into_iter()
152                        .map(|field| ElasticsearchCompatibilityHeaderV1 {
153                            name: field.name,
154                            value: field.value,
155                        })
156                        .collect(),
157                    body: request.body,
158                    timeout: request.timeout,
159                },
160                mutation,
161            )
162            .await
163            .map_err(map_error)?;
164        Ok(OpenSearchCompatibilityResponseV1 {
165            status: response.status,
166            headers: response
167                .headers
168                .into_iter()
169                .map(|field| OpenSearchCompatibilityHeaderV1 {
170                    name: field.name,
171                    value: field.value,
172                })
173                .collect(),
174            body: response.body,
175            effect: match response.effect {
176                ElasticsearchCompatibilityEffectV1::ReadOnly => OpenSearchCompatibilityEffectV1::ReadOnly,
177                ElasticsearchCompatibilityEffectV1::MutationCommitted => {
178                    OpenSearchCompatibilityEffectV1::MutationCommitted
179                }
180                ElasticsearchCompatibilityEffectV1::MutationNoEffect => {
181                    OpenSearchCompatibilityEffectV1::MutationNoEffect
182                }
183            },
184        })
185    }
186}
187
188fn validate_request(request: &OpenSearchCompatibilityRequestV1) -> Result<(), OpenSearchCompatibilityErrorV1> {
189    let mut header_names = BTreeSet::new();
190    let header_bytes = request.headers.iter().try_fold(0usize, |total, field| {
191        if !valid_header_name(&field.name)
192            || !header_names.insert(field.name.as_str())
193            || forbidden_header(&field.name)
194            || field.value.len() > 8 * 1024
195            || field.value.iter().any(|byte| matches!(byte, b'\r' | b'\n' | 0))
196        {
197            return Err(OpenSearchCompatibilityErrorV1::InvalidRequest);
198        }
199        total
200            .checked_add(field.name.len())
201            .and_then(|value| value.checked_add(field.value.len()))
202            .ok_or(OpenSearchCompatibilityErrorV1::InvalidRequest)
203    })?;
204    let request_bytes = request
205        .path
206        .len()
207        .checked_add(request.query.as_deref().map_or(0, str::len))
208        .and_then(|value| value.checked_add(header_bytes))
209        .and_then(|value| value.checked_add(request.body.len()))
210        .ok_or(OpenSearchCompatibilityErrorV1::InvalidRequest)?;
211    if !valid_uuid_v7(&request.operation_id)
212        || !valid_index_name(&request.index_name)
213        || !valid_lower_hex(&request.expected_index_id, 16)
214        || !valid_path(&request.path)
215        || request.query.as_deref().is_some_and(|value| {
216            value.len() > 4_096 || value.starts_with('?') || value.contains('#') || !value.is_ascii()
217        })
218        || request.headers.len() > MAX_OPENSEARCH_COMPATIBILITY_HEADERS
219        || header_bytes > MAX_OPENSEARCH_COMPATIBILITY_HEADER_BYTES
220        || request_bytes > MAX_OPENSEARCH_COMPATIBILITY_REQUEST_BYTES
221        || request.timeout.is_zero()
222        || request.timeout > Duration::from_secs(60)
223        || !supported_route(request.method, &request.path, &request.index_name)
224    {
225        return Err(OpenSearchCompatibilityErrorV1::InvalidRequest);
226    }
227    Ok(())
228}
229
230fn supported_route(method: OpenSearchCompatibilityMethodV1, path: &str, admitted_index: &str) -> bool {
231    let segments = path.trim_start_matches('/').split('/').collect::<Vec<_>>();
232    let exact = |observed: &str| observed == admitted_index;
233    match (method, segments.as_slice()) {
234        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, ["_bulk"])
235        | (OpenSearchCompatibilityMethodV1::Post, ["_mget" | "_msearch" | "_search"])
236        | (OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post, ["_validate", "query"])
237        | (OpenSearchCompatibilityMethodV1::Get, ["_search"]) => true,
238        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, [index, "_bulk"])
239            if exact(index) =>
240        {
241            true
242        }
243        (OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Post, [index, "_doc", _])
244            if exact(index) =>
245        {
246            true
247        }
248        (OpenSearchCompatibilityMethodV1::Put, [index, "_create", _]) if exact(index) => true,
249        (OpenSearchCompatibilityMethodV1::Post, [index, "_update", _]) if exact(index) => true,
250        (
251            OpenSearchCompatibilityMethodV1::Get
252            | OpenSearchCompatibilityMethodV1::Head
253            | OpenSearchCompatibilityMethodV1::Delete,
254            [index, "_doc", _],
255        ) if exact(index) => true,
256        (
257            OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post,
258            [index, "_search" | "_count" | "_terms_enum" | "_field_caps"],
259        ) if exact(index) => true,
260        (OpenSearchCompatibilityMethodV1::Post, [index, "_mget" | "_msearch"]) if exact(index) => true,
261        (
262            OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Post,
263            [index, "_validate", "query"],
264        ) if exact(index) => true,
265        _ => false,
266    }
267}
268
269fn request_is_mutation(method: OpenSearchCompatibilityMethodV1, path: &str) -> bool {
270    match method {
271        OpenSearchCompatibilityMethodV1::Get | OpenSearchCompatibilityMethodV1::Head => false,
272        OpenSearchCompatibilityMethodV1::Put | OpenSearchCompatibilityMethodV1::Delete => true,
273        OpenSearchCompatibilityMethodV1::Post => {
274            !(matches!(path, "/_search" | "/_msearch" | "/_mget")
275                || path.ends_with("/_search")
276                || path.ends_with("/_msearch")
277                || path.ends_with("/_mget")
278                || path.ends_with("/_count")
279                || path.ends_with("/_terms_enum")
280                || path.ends_with("/_field_caps")
281                || path.ends_with("/_validate/query"))
282        }
283    }
284}
285
286fn valid_path(value: &str) -> bool {
287    value.starts_with('/')
288        && !value.starts_with("//")
289        && value.len() <= 2_048
290        && !value.contains(['?', '#', '\\'])
291        && !value.split('/').any(|segment| segment == "..")
292        && value.bytes().all(|byte| byte.is_ascii_graphic())
293}
294
295fn valid_index_name(value: &str) -> bool {
296    !value.is_empty()
297        && value.len() <= 255
298        && value
299            .bytes()
300            .all(|byte| byte.is_ascii_graphic() && byte != b'/' && byte != b'\\')
301}
302
303fn valid_header_name(value: &str) -> bool {
304    !value.is_empty()
305        && value.len() <= 128
306        && value
307            .bytes()
308            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
309}
310
311fn forbidden_header(name: &str) -> bool {
312    name.starts_with("x-lodestone-")
313        || matches!(
314            name,
315            "authorization"
316                | "proxy-authorization"
317                | "cookie"
318                | "host"
319                | "connection"
320                | "content-length"
321                | "keep-alive"
322                | "proxy-connection"
323                | "te"
324                | "trailer"
325                | "transfer-encoding"
326                | "upgrade"
327                | "forwarded"
328                | "x-forwarded-for"
329                | "x-forwarded-host"
330                | "x-forwarded-proto"
331                | "x-opaque-id"
332        )
333}
334
335fn valid_lower_hex(value: &str, bytes: usize) -> bool {
336    value.len() == bytes.saturating_mul(2)
337        && value
338            .bytes()
339            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
340        && value.bytes().any(|byte| byte != b'0')
341}
342
343fn valid_uuid_v7(value: &str) -> bool {
344    value.len() == 36
345        && value.as_bytes().get(8) == Some(&b'-')
346        && value.as_bytes().get(13) == Some(&b'-')
347        && value.as_bytes().get(18) == Some(&b'-')
348        && value.as_bytes().get(23) == Some(&b'-')
349        && value.as_bytes().get(14) == Some(&b'7')
350        && value
351            .as_bytes()
352            .get(19)
353            .is_some_and(|value| matches!(value, b'8' | b'9' | b'a' | b'b'))
354        && value.bytes().enumerate().all(|(index, byte)| {
355            matches!(index, 8 | 13 | 18 | 23) || byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)
356        })
357}
358
359fn map_error(error: ElasticsearchCompatibilityErrorV1) -> OpenSearchCompatibilityErrorV1 {
360    match error {
361        ElasticsearchCompatibilityErrorV1::InvalidRequest => OpenSearchCompatibilityErrorV1::InvalidRequest,
362        ElasticsearchCompatibilityErrorV1::CredentialUnavailable => {
363            OpenSearchCompatibilityErrorV1::CredentialUnavailable
364        }
365        ElasticsearchCompatibilityErrorV1::NotSubmitted => OpenSearchCompatibilityErrorV1::NotSubmitted,
366        ElasticsearchCompatibilityErrorV1::Indeterminate => OpenSearchCompatibilityErrorV1::Indeterminate,
367        ElasticsearchCompatibilityErrorV1::DeadlineExceeded => OpenSearchCompatibilityErrorV1::DeadlineExceeded,
368        ElasticsearchCompatibilityErrorV1::DependencyUnavailable => {
369            OpenSearchCompatibilityErrorV1::DependencyUnavailable
370        }
371        ElasticsearchCompatibilityErrorV1::InvalidResponse => OpenSearchCompatibilityErrorV1::InvalidResponse,
372    }
373}
374
375#[cfg(test)]
376mod tests {
377    use super::*;
378
379    #[test]
380    fn endpoint_and_opensearch_matrix_are_independently_closed() {
381        assert!(OpenSearchCompatibilityEndpointV1::parse("https://core.example:8443").is_ok());
382        assert!(OpenSearchCompatibilityEndpointV1::parse("http://core.example:8443").is_err());
383        assert!(supported_route(
384            OpenSearchCompatibilityMethodV1::Post,
385            "/logs/_search",
386            "logs"
387        ));
388        assert!(supported_route(OpenSearchCompatibilityMethodV1::Post, "/_bulk", "logs"));
389        assert!(!supported_route(
390            OpenSearchCompatibilityMethodV1::Post,
391            "/other/_search",
392            "logs"
393        ));
394        assert!(!supported_route(
395            OpenSearchCompatibilityMethodV1::Post,
396            "/_plugins/_security/api/roles",
397            "logs"
398        ));
399        assert!(!supported_route(
400            OpenSearchCompatibilityMethodV1::Post,
401            "/_reindex",
402            "logs"
403        ));
404    }
405
406    #[test]
407    fn mutation_classification_never_turns_response_loss_into_a_retry() {
408        assert!(request_is_mutation(
409            OpenSearchCompatibilityMethodV1::Post,
410            "/logs/_update/1"
411        ));
412        assert!(!request_is_mutation(
413            OpenSearchCompatibilityMethodV1::Post,
414            "/logs/_search"
415        ));
416    }
417}
lodestone_client/opensearch_compatibility.rs

lodestone_client/
opensearch_compatibility.rs
