mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 06:00:39 +00:00
feat(d1): support https proxy and custom ca for d1 and turso
This commit is contained in:
@@ -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) });
|
||||
}
|
||||
|
||||
@@ -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"] }
|
||||
|
||||
@@ -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<ClientBuilder, String> {
|
||||
// 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<ClientBuilder, String> {
|
||||
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<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
|
||||
@@ -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<String>, Option<String>) {
|
||||
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<String> {
|
||||
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<Option<reqwest::Proxy>, 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<reqwest::Proxy>,
|
||||
proxy_url: Option<&str>,
|
||||
no_proxy: Option<&str>,
|
||||
) -> Result<ClientBuilder, String> {
|
||||
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<String>, Option<String>) {
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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, String> {
|
||||
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<reqwest::Proxy>,
|
||||
) -> Result<Self, String> {
|
||||
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<Self, String> {
|
||||
pub fn with_endpoint(endpoint: String, api_token: &str, timeout: Duration) -> Result<Self, String> {
|
||||
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<reqwest::Proxy>,
|
||||
explicit_proxy_url: Option<&str>,
|
||||
no_proxy: Option<&str>,
|
||||
) -> Result<Self, String> {
|
||||
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<Option<reqwest::Proxy>, 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<CloudflareD1Client, String> {
|
||||
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<reqwest::Proxy>,
|
||||
timeout: Duration,
|
||||
) -> Result<CloudflareD1Client, String> {
|
||||
let client = client_from_config(config, proxy, timeout)?;
|
||||
test_connection(&client, timeout).await?;
|
||||
Ok(client)
|
||||
}
|
||||
|
||||
pub fn client_from_config(
|
||||
config: &ConnectionConfig,
|
||||
proxy: Option<reqwest::Proxy>,
|
||||
timeout: Duration,
|
||||
) -> Result<CloudflareD1Client, String> {
|
||||
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<Vec<DatabaseInfo>, 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();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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, String> {
|
||||
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<reqwest::Proxy>,
|
||||
) -> Result<Self, String> {
|
||||
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"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user