From 00cbb87c5dff77a880cfff3e248b0e06a7ce8cfb Mon Sep 17 00:00:00 2001 From: lxk955 <164191365+lxk955@users.noreply.github.com> Date: Sat, 17 Oct 2026 20:18:34 +0800 Subject: [PATCH] feat(d1): support https proxy and custom ca for d1 and turso --- crates/dbx-core/src/connection/mod.rs | 34 +- crates/dbx-driver-support/Cargo.toml | 2 +- crates/dbx-driver-support/src/db.rs | 303 +++++++++++++++++- .../dbx-drivers/src/db/cloudflare_d1/mod.rs | 232 +++++++++++++- crates/dbx-drivers/src/db/turso_driver.rs | 38 ++- src-tauri/src/commands/connection.rs | 30 +- 6 files changed, 605 insertions(+), 34 deletions(-) diff --git a/crates/dbx-core/src/connection/mod.rs b/crates/dbx-core/src/connection/mod.rs index cdbe47e59c..f7e06031f0 100644 --- a/crates/dbx-core/src/connection/mod.rs +++ b/crates/dbx-core/src/connection/mod.rs @@ -2919,12 +2919,37 @@ impl AppState { } else { db_config.url_params.as_deref().and_then(extract_auth_token_from_params).unwrap_or_default() }; - let client = db::turso_driver::TursoClient::new(&url, &auth_token, db_config.ssl, connect_timeout)?; + let http_proxy = clickhouse_http_proxy(runtime_proxy.as_ref())?; + let ca_cert = + if db_config.ca_cert_path.trim().is_empty() { None } else { Some(db_config.ca_cert_path.as_str()) }; + let client_cert = if db_config.client_cert_path.trim().is_empty() { + None + } else { + Some(db_config.client_cert_path.as_str()) + }; + let client_key = if db_config.client_key_path.trim().is_empty() { + None + } else { + Some(db_config.client_key_path.as_str()) + }; + let client = db::turso_driver::TursoClient::new_with_options( + &url, + &auth_token, + db_config.ssl, + connect_timeout, + ca_cert, + client_cert, + client_key, + http_proxy, + )?; db::turso_driver::test_connection(&client, connect_timeout).await?; PoolKind::Turso(client) } DatabaseType::CloudflareD1 => { - PoolKind::CloudflareD1(db::cloudflare_d1_driver::connect(&db_config, connect_timeout).await?) + let http_proxy = clickhouse_http_proxy(runtime_proxy.as_ref())?; + PoolKind::CloudflareD1( + db::cloudflare_d1_driver::connect_with_proxy(&db_config, http_proxy, connect_timeout).await?, + ) } DatabaseType::Redis => { let con = if db_config.uses_redis_cluster() { @@ -3688,7 +3713,10 @@ impl AppState { if transport_layers.is_empty() || db::sqlite_worker::sqlite_remote_worker_requested(config) { return Ok(ConnectionEndpoint::direct(config.host.clone(), config.port)); } - if config.db_type == DatabaseType::ClickHouse { + if config.db_type == DatabaseType::ClickHouse + || config.db_type == DatabaseType::CloudflareD1 + || config.db_type == DatabaseType::Turso + { if let Some(proxy) = self.clickhouse_proxy_for_transport_layers(connection_id, &transport_layers).await? { return Ok(ConnectionEndpoint { host: config.host.clone(), port: config.port, proxy: Some(proxy) }); } diff --git a/crates/dbx-driver-support/Cargo.toml b/crates/dbx-driver-support/Cargo.toml index e8959339b0..e2c3bb2507 100644 --- a/crates/dbx-driver-support/Cargo.toml +++ b/crates/dbx-driver-support/Cargo.toml @@ -15,7 +15,7 @@ dbx-types = { path = "../dbx-types", default-features = false } bytes = "1" glob = "0.3" portpicker = "0.1.1" -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "socks"] } russh = "0.60" serde = { version = "1.0", features = ["derive"] } serde_json = { version = "1.0", features = ["arbitrary_precision", "preserve_order"] } diff --git a/crates/dbx-driver-support/src/db.rs b/crates/dbx-driver-support/src/db.rs index 0b65c5b548..5b33d3b2ba 100644 --- a/crates/dbx-driver-support/src/db.rs +++ b/crates/dbx-driver-support/src/db.rs @@ -89,10 +89,42 @@ pub fn http_client_builder(timeout: Duration) -> ClientBuilder { /// 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 fn apply_tls_certificates( + builder: ClientBuilder, + ca_cert_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, +) -> Result { + // No SSL_CERT_FILE environment fallback here: Elasticsearch/Solr/CouchDB keep + // their historical behavior (bundled roots only when no CA is configured). + apply_tls_certificates_impl(builder, ca_cert_path, client_cert_path, client_key_path, "Elasticsearch", false) +} + +/// Augments an HTTP client builder with custom TLS material for a specific driver. +/// +/// * `ca_cert_path`: adds an extra root certificate (PEM bundle or DER). If not provided +/// or empty, falls back to `SSL_CERT_FILE` from the environment if present and pointing +/// to an existing file. This environment fallback applies ONLY to drivers that call +/// this function (Cloudflare D1, Turso), never to the `apply_tls_certificates` +/// wrapper above. +/// * `client_cert_path` + `client_key_path`: concatenated into a single PEM identity. +/// * `driver_label`: included in descriptive error messages (e.g. "Cloudflare D1", "Turso"). +pub fn apply_tls_certificates_for_driver( + builder: ClientBuilder, + ca_cert_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, + driver_label: &str, +) -> Result { + apply_tls_certificates_impl(builder, ca_cert_path, client_cert_path, client_key_path, driver_label, true) +} + +fn apply_tls_certificates_impl( mut builder: ClientBuilder, ca_cert_path: Option<&str>, client_cert_path: Option<&str>, client_key_path: Option<&str>, + driver_label: &str, + env_ca_fallback: bool, ) -> Result { // The desktop build unifies reqwest with both `default-tls` and `rustls-tls`, // and `TlsBackend` defaults to native-tls there, which rejects the PEM identity @@ -100,40 +132,190 @@ pub fn apply_tls_certificates( // 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 explicit_ca = ca_cert_path.map(str::trim).filter(|path| !path.is_empty()).map(ToOwned::to_owned); + let env_ca = if env_ca_fallback { + std::env::var("SSL_CERT_FILE") + .ok() + .map(|s| s.trim().to_string()) + .filter(|path| !path.is_empty() && std::path::Path::new(&expand_tilde(path)).exists()) + } else { + None + }; + let effective_ca = explicit_ca.or(env_ca); + + if let Some(ca) = effective_ca { + let ca = expand_tilde(&ca); let contents = std::fs::read(&ca) - .map_err(|error| format!("Failed to read Elasticsearch CA certificate at {ca}: {error}"))?; + .map_err(|error| format!("Failed to read {driver_label} 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}"))?; + .or_else(|_| Certificate::from_der(&contents).map(|cert| vec![cert])) + .map_err(|error| format!("Failed to parse {driver_label} 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())) - { + if let (Some(cert_path), Some(key_path)) = ( + client_cert_path.map(str::trim).filter(|path| !path.is_empty()), + client_key_path.map(str::trim).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. + .map_err(|error| format!("Failed to read {driver_label} client certificate at {cert_path}: {error}"))?; 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}"))?, + .map_err(|error| format!("Failed to read {driver_label} client key at {key_path}: {error}"))?, ); let identity = Identity::from_pem(&pem) - .map_err(|error| format!("Failed to parse Elasticsearch client identity: {error}"))?; + .map_err(|error| format!("Failed to parse {driver_label} client identity: {error}"))?; builder = builder.identity(identity); } Ok(builder) } +/// Extracts the lowercased URL scheme from a target URL or endpoint (defaults to "https" if unspecified). +pub fn normalize_url_scheme(url: &str) -> String { + let trimmed = url.trim(); + if let Some((scheme, _)) = trimmed.split_once("://") { + scheme.to_ascii_lowercase() + } else { + "https".to_string() + } +} + +/// Resolves standard environment proxy settings (HTTPS_PROXY/HTTP_PROXY/ALL_PROXY and NO_PROXY). +pub fn standard_env_proxy(scheme: &str) -> (Option, Option) { + let http_proxy = first_non_empty_env(&["HTTP_PROXY", "http_proxy"]); + let https_proxy = first_non_empty_env(&["HTTPS_PROXY", "https_proxy"]); + let all_proxy = first_non_empty_env(&["ALL_PROXY", "all_proxy"]); + let proxy_url = match scheme { + "https" => https_proxy.or(all_proxy), + "http" => http_proxy.or(all_proxy), + _ => https_proxy.or(http_proxy).or(all_proxy), + }; + let no_proxy = first_non_empty_env(&["NO_PROXY", "no_proxy"]); + (proxy_url, no_proxy) +} + +fn first_non_empty_env(names: &[&str]) -> Option { + names.iter().find_map(|name| { + std::env::var(name).ok().map(|value| value.trim().to_string()).filter(|value| !value.is_empty()) + }) +} + +/// Resolves an HTTP/HTTPS/SOCKS proxy for a given target URL. +/// +/// Priority: +/// 1. `explicit_proxy` (if provided): +/// - "direct", "none", "off", "false": forces direct connection (no proxy). +/// - "env", "auto", "system": forces lookup from environment variables. +/// - Any other string: treated as proxy URL (e.g. "http://proxy:8080", "socks5://proxy:1080"). +/// 2. If no explicit proxy: +/// - Checks `DBX_HTTP_PROXY_DISABLE`: if set to true/1/yes, skips proxy. +/// - Resolves from environment (`HTTPS_PROXY` / `HTTP_PROXY` / `ALL_PROXY`). +/// - Attaches `NO_PROXY` bypass rules if present. +pub fn resolve_http_proxy( + target_url: &str, + explicit_proxy: Option<&str>, + no_proxy: Option<&str>, +) -> Result, String> { + if let Some(explicit) = explicit_proxy.map(str::trim).filter(|s| !s.is_empty()) { + if matches!(explicit.to_ascii_lowercase().as_str(), "direct" | "none" | "off" | "false") { + return Ok(None); + } + if !matches!(explicit.to_ascii_lowercase().as_str(), "env" | "auto" | "system") { + let mut proxy = reqwest::Proxy::all(explicit) + .map_err(|e| format!("Invalid proxy URL '{}': {e}", redact_proxy_userinfo(explicit)))?; + if let Some(no_proxy_val) = no_proxy.map(str::trim).filter(|s| !s.is_empty()) { + proxy = proxy.no_proxy(reqwest::NoProxy::from_string(no_proxy_val)); + } + return Ok(Some(proxy)); + } + } + + if let Ok(val) = std::env::var("DBX_HTTP_PROXY_DISABLE") { + if matches!(val.trim().to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on") { + return Ok(None); + } + } + + let scheme = normalize_url_scheme(target_url); + let (proxy_url, env_no_proxy) = standard_env_proxy(&scheme); + let effective_no_proxy = no_proxy.map(str::trim).filter(|s| !s.is_empty()).or(env_no_proxy.as_deref()); + + if let Some(url) = proxy_url { + let mut proxy = reqwest::Proxy::all(&url) + .map_err(|e| format!("Invalid environment proxy URL '{}': {e}", redact_proxy_userinfo(&url)))?; + if let Some(no_proxy_val) = effective_no_proxy { + proxy = proxy.no_proxy(reqwest::NoProxy::from_string(no_proxy_val)); + } + Ok(Some(proxy)) + } else { + Ok(None) + } +} + +/// Redacts userinfo (credentials) from a proxy URL so error messages never leak +/// them into logs: `http://user:pass@proxy:8080` -> `http://proxy:8080`. +fn redact_proxy_userinfo(url: &str) -> String { + let (prefix, rest) = match url.split_once("://") { + Some((scheme, rest)) => (format!("{scheme}://"), rest), + None => (String::new(), url), + }; + let (authority, path) = match rest.find('/') { + Some(idx) => rest.split_at(idx), + None => (rest, ""), + }; + match authority.rfind('@') { + Some(at) => format!("{prefix}{}{path}", &authority[at + 1..]), + None => url.to_string(), + } +} + +/// Applies proxy configuration to a `ClientBuilder`. +/// +/// If `explicit_proxy` is provided, it is attached directly. +/// Otherwise, resolves the proxy from `proxy_url` (or environment) and `no_proxy`. +/// If no proxy is applicable, preserves direct connection (`no_proxy()`). +pub fn apply_http_proxy( + builder: ClientBuilder, + target_url: &str, + explicit_proxy: Option, + proxy_url: Option<&str>, + no_proxy: Option<&str>, +) -> Result { + if let Some(proxy) = explicit_proxy { + return Ok(builder.proxy(proxy)); + } + if let Some(proxy) = resolve_http_proxy(target_url, proxy_url, no_proxy)? { + return Ok(builder.proxy(proxy)); + } + Ok(builder) +} + +/// Parses `proxy` and `no_proxy` options from URL query parameters. +pub fn parse_proxy_url_params(params: Option<&str>) -> (Option, Option) { + let mut proxy = None; + let mut no_proxy = None; + if let Some(params) = params { + for part in params.trim().trim_start_matches(['?', '&', ';']).split(['&', ';']) { + if let Some((k, v)) = part.split_once('=') { + let k = k.trim().to_ascii_lowercase(); + let v = v.trim().to_string(); + if k == "proxy" || k == "http_proxy" || k == "https_proxy" { + proxy = Some(v); + } else if k == "no_proxy" { + no_proxy = Some(v); + } + } + } + } + (proxy, no_proxy) +} + pub use dbx_types::json::JS_MAX_SAFE_INTEGER; pub use dbx_types::json::{json_value_for_js, safe_i64_to_json, safe_u64_to_json}; @@ -216,7 +398,10 @@ mod tests { #[cfg(test)] mod tls_tests { - use super::{apply_tls_certificates, http_client_builder}; + use super::{ + apply_tls_certificates, apply_tls_certificates_for_driver, http_client_builder, parse_proxy_url_params, + redact_proxy_userinfo, resolve_http_proxy, + }; use std::path::Path; use std::process::Command; use std::time::Duration; @@ -315,4 +500,94 @@ mod tls_tests { let _ = std::fs::remove_dir_all(&dir); } + + #[test] + fn apply_tls_certificates_for_driver_formats_custom_label() { + let err = apply_tls_certificates_for_driver( + http_client_builder(Duration::from_secs(5)), + Some("/nonexistent/custom-ca.pem"), + None, + None, + "Cloudflare D1", + ) + .err() + .unwrap(); + assert!(err.contains("Failed to read Cloudflare D1 CA certificate at /nonexistent/custom-ca.pem")); + } + + /// t8y2 review (PR #12534): the `SSL_CERT_FILE` environment fallback must stay + /// scoped to the per-driver helper — the legacy `apply_tls_certificates` + /// wrapper (Elasticsearch/Solr/CouchDB) keeps its bundled-roots-only behavior + /// when no CA is configured. + #[test] + fn ssl_cert_file_env_fallback_is_driver_scoped() { + let dir = std::env::temp_dir().join(format!("dbx-tls-env-test-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + // Point SSL_CERT_FILE at a directory: it exists(), so the env fallback + // picks it up, but reading it fails — proving the fallback was honored. + let dir_ca = dir.join("ca-dir"); + std::fs::create_dir_all(&dir_ca).unwrap(); + std::env::set_var("SSL_CERT_FILE", &dir_ca); + + // The per-driver variant honors the environment fallback: the directory + // is "read" and fails — proving SSL_CERT_FILE was picked up. + let err = apply_tls_certificates_for_driver( + http_client_builder(Duration::from_secs(5)), + None, + None, + None, + "Cloudflare D1", + ) + .expect_err("driver-scoped helper must honor SSL_CERT_FILE"); + assert!(err.contains("Failed to read Cloudflare D1 CA certificate"), "unexpected error: {err}"); + + // The legacy wrapper must ignore SSL_CERT_FILE entirely. + let ok = apply_tls_certificates(http_client_builder(Duration::from_secs(5)), None, None, None); + assert!(ok.is_ok(), "legacy wrapper must ignore SSL_CERT_FILE"); + + std::env::remove_var("SSL_CERT_FILE"); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn redact_proxy_userinfo_strips_credentials() { + assert_eq!(redact_proxy_userinfo("http://user:pass@proxy:8080"), "http://proxy:8080"); + assert_eq!(redact_proxy_userinfo("socks5://u:p@[::1]:1080/path"), "socks5://[::1]:1080/path"); + assert_eq!(redact_proxy_userinfo("http://proxy:8080"), "http://proxy:8080"); + assert_eq!(redact_proxy_userinfo("user@host:8080"), "host:8080"); + assert_eq!(redact_proxy_userinfo(""), ""); + } + + #[test] + fn parse_proxy_url_params_extracts_proxy_and_no_proxy() { + let (proxy, no_proxy) = + parse_proxy_url_params(Some("?proxy=http://127.0.0.1:8080&no_proxy=localhost,127.0.0.1")); + assert_eq!(proxy.as_deref(), Some("http://127.0.0.1:8080")); + assert_eq!(no_proxy.as_deref(), Some("localhost,127.0.0.1")); + + let (proxy, no_proxy) = parse_proxy_url_params(Some("proxy=direct;no_proxy=.internal")); + assert_eq!(proxy.as_deref(), Some("direct")); + assert_eq!(no_proxy.as_deref(), Some(".internal")); + + let (proxy, no_proxy) = parse_proxy_url_params(None); + assert_eq!(proxy, None); + assert_eq!(no_proxy, None); + } + + #[test] + fn resolve_http_proxy_honors_explicit_direct_directive() { + let result = resolve_http_proxy("https://api.cloudflare.com/raw", Some("direct"), None).unwrap(); + assert!(result.is_none()); + + let result = resolve_http_proxy("https://api.cloudflare.com/raw", Some("none"), None).unwrap(); + assert!(result.is_none()); + } + + #[test] + fn resolve_http_proxy_parses_explicit_url() { + let proxy = + resolve_http_proxy("https://api.cloudflare.com/raw", Some("http://proxy.test:8080"), Some("localhost")) + .unwrap(); + assert!(proxy.is_some()); + } } diff --git a/crates/dbx-drivers/src/db/cloudflare_d1/mod.rs b/crates/dbx-drivers/src/db/cloudflare_d1/mod.rs index 4be3e2560e..e19244df47 100644 --- a/crates/dbx-drivers/src/db/cloudflare_d1/mod.rs +++ b/crates/dbx-drivers/src/db/cloudflare_d1/mod.rs @@ -2,8 +2,11 @@ use reqwest::Client as HttpClient; use serde::Deserialize; use std::time::{Duration, Instant}; -use super::{http_client_builder, json_value_for_js, with_connection_timeout}; -use crate::models::connection::ConnectionConfig; +use super::{ + apply_http_proxy, apply_tls_certificates_for_driver, http_client_builder, json_value_for_js, + parse_proxy_url_params, with_connection_timeout, +}; +use crate::models::connection::{ConnectionConfig, ProxyType, TransportLayerConfig}; use crate::types::{ ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectSource, ObjectSourceKind, QueryResult, TableInfo, TriggerInfo, @@ -27,17 +30,64 @@ pub struct CloudflareD1Client { impl CloudflareD1Client { pub fn new(account_id: &str, database_id: &str, api_token: &str, timeout: Duration) -> Result { + Self::new_with_options(account_id, database_id, api_token, timeout, None, None, None, None) + } + + #[allow(clippy::too_many_arguments)] + pub fn new_with_options( + account_id: &str, + database_id: &str, + api_token: &str, + timeout: Duration, + ca_cert_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, + proxy: Option, + ) -> Result { let account_id = required_identifier(account_id, "Cloudflare Account ID")?; let database_id = required_identifier(database_id, "Cloudflare D1 Database ID")?; let endpoint = format!("{CLOUDFLARE_API_BASE_URL}/accounts/{account_id}/d1/database/{database_id}/raw"); - Self::with_endpoint(endpoint, api_token, timeout) + Self::with_endpoint_and_options( + endpoint, + api_token, + timeout, + ca_cert_path, + client_cert_path, + client_key_path, + proxy, + None, + None, + ) } - fn with_endpoint(endpoint: String, api_token: &str, timeout: Duration) -> Result { + pub fn with_endpoint(endpoint: String, api_token: &str, timeout: Duration) -> Result { + Self::with_endpoint_and_options(endpoint, api_token, timeout, None, None, None, None, None, None) + } + + #[allow(clippy::too_many_arguments)] + pub fn with_endpoint_and_options( + endpoint: String, + api_token: &str, + timeout: Duration, + ca_cert_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, + proxy: Option, + explicit_proxy_url: Option<&str>, + no_proxy: Option<&str>, + ) -> Result { let api_token = required_field(api_token, "Cloudflare API Token")?; - let http = http_client_builder(timeout) - .build() - .map_err(|error| format!("Failed to configure Cloudflare D1 HTTP client: {error}"))?; + let mut builder = http_client_builder(timeout); + builder = apply_http_proxy(builder, &endpoint, proxy, explicit_proxy_url, no_proxy)?; + builder = apply_tls_certificates_for_driver( + builder, + ca_cert_path, + client_cert_path, + client_key_path, + "Cloudflare D1", + )?; + let http = + builder.build().map_err(|error| format!("Failed to configure Cloudflare D1 HTTP client: {error}"))?; Ok(Self { http, endpoint, api_token: api_token.to_string() }) } @@ -50,6 +100,32 @@ impl CloudflareD1Client { } } +pub fn proxy_from_transport_layers(layers: &[TransportLayerConfig]) -> Result, String> { + for layer in layers.iter().rev() { + if let TransportLayerConfig::Proxy(proxy) = layer { + if !proxy.enabled { + continue; + } + let scheme = match proxy.proxy_type { + ProxyType::Http => "http", + ProxyType::Socks5 => "socks5h", + }; + let host = if proxy.host.contains(':') && !proxy.host.starts_with('[') { + format!("[{}]", proxy.host) + } else { + proxy.host.clone() + }; + let mut reqwest_proxy = reqwest::Proxy::all(format!("{scheme}://{host}:{}", proxy.port)) + .map_err(|e| format!("Invalid proxy address: {e}"))?; + if !proxy.username.is_empty() || !proxy.password.is_empty() { + reqwest_proxy = reqwest_proxy.basic_auth(&proxy.username, &proxy.password); + } + return Ok(Some(reqwest_proxy)); + } + } + Ok(None) +} + #[derive(Debug, Deserialize)] struct D1ApiResponse { #[serde(default)] @@ -96,16 +172,52 @@ pub async fn test_connection(client: &CloudflareD1Client, timeout: Duration) -> } pub async fn connect(config: &ConnectionConfig, timeout: Duration) -> Result { - let client = CloudflareD1Client::new( - &config.host, - config.database.as_deref().unwrap_or_default(), - &config.password, - timeout, - )?; + connect_with_proxy(config, None, timeout).await +} + +pub async fn connect_with_proxy( + config: &ConnectionConfig, + proxy: Option, + timeout: Duration, +) -> Result { + let client = client_from_config(config, proxy, timeout)?; test_connection(&client, timeout).await?; Ok(client) } +pub fn client_from_config( + config: &ConnectionConfig, + proxy: Option, + timeout: Duration, +) -> Result { + let proxy = match proxy { + Some(p) => Some(p), + None => proxy_from_transport_layers(&config.transport_layers)?, + }; + let (explicit_proxy_url, explicit_no_proxy) = parse_proxy_url_params(config.url_params.as_deref()); + let ca_cert_path = if config.ca_cert_path.trim().is_empty() { None } else { Some(config.ca_cert_path.as_str()) }; + let client_cert_path = + if config.client_cert_path.trim().is_empty() { None } else { Some(config.client_cert_path.as_str()) }; + let client_key_path = + if config.client_key_path.trim().is_empty() { None } else { Some(config.client_key_path.as_str()) }; + + let account_id = required_identifier(&config.host, "Cloudflare Account ID")?; + let database_id = required_identifier(config.database.as_deref().unwrap_or_default(), "Cloudflare D1 Database ID")?; + let endpoint = format!("{CLOUDFLARE_API_BASE_URL}/accounts/{account_id}/d1/database/{database_id}/raw"); + + CloudflareD1Client::with_endpoint_and_options( + endpoint, + &config.password, + timeout, + ca_cert_path, + client_cert_path, + client_key_path, + proxy, + explicit_proxy_url.as_deref(), + explicit_no_proxy.as_deref(), + ) +} + pub async fn list_databases(_client: &CloudflareD1Client) -> Result, String> { Ok(vec![DatabaseInfo { name: "main".to_string(), ..Default::default() }]) } @@ -649,4 +761,98 @@ mod tests { } String::from_utf8(bytes).unwrap() } + + #[test] + fn validates_ca_cert_path_error() { + let err = CloudflareD1Client::new_with_options( + "account", + "database", + "token", + Duration::from_secs(1), + Some("/nonexistent/ca.pem"), + None, + None, + None, + ) + .err() + .unwrap(); + assert!(err.contains("Failed to read Cloudflare D1 CA certificate at /nonexistent/ca.pem")); + } + + #[test] + fn proxy_from_transport_layers_extracts_http_and_socks5_proxies() { + use crate::models::connection::ProxyTunnelConfig; + + let layer = TransportLayerConfig::Proxy(ProxyTunnelConfig { + id: "p1".to_string(), + profile_id: String::new(), + test_target: None, + name: "test-proxy".to_string(), + enabled: true, + proxy_type: ProxyType::Http, + host: "proxy.example.com".to_string(), + port: 8080, + username: "user".to_string(), + password: "pwd".to_string(), + }); + let proxy = proxy_from_transport_layers(&[layer]).unwrap(); + assert!(proxy.is_some()); + + let disabled_layer = TransportLayerConfig::Proxy(ProxyTunnelConfig { + id: "p2".to_string(), + profile_id: String::new(), + test_target: None, + name: "disabled-proxy".to_string(), + enabled: false, + proxy_type: ProxyType::Socks5, + host: "proxy.example.com".to_string(), + port: 1080, + username: "".to_string(), + password: "".to_string(), + }); + let proxy = proxy_from_transport_layers(&[disabled_layer]).unwrap(); + assert!(proxy.is_none()); + } + + #[tokio::test] + async fn routes_requests_through_configured_http_proxy() { + let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_addr = proxy_listener.local_addr().unwrap(); + + let proxy_handle = tokio::spawn(async move { + let (mut socket, _) = proxy_listener.accept().await.unwrap(); + let request = read_http_request(&mut socket).await; + assert!( + request.contains("POST http://test.d1.internal/raw") + || request.contains("CONNECT test.d1.internal") + || request.contains("POST /raw") + ); + + let body = r#"{"success":true,"errors":[],"result":[{"success":true,"results":{"columns":["proxied"],"rows":[[1]]},"meta":{"changes":0}}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + socket.write_all(response.as_bytes()).await.unwrap(); + }); + + let proxy = reqwest::Proxy::all(format!("http://{proxy_addr}")).unwrap(); + let client = CloudflareD1Client::with_endpoint_and_options( + "http://test.d1.internal/raw".to_string(), + "test-token", + Duration::from_secs(5), + None, + None, + None, + Some(proxy), + None, + None, + ) + .unwrap(); + + let result = execute_query_with_max_rows(&client, "SELECT 1", None).await.unwrap(); + assert_eq!(result.columns, ["proxied"]); + proxy_handle.await.unwrap(); + } } diff --git a/crates/dbx-drivers/src/db/turso_driver.rs b/crates/dbx-drivers/src/db/turso_driver.rs index d8f409cfa1..fd452420e9 100644 --- a/crates/dbx-drivers/src/db/turso_driver.rs +++ b/crates/dbx-drivers/src/db/turso_driver.rs @@ -3,7 +3,10 @@ use reqwest::Client as HttpClient; use serde::Deserialize; use std::time::{Duration, Instant}; -use super::{binary_value_to_json, http_client_builder, with_connection_timeout}; +use super::{ + apply_http_proxy, apply_tls_certificates_for_driver, binary_value_to_json, http_client_builder, + with_connection_timeout, +}; use crate::sql::starts_with_executable_sql_keyword; use crate::types::{ ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectSource, ObjectSourceKind, QueryResult, TableInfo, @@ -19,10 +22,26 @@ pub struct TursoClient { impl TursoClient { pub fn new(url: &str, auth_token: &str, tls_enabled: bool, timeout: Duration) -> Result { + Self::new_with_options(url, auth_token, tls_enabled, timeout, None, None, None, None) + } + + #[allow(clippy::too_many_arguments)] + pub fn new_with_options( + url: &str, + auth_token: &str, + tls_enabled: bool, + timeout: Duration, + ca_cert_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, + proxy: Option, + ) -> Result { let mut builder = http_client_builder(timeout); if turso_accept_invalid_certs(tls_enabled) { builder = builder.danger_accept_invalid_certs(true); } + builder = apply_http_proxy(builder, url, proxy, None, None)?; + builder = apply_tls_certificates_for_driver(builder, ca_cert_path, client_cert_path, client_key_path, "Turso")?; let http = builder.build().map_err(|e| format!("Failed to configure Turso HTTP client: {e}"))?; Ok(Self { http, base_url: url.trim_end_matches('/').to_string(), auth_token: auth_token.to_string() }) } @@ -845,4 +864,21 @@ mod tests { assert_eq!(qr.rows.len(), 2); assert!(!qr.truncated); } + + #[test] + fn validates_turso_ca_cert_path_error() { + let err = TursoClient::new_with_options( + "https://test.turso.io", + "token", + false, + Duration::from_secs(1), + Some("/nonexistent/turso-ca.pem"), + None, + None, + None, + ) + .err() + .unwrap(); + assert!(err.contains("Failed to read Turso CA certificate at /nonexistent/turso-ca.pem")); + } } diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index b87aba9ebf..a6e96285a4 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -1879,7 +1879,20 @@ async fn test_connection_with_info_impl( }) .unwrap_or_default() }; - let client = db::turso_driver::TursoClient::new(&url, &auth_token, config.ssl, connect_timeout)?; + // Route the test through the same transport-layer proxy resolution the + // real connect uses: a test that bypasses the proxy may succeed while + // the real connect fails through the proxy. + let proxy = clickhouse_http_proxy(runtime_proxy.as_ref())?; + let client = db::turso_driver::TursoClient::new_with_options( + &url, + &auth_token, + config.ssl, + connect_timeout, + Some(config.ca_cert_path.as_str()), + Some(config.client_cert_path.as_str()), + Some(config.client_key_path.as_str()), + proxy, + )?; db::turso_driver::test_connection(&client, connect_timeout) .await .map(|_| "Connection successful".to_string()) @@ -2424,7 +2437,20 @@ async fn connect_db_inner( }) .unwrap_or_default() }; - let client = db::turso_driver::TursoClient::new(&url, &auth_token, db_config.ssl, connect_timeout)?; + // Same transport-layer proxy resolution as the test path and the real + // connect path in dbx-core: bypassing the proxy here would diverge from + // both. + let proxy = clickhouse_http_proxy(runtime_proxy.as_ref())?; + let client = db::turso_driver::TursoClient::new_with_options( + &url, + &auth_token, + db_config.ssl, + 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()), + proxy, + )?; db::turso_driver::test_connection(&client, connect_timeout).await?; PoolKind::Turso(client) }