feat(elasticsearch): support custom CA and mutual TLS

crates/dbx-core/src/connection.rs

@@ -2468,7 +2468,10 @@ impl AppState {
                     db_config.url_params.as_deref(),
                     db_config.external_config.as_ref(),
                     connect_timeout,
-                );
+                    Some(db_config.ca_cert_path.as_str()),
+                    Some(db_config.client_cert_path.as_str()),
+                    Some(db_config.client_key_path.as_str()),
+                )?;
                 db::elasticsearch_driver::test_connection(&mut client, connect_timeout).await?;
                 PoolKind::Elasticsearch(client)
             }
@@ -2481,7 +2484,10 @@ impl AppState {
                     db_config.url_params.as_deref(),
                     db_config.external_config.as_ref(),
                     connect_timeout,
-                );
+                    Some(db_config.ca_cert_path.as_str()),
+                    Some(db_config.client_cert_path.as_str()),
+                    Some(db_config.client_key_path.as_str()),
+                )?;
                 db::easysearch_driver::test_connection(&mut client, connect_timeout).await?;
                 PoolKind::Easysearch(client)
             }

crates/dbx-core/src/db/easysearch_driver.rs

@@ -20,10 +20,24 @@ impl EasysearchClient {
         url_params: Option<&str>,
         external_config: Option<&Value>,
         timeout: Duration,
-    ) -> Self {
-        Self {
-            inner: EsClient::from_config(url, username, password, tls_enabled, url_params, external_config, timeout),
-        }
+        ca_cert_path: Option<&str>,
+        client_cert_path: Option<&str>,
+        client_key_path: Option<&str>,
+    ) -> Result<Self, String> {
+        Ok(Self {
+            inner: EsClient::from_config(
+                url,
+                username,
+                password,
+                tls_enabled,
+                url_params,
+                external_config,
+                timeout,
+                ca_cert_path,
+                client_cert_path,
+                client_key_path,
+            )?,
+        })
     }
 }
 

crates/dbx-core/src/db/elasticsearch_driver.rs

@@ -10,7 +10,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
-use super::{http_client_builder, with_connection_timeout};
+use super::{apply_tls_certificates, http_client_builder, with_connection_timeout};
 use crate::db::document_result::DocumentQueryResult;
 use crate::types::QueryResult;
 
@@ -83,9 +83,14 @@ impl EsClient {
             "/".to_string(),
             false,
             None,
+            None,
+            None,
+            None,
         )
+        .expect("failed to build Elasticsearch HTTP client")
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn new_with_mode(
         url: &str,
         username: Option<&str>,
@@ -96,19 +101,24 @@ impl EsClient {
         connectivity_check_path: String,
         connectivity_check_disabled: bool,
         index_grouping: Option<Regex>,
-    ) -> Self {
+        ca_cert_path: Option<&str>,
+        client_cert_path: Option<&str>,
+        client_key_path: Option<&str>,
+    ) -> Result<Self, String> {
         let base_url = url.trim_end_matches('/').to_string();
         let auth = match (username, password) {
             (Some(u), Some(p)) if !u.is_empty() => Some((u.to_string(), p.to_string())),
             _ => None,
         };
         let mut builder = http_client_builder(timeout).danger_accept_invalid_certs(accept_invalid_certs);
+        builder = apply_tls_certificates(builder, ca_cert_path, client_cert_path, client_key_path)?;
         if let Some(addrs) = elasticsearch_localhost_resolve_addrs(&base_url, connectivity_check_disabled) {
             builder = builder.resolve_to_addrs("localhost", &addrs);
         }
-        let http = builder.build().unwrap_or_else(|_| HttpClient::new());
+        let http =
+            builder.build().map_err(|error| format!("Failed to initialize Elasticsearch HTTP client: {error}"))?;
         let fallback_base_urls = elasticsearch_base_url_fallbacks(&base_url);
-        Self {
+        Ok(Self {
             http,
             base_url,
             fallback_base_urls,
@@ -118,7 +128,7 @@ impl EsClient {
             connectivity_check_disabled,
             index_grouping,
             pit_search_supported: Arc::new(AtomicBool::new(true)),
-        }
+        })
     }
 
     pub fn from_config(
@@ -129,7 +139,10 @@ impl EsClient {
         url_params: Option<&str>,
         external_config: Option<&Value>,
         timeout: Duration,
-    ) -> Self {
+        ca_cert_path: Option<&str>,
+        client_cert_path: Option<&str>,
+        client_key_path: Option<&str>,
+    ) -> Result<Self, String> {
         let kibana_base_path = elasticsearch_kibana_base_path(external_config);
         let transport_mode = if kibana_base_path.is_some() {
             ElasticsearchTransportMode::KibanaProxy
@@ -150,6 +163,9 @@ impl EsClient {
             connectivity_check_path,
             connectivity_check_disabled,
             index_grouping,
+            ca_cert_path,
+            client_cert_path,
+            client_key_path,
         )
     }
 
@@ -3536,7 +3552,11 @@ mod tests {
             None,
             Some(&json!({ "connectivityCheckDisabled": "yes" })),
             Duration::from_secs(2),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
         super::test_connection(&mut client, Duration::from_secs(2)).await.unwrap();
         let response = client.get("/_cluster/health").send().await.unwrap();
 
@@ -3554,7 +3574,11 @@ mod tests {
             Some("sslmode=disable"),
             None,
             Duration::from_secs(1),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
 
         assert_eq!(client.base_url, "https://localhost:9200");
         assert_eq!(client.fallback_base_urls, vec!["https://127.0.0.1:9200"]);
@@ -3589,7 +3613,11 @@ mod tests {
                 "connectivityCheckPath": "GET pro-logs-*/_search"
             })),
             Duration::from_secs(1),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
         assert_eq!(client.connectivity_check_path, "/pro-logs-*/_search");
     }
 
@@ -3622,7 +3650,11 @@ mod tests {
             None,
             Some(&json!({ "connectivityCheckPath": "/pro-logs-*/_search" })),
             Duration::from_secs(2),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
         super::test_connection(&mut client, Duration::from_secs(2)).await.unwrap();
         server.await.unwrap();
     }
@@ -3638,7 +3670,11 @@ mod tests {
             None,
             Some(&external_config),
             Duration::from_secs(1),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
 
         assert_eq!(client.base_url, "https://localhost:5601/kibana/s/analytics");
         assert_eq!(client.fallback_base_urls, vec!["https://127.0.0.1:5601/kibana/s/analytics"]);
@@ -5146,7 +5182,11 @@ mod tests {
             None,
             Some(&external_config),
             Duration::from_secs(1),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
         let result =
             super::execute_rest_query(&client, "DELETE /missing/_doc/1?refresh=true\n{\"reason\":\"cleanup\"}")
                 .await
@@ -5499,7 +5539,11 @@ mod tests {
             None,
             Some(&external_config),
             Duration::from_secs(20),
-        );
+            None,
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch test client");
         super::test_connection(&mut client, Duration::from_secs(20)).await.unwrap();
 
         let index = format!("dbx-kibana-proxy-{}", uuid::Uuid::new_v4().simple());
@@ -5590,4 +5634,48 @@ mod tests {
 
         assert_eq!(serde_json::to_string(&doc).unwrap(), r#"{"z":1,"a":2}"#);
     }
+
+    // Integration test against a real TLS-enabled Elasticsearch. Gated by env vars
+    // so it stays a no-op in CI:
+    //   DBX_ELASTICSEARCH_TLS_TEST_URL            e.g. https://localhost:9200
+    //   DBX_ELASTICSEARCH_TLS_TEST_CA_CERT_PATH   path to the server's CA cert (PEM)
+    // Mirrors the ZooKeeper TLS integration test: tls_enabled=true loads the custom
+    // CA (and any client cert) through apply_tls_certificates, then connects over HTTPS.
+    #[tokio::test]
+    async fn tls_integration_connects_with_custom_ca() {
+        let url = match std::env::var("DBX_ELASTICSEARCH_TLS_TEST_URL") {
+            Ok(u) => u,
+            Err(_) => {
+                eprintln!("skipping ES TLS integration test: DBX_ELASTICSEARCH_TLS_TEST_URL not set");
+                return;
+            }
+        };
+        let ca_cert_path = match std::env::var("DBX_ELASTICSEARCH_TLS_TEST_CA_CERT_PATH") {
+            Ok(c) => c,
+            Err(_) => {
+                eprintln!("skipping ES TLS integration test: DBX_ELASTICSEARCH_TLS_TEST_CA_CERT_PATH not set");
+                return;
+            }
+        };
+        // Optional basic-auth credentials (required when the server has security enabled).
+        let username = std::env::var("DBX_ELASTICSEARCH_TLS_TEST_USERNAME").ok();
+        let password = std::env::var("DBX_ELASTICSEARCH_TLS_TEST_PASSWORD").ok();
+        let timeout = std::time::Duration::from_secs(15);
+        let mut client = EsClient::from_config(
+            &url,
+            username.as_deref(),
+            password.as_deref(),
+            true,
+            None,
+            None,
+            timeout,
+            Some(&ca_cert_path),
+            None,
+            None,
+        )
+        .expect("failed to build Elasticsearch TLS client");
+        super::test_connection(&mut client, timeout)
+            .await
+            .expect("Elasticsearch TLS connection with custom CA should succeed");
+    }
 }

crates/dbx-core/src/db/mod.rs

@@ -54,7 +54,9 @@ pub mod vector_driver;
 pub mod victoriametrics_driver;
 pub mod wkb;
 
+use crate::path_utils::expand_tilde;
 use reqwest::ClientBuilder;
+use reqwest::{Certificate, Identity};
 use std::fmt;
 use std::future::Future;
 use std::time::Duration;
@@ -131,6 +133,64 @@ pub fn http_client_builder(timeout: Duration) -> ClientBuilder {
     reqwest::Client::builder().connect_timeout(timeout).no_proxy()
 }
 
+/// Augments an HTTP client builder with custom TLS material for drivers that
+/// need to trust a non-system CA or present a client certificate (mTLS).
+///
+/// * `ca_cert_path` adds an extra root certificate (PEM bundle) alongside the
+///   system roots, so self-signed server certificates are trusted.
+/// * `client_cert_path` + `client_key_path` (required together) are concatenated
+///   into a single PEM identity, as expected by reqwest's rustls backend.
+///
+/// Returns the augmented builder, or a descriptive error if a configured
+/// certificate/key file cannot be read or parsed. Callers surface that error
+/// instead of silently falling back to the system trust store, so a typo in a
+/// cert path fails the connection loudly rather than being ignored.
+pub(crate) fn apply_tls_certificates(
+    mut builder: ClientBuilder,
+    ca_cert_path: Option<&str>,
+    client_cert_path: Option<&str>,
+    client_key_path: Option<&str>,
+) -> Result<ClientBuilder, String> {
+    // The desktop build unifies reqwest with both `default-tls` and `rustls-tls`,
+    // and `TlsBackend` defaults to native-tls there, which rejects the PEM identity
+    // below with `incompatible TLS identity type`. Pin rustls like the Consul and
+    // etcd-metrics clients so custom CA roots and PEM client identities work in
+    // every target.
+    builder = builder.use_rustls_tls();
+    if let Some(ca) = ca_cert_path.filter(|path| !path.is_empty()) {
+        let ca = expand_tilde(ca);
+        let contents = std::fs::read(&ca)
+            .map_err(|error| format!("Failed to read Elasticsearch CA certificate at {ca}: {error}"))?;
+        let certificates = Certificate::from_pem_bundle(&contents)
+            .map_err(|error| format!("Failed to parse Elasticsearch CA certificate at {ca}: {error}"))?;
+        for certificate in certificates {
+            builder = builder.add_root_certificate(certificate);
+        }
+    }
+    if let (Some(cert_path), Some(key_path)) =
+        (client_cert_path.filter(|path| !path.is_empty()), client_key_path.filter(|path| !path.is_empty()))
+    {
+        let cert_path = expand_tilde(cert_path);
+        let key_path = expand_tilde(key_path);
+        let mut pem = std::fs::read(&cert_path)
+            .map_err(|error| format!("Failed to read Elasticsearch client certificate at {cert_path}: {error}"))?;
+        // reqwest's rustls backend concatenates the cert and key as two PEM
+        // blocks; ensure they are newline-separated so parsing succeeds even
+        // when the cert file does not end with a trailing newline.
+        if !pem.ends_with(b"\n") {
+            pem.push(b'\n');
+        }
+        pem.extend(
+            std::fs::read(&key_path)
+                .map_err(|error| format!("Failed to read Elasticsearch client key at {key_path}: {error}"))?,
+        );
+        let identity = Identity::from_pem(&pem)
+            .map_err(|error| format!("Failed to parse Elasticsearch client identity: {error}"))?;
+        builder = builder.identity(identity);
+    }
+    Ok(builder)
+}
+
 pub(crate) const JS_MAX_SAFE_INTEGER: i64 = 9_007_199_254_740_991;
 
 pub fn safe_i64_to_json(v: i64) -> serde_json::Value {
@@ -294,3 +354,106 @@ mod tests {
         assert_eq!(json_value_for_js(value), expected);
     }
 }
+
+#[cfg(test)]
+mod tls_tests {
+    use super::{apply_tls_certificates, http_client_builder};
+    use std::path::Path;
+    use std::process::Command;
+    use std::time::Duration;
+
+    /// Generates a throwaway CA + client certificate/key (PEM) via openssl and
+    /// returns their paths. If openssl is unavailable the files stay empty, which
+    /// makes `apply_tls_certificates` surface a read error instead of degrading.
+    fn gen_certs(dir: &Path) -> (std::path::PathBuf, std::path::PathBuf, std::path::PathBuf) {
+        let ca_key = dir.join("ca.key");
+        let ca_pem = dir.join("ca.pem");
+        let csr = dir.join("client.csr");
+        let client_key = dir.join("client.key");
+        let client_crt = dir.join("client.crt");
+        // Extensions force an X.509 v3 client cert; some toolchains (e.g.
+        // OpenSSL 3.0 on CI) emit v1 without one, which rustls rejects.
+        let client_ext = dir.join("client.ext");
+        let _ = std::fs::write(&client_ext, "basicConstraints=critical,CA:FALSE\n");
+        let _ = Command::new("openssl")
+            .args([
+                "req",
+                "-x509",
+                "-newkey",
+                "rsa:2048",
+                "-nodes",
+                "-keyout",
+                ca_key.to_str().unwrap(),
+                "-out",
+                ca_pem.to_str().unwrap(),
+                "-days",
+                "1",
+                "-subj",
+                "/CN=dbx-test-ca",
+            ])
+            .output();
+        let _ = Command::new("openssl")
+            .args([
+                "req",
+                "-newkey",
+                "rsa:2048",
+                "-nodes",
+                "-keyout",
+                client_key.to_str().unwrap(),
+                "-out",
+                csr.to_str().unwrap(),
+                "-subj",
+                "/CN=dbx-test-client",
+            ])
+            .output();
+        let _ = Command::new("openssl")
+            .args([
+                "x509",
+                "-req",
+                "-in",
+                csr.to_str().unwrap(),
+                "-CA",
+                ca_pem.to_str().unwrap(),
+                "-CAkey",
+                ca_key.to_str().unwrap(),
+                "-CAcreateserial",
+                "-out",
+                client_crt.to_str().unwrap(),
+                "-days",
+                "1",
+                "-extfile",
+                client_ext.to_str().unwrap(),
+            ])
+            .output();
+        (ca_pem, client_crt, client_key)
+    }
+
+    #[test]
+    fn apply_tls_certificates_builds_with_ca_and_client_cert() {
+        let dir = std::env::temp_dir().join(format!("dbx-tls-test-{}", std::process::id()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let (ca, client_crt, client_key) = gen_certs(&dir);
+
+        let built = apply_tls_certificates(
+            http_client_builder(Duration::from_secs(5)),
+            ca.to_str(),
+            client_crt.to_str(),
+            client_key.to_str(),
+        )
+        .expect("client should build with CA + client cert")
+        .build();
+        assert!(built.is_ok(), "built client should be usable with CA + client cert: {:?}", built.err());
+
+        // A misconfigured CA path must now surface a descriptive error instead of
+        // silently degrading to the system trust store.
+        let missing = apply_tls_certificates(
+            http_client_builder(Duration::from_secs(5)),
+            Some("/nonexistent/ca.pem"),
+            None,
+            None,
+        );
+        assert!(missing.is_err(), "missing CA path must surface a read error");
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}