github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-28 16:09:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 't8y2:main' into main

Browse Source
This commit is contained in:
Alexandr
2026-05-11 17:11:12 +04:00
committed by GitHub
parent 23bbfcb157 dddb813fc6
commit ad5e6fcce4
45 changed files with 1573 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/Cargo.toml
+1 -1
View File
@@ -23,7 +23,7 @@ redis = { version = "0.32.2", features = ["tokio-comp", "tls-rustls", "tokio-rus
rustls = { version = "0.23", features = ["aws-lc-rs"] }
duckdb = "1.3.2"
tiberius = { version = "0.12.3", default-features = false, features = ["tds73", "chrono", "rust_decimal", "rustls"] }
reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls"] }
reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "rustls-tls", "socks"] }
futures = "0.3"
mongodb = "3.2.5"
rust-oracle = { git = "https://github.com/t8y2/rust-oracle", branch = "main" }
crates/dbx-core/src/ai.rs
+53 -6
View File
@@ -70,6 +70,10 @@ pub struct AiConfig {
pub model: String,
#[serde(default)]
pub api_style: AiApiStyle,
#[serde(default)]
pub proxy_enabled: bool,
#[serde(default)]
pub proxy_url: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -218,6 +222,15 @@ fn validate_config(config: &AiConfig) -> Result<(), String> {
Ok(())
}
pub fn build_ai_http_client(config: &AiConfig, timeout_secs: u64) -> Result<reqwest::Client, String> {
let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(timeout_secs));
if config.proxy_enabled && !config.proxy_url.trim().is_empty() {
let proxy = reqwest::Proxy::all(config.proxy_url.trim()).map_err(|e| format!("Invalid AI proxy URL: {e}"))?;
builder = builder.proxy(proxy);
}
builder.build().map_err(|e| e.to_string())
}
// ---------------------------------------------------------------------------
// Non-streaming calls
// ---------------------------------------------------------------------------
@@ -339,8 +352,7 @@ pub async fn call_responses_api(client: &reqwest::Client, request: AiCompletionR
pub async fn test_connection_core(config: &AiConfig) -> Result<String, String> {
validate_config(config)?;
let client =
reqwest::Client::builder().timeout(std::time::Duration::from_secs(15)).build().map_err(|e| e.to_string())?;
let client = build_ai_http_client(config, 15)?;
let request = AiCompletionRequest {
config: config.clone(),
@@ -366,8 +378,7 @@ pub async fn test_connection_core(config: &AiConfig) -> Result<String, String> {
pub async fn complete(request: &AiCompletionRequest) -> Result<String, String> {
validate_config(&request.config)?;
let client =
reqwest::Client::builder().timeout(std::time::Duration::from_secs(60)).build().map_err(|e| e.to_string())?;
let client = build_ai_http_client(&request.config, 60)?;
match request.config.provider {
AiProvider::Claude => call_claude(&client, request.clone()).await,
@@ -393,8 +404,7 @@ pub async fn stream(
) -> Result<(), String> {
validate_config(&request.config)?;
let client =
reqwest::Client::builder().timeout(std::time::Duration::from_secs(120)).build().map_err(|e| e.to_string())?;
let client = build_ai_http_client(&request.config, 120)?;
match request.config.provider {
AiProvider::Claude => stream_claude(&client, session_id, request, cancelled, &on_chunk).await,
@@ -724,3 +734,40 @@ pub fn load_config(path: &Path) -> Result<Option<AiConfig>, String> {
let json = std::fs::read_to_string(path).map_err(|e| e.to_string())?;
serde_json::from_str(&json).map(Some).map_err(|e| e.to_string())
}
#[cfg(test)]
mod tests {
use super::{build_ai_http_client, AiApiStyle, AiConfig, AiProvider};
#[test]
fn ai_config_proxy_fields_default_for_legacy_config() {
let config: AiConfig = serde_json::from_value(serde_json::json!({
"provider": "openai",
"apiKey": "key",
"endpoint": "https://api.openai.com/v1/chat/completions",
"model": "gpt-4o",
"apiStyle": "completions"
}))
.unwrap();
assert_eq!(config.proxy_enabled, false);
assert_eq!(config.proxy_url, "");
}
#[test]
fn ai_http_client_rejects_invalid_proxy_url() {
let config = AiConfig {
provider: AiProvider::Openai,
api_key: "key".to_string(),
endpoint: "https://api.openai.com/v1/chat/completions".to_string(),
model: "gpt-4o".to_string(),
api_style: AiApiStyle::Completions,
proxy_enabled: true,
proxy_url: "not a proxy url".to_string(),
};
let err = build_ai_http_client(&config, 1).unwrap_err();
assert!(err.contains("Invalid AI proxy URL"));
}
}
crates/dbx-core/src/connection.rs
+57 -1
View File
@@ -5,6 +5,7 @@ use std::sync::Arc;
use tokio::sync::RwLock;
use crate::db;
use crate::db::proxy_tunnel::ProxyTunnelManager;
use crate::db::ssh_tunnel::TunnelManager;
use crate::external;
use crate::models::connection::{parse_mongo_first_host, ConnectionConfig, DatabaseType};
@@ -93,6 +94,7 @@ pub struct AppState {
pub configs: RwLock<HashMap<String, ConnectionConfig>>,
pub running_queries: RunningQueries,
pub tunnels: TunnelManager,
pub proxy_tunnels: ProxyTunnelManager,
pub storage: Storage,
pub plugins: PluginRegistry,
}
@@ -126,6 +128,7 @@ impl AppState {
configs: RwLock::new(HashMap::new()),
running_queries: RunningQueries::default(),
tunnels: TunnelManager::new(),
proxy_tunnels: ProxyTunnelManager::new(),
storage,
plugins: PluginRegistry::new(plugin_dir),
}
@@ -302,6 +305,37 @@ impl AppState {
config: &ConnectionConfig,
) -> Result<(String, u16), String> {
if !config.ssh_enabled || config.ssh_host.is_empty() {
if config.proxy_enabled && !config.proxy_host.is_empty() {
if let Some(local_port) = self.proxy_tunnels.local_port(connection_id).await {
return Ok(("127.0.0.1".to_string(), local_port));
}
let (remote_host, remote_port) = if config.db_type == DatabaseType::MongoDb {
config
.connection_string
.as_deref()
.filter(|s| !s.is_empty())
.and_then(parse_mongo_first_host)
.unwrap_or_else(|| (config.host.clone(), config.port))
} else {
(config.host.clone(), config.port)
};
let local_port = self
.proxy_tunnels
.start_tunnel(
connection_id,
config.proxy_type,
&config.proxy_host,
config.proxy_port,
&config.proxy_username,
&config.proxy_password,
&remote_host,
remote_port,
)
.await?;
return Ok(("127.0.0.1".to_string(), local_port));
}
return Ok((config.host.clone(), config.port));
}
@@ -412,7 +446,7 @@ async fn detect_ob_oracle_mode(config: &ConnectionConfig, pool: &sqlx::mysql::My
#[cfg(test)]
mod tests {
use super::{database_connection_config, metadata_connection_config, AppState};
use crate::models::connection::{ConnectionConfig, DatabaseType};
use crate::models::connection::{ConnectionConfig, DatabaseType, ProxyType};
use crate::schema;
use crate::storage::Storage;
@@ -439,6 +473,12 @@ mod tests {
ssh_key_passphrase: String::new(),
ssh_expose_lan: false,
ssh_connect_timeout_secs: crate::models::connection::default_ssh_connect_timeout_secs(),
proxy_enabled: false,
proxy_type: ProxyType::Socks5,
proxy_host: String::new(),
proxy_port: 1080,
proxy_username: String::new(),
proxy_password: String::new(),
ssl: false,
sysdba: false,
connection_string: None,
@@ -517,4 +557,20 @@ mod tests {
let _ = std::fs::remove_dir_all(dir);
}
#[tokio::test]
async fn proxy_connection_uses_local_forward_endpoint() {
let (state, dir) = test_app_state().await;
let mut config = mysql_config(Some("app"));
config.proxy_enabled = true;
config.proxy_host = "127.0.0.1".to_string();
config.proxy_port = 65000;
let (host, port) = state.connection_host_port("proxied", &config).await.unwrap();
assert_eq!(host, "127.0.0.1");
assert_ne!(port, config.port);
state.proxy_tunnels.stop_tunnel("proxied").await;
let _ = std::fs::remove_dir_all(dir);
}
}
crates/dbx-core/src/connection_secrets.rs
+19 -1
View File
@@ -6,6 +6,7 @@ use std::path::{Path, PathBuf};
pub const MAIN_PASSWORD_KEY: &str = "password";
pub const SSH_PASSWORD_KEY: &str = "ssh_password";
pub const SSH_KEY_PASSPHRASE_KEY: &str = "ssh_key_passphrase";
pub const PROXY_PASSWORD_KEY: &str = "proxy_password";
pub const CONNECTION_STRING_KEY: &str = "connection_string";
pub trait ConnectionSecretStore {
@@ -61,6 +62,7 @@ pub fn save_connections_to_file(
persist_secret(store, &config.id, MAIN_PASSWORD_KEY, &config.password)?;
persist_secret(store, &config.id, SSH_PASSWORD_KEY, &config.ssh_password)?;
persist_secret(store, &config.id, SSH_KEY_PASSPHRASE_KEY, &config.ssh_key_passphrase)?;
persist_secret(store, &config.id, PROXY_PASSWORD_KEY, &config.proxy_password)?;
persist_optional_secret(store, &config.id, CONNECTION_STRING_KEY, config.connection_string.as_deref())?;
}
@@ -105,6 +107,15 @@ pub fn load_connections_from_file(
needs_rewrite = true;
}
if config.proxy_password.is_empty() {
if let Some(secret) = store.get_secret(&config.id, PROXY_PASSWORD_KEY)? {
config.proxy_password = secret;
}
} else {
store.set_secret(&config.id, PROXY_PASSWORD_KEY, &config.proxy_password)?;
needs_rewrite = true;
}
match config.connection_string.as_deref().filter(|secret| !secret.is_empty()) {
Some(secret) => {
store.set_secret(&config.id, CONNECTION_STRING_KEY, secret)?;
@@ -195,6 +206,7 @@ fn sanitize_connections(configs: &[ConnectionConfig]) -> Vec<ConnectionConfig> {
config.password.clear();
config.ssh_password.clear();
config.ssh_key_passphrase.clear();
config.proxy_password.clear();
config.connection_string = None;
config
})
@@ -211,7 +223,7 @@ mod tests {
load_connections_from_file, save_connections_to_file, ConnectionSecretStore, CONNECTION_STRING_KEY,
MAIN_PASSWORD_KEY, SSH_PASSWORD_KEY,
};
use crate::models::connection::{ConnectionConfig, DatabaseType};
use crate::models::connection::{ConnectionConfig, DatabaseType, ProxyType};
use std::cell::RefCell;
use std::collections::HashMap;
use std::path::Path;
@@ -286,6 +298,12 @@ mod tests {
ssh_key_passphrase: String::new(),
ssh_expose_lan: false,
ssh_connect_timeout_secs: crate::models::connection::default_ssh_connect_timeout_secs(),
proxy_enabled: false,
proxy_type: ProxyType::Socks5,
proxy_host: String::new(),
proxy_port: 1080,
proxy_username: String::new(),
proxy_password: String::new(),
ssl: false,
sysdba: false,
connection_string: None,
crates/dbx-core/src/db/mod.rs
+1
View File
@@ -9,6 +9,7 @@ pub mod mysql;
pub mod ob_oracle;
pub mod oracle_driver;
pub mod postgres;
pub mod proxy_tunnel;
pub mod redis_driver;
pub mod sqlite;
pub mod sqlserver;
crates/dbx-core/src/db/oracle_driver.rs
+31 -1
View File
@@ -332,10 +332,12 @@ pub async fn list_triggers(conn: &OracleClient, schema: &str, table: &str) -> Re
pub async fn execute_query_with_schema(conn: &OracleClient, schema: &str, sql: &str) -> Result<QueryResult, String> {
let set_schema = format!("ALTER SESSION SET CURRENT_SCHEMA = \"{}\"", schema);
log::info!("[oracle][set-schema:start] schema={schema}");
conn.execute(&set_schema, &[]).await.map_err(|e| {
log::error!("[oracle] set current_schema failed: {e}");
e.to_string()
})?;
log::info!("[oracle][set-schema:done] schema={schema}");
execute_query(conn, sql).await
}
@@ -343,17 +345,31 @@ pub async fn execute_query(conn: &OracleClient, sql: &str) -> Result<QueryResult
let start = Instant::now();
let sql = sql.trim().trim_end_matches(';');
let explicit_limit = explicit_select_row_limit(sql);
log::info!("[oracle][execute:start] explicit_limit={:?} sql={}", explicit_limit, sql);
// Rewrite FETCH FIRST N ROWS ONLY → ROWNUM for Oracle 11g compatibility.
let sql = rewrite_fetch_first(sql);
log::info!("[oracle][execute:rewritten] sql={}", sql.as_ref());
if starts_with_executable_sql_keyword(sql.as_ref(), &["SELECT", "WITH", "SHOW", "DESCRIBE", "EXPLAIN"]) {
let capped_sql = cap_select_rows(sql.as_ref());
let query_limit = explicit_limit.unwrap_or(ORACLE_QUERY_LIMIT).min(ORACLE_QUERY_LIMIT);
log::info!(
"[oracle][query_with_limit:start] query_limit={} fetch_size=500 sql={}",
query_limit,
capped_sql.as_ref()
);
let result = conn.query_with_limit(capped_sql.as_ref(), &[], query_limit, 500).await.map_err(|e| {
log::error!("[oracle] execute_query SELECT failed: {e}");
e.to_string()
})?;
log::info!(
"[oracle][query_with_limit:done] column_count={} row_count={} has_more_rows={} elapsed_ms={}",
result.columns.len(),
result.rows.len(),
result.has_more_rows,
start.elapsed().as_millis()
);
let columns: Vec<String> = result.columns.iter().map(|c| c.name.clone()).collect();
let mut rows: Vec<Vec<serde_json::Value>> = result
.rows
@@ -369,11 +385,24 @@ pub async fn execute_query(conn: &OracleClient, sql: &str) -> Result<QueryResult
rows.truncate(crate::query::MAX_ROWS);
}
log::info!(
"[oracle][execute:done] column_count={} row_count={} truncated={} elapsed_ms={}",
columns.len(),
rows.len(),
truncated,
start.elapsed().as_millis()
);
Ok(QueryResult { columns, rows, affected_rows: 0, execution_time_ms: start.elapsed().as_millis(), truncated })
} else {
log::info!("[oracle][execute-non-select:start] sql={}", sql.as_ref());
match conn.execute(sql.as_ref(), &[]).await {
Ok(result) => {
let _ = conn.commit().await;
log::info!(
"[oracle][execute-non-select:done] affected_rows={} elapsed_ms={}",
result.rows_affected,
start.elapsed().as_millis()
);
Ok(QueryResult {
columns: vec![],
rows: vec![],
@@ -384,8 +413,9 @@ pub async fn execute_query(conn: &OracleClient, sql: &str) -> Result<QueryResult
}
Err(e) => {
let msg = e.to_string();
log::error!("[oracle][execute-non-select:error] {msg}");
if msg.contains("Server rejected") || msg.contains("closed the connection") {
Err("Operation failed (connection closed) — possibly a constraint violation (foreign key, unique, or check constraint).".to_string())
Err(format!("Operation failed (connection closed). Original driver error: {msg}"))
} else {
Err(msg)
}
crates/dbx-core/src/db/proxy_tunnel.rs
+212
View File
@@ -0,0 +1,212 @@
use crate::models::connection::ProxyType;
use base64::engine::general_purpose::STANDARD as BASE64;
use base64::Engine;
use std::collections::HashMap;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::task::JoinHandle;
use tokio::time::{timeout, Duration};
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Default)]
pub struct ProxyTunnelManager {
tunnels: tokio::sync::Mutex<HashMap<String, (JoinHandle<()>, u16)>>,
}
impl ProxyTunnelManager {
pub fn new() -> Self {
Self { tunnels: tokio::sync::Mutex::new(HashMap::new()) }
}
pub async fn start_tunnel(
&self,
connection_id: &str,
proxy_type: ProxyType,
proxy_host: &str,
proxy_port: u16,
proxy_username: &str,
proxy_password: &str,
remote_host: &str,
remote_port: u16,
) -> Result<u16, String> {
let local_port = portpicker::pick_unused_port().ok_or("No available port")?;
let listener = TcpListener::bind(("127.0.0.1", local_port))
.await
.map_err(|e| format!("Failed to bind proxy tunnel local port: {e}"))?;
let proxy = ProxyEndpoint {
proxy_type,
host: proxy_host.to_string(),
port: proxy_port,
username: proxy_username.to_string(),
password: proxy_password.to_string(),
};
let remote = RemoteEndpoint { host: remote_host.to_string(), port: remote_port };
let handle = tokio::spawn(proxy_forward_loop(listener, proxy, remote));
self.tunnels.lock().await.insert(connection_id.to_string(), (handle, local_port));
Ok(local_port)
}
pub async fn local_port(&self, connection_id: &str) -> Option<u16> {
self.tunnels.lock().await.get(connection_id).map(|(_, port)| *port)
}
pub async fn stop_tunnel(&self, connection_id: &str) {
if let Some((handle, _)) = self.tunnels.lock().await.remove(connection_id) {
handle.abort();
}
}
}
#[derive(Clone)]
struct ProxyEndpoint {
proxy_type: ProxyType,
host: String,
port: u16,
username: String,
password: String,
}
#[derive(Clone)]
struct RemoteEndpoint {
host: String,
port: u16,
}
async fn proxy_forward_loop(listener: TcpListener, proxy: ProxyEndpoint, remote: RemoteEndpoint) {
loop {
let (mut inbound, _) = match listener.accept().await {
Ok(pair) => pair,
Err(_) => break,
};
let proxy = proxy.clone();
let remote = remote.clone();
tokio::spawn(async move {
let Ok(mut outbound) = connect_via_proxy(&proxy, &remote).await else {
return;
};
let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await;
});
}
}
async fn connect_via_proxy(proxy: &ProxyEndpoint, remote: &RemoteEndpoint) -> Result<TcpStream, String> {
let stream = timeout(CONNECT_TIMEOUT, TcpStream::connect((proxy.host.as_str(), proxy.port)))
.await
.map_err(|_| "Proxy connection timed out".to_string())?
.map_err(|e| format!("Failed to connect proxy: {e}"))?;
match proxy.proxy_type {
ProxyType::Http => http_connect(stream, proxy, remote).await,
ProxyType::Socks5 => socks5_connect(stream, proxy, remote).await,
}
}
async fn http_connect(
mut stream: TcpStream,
proxy: &ProxyEndpoint,
remote: &RemoteEndpoint,
) -> Result<TcpStream, String> {
let target = format!("{}:{}", remote.host, remote.port);
let mut request = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n");
if !proxy.username.is_empty() || !proxy.password.is_empty() {
let token = BASE64.encode(format!("{}:{}", proxy.username, proxy.password));
request.push_str(&format!("Proxy-Authorization: Basic {token}\r\n"));
}
request.push_str("\r\n");
stream.write_all(request.as_bytes()).await.map_err(|e| format!("Failed to send CONNECT request: {e}"))?;
let mut response = Vec::new();
let mut buf = [0_u8; 1];
while !response.ends_with(b"\r\n\r\n") && response.len() < 8192 {
let n = stream.read(&mut buf).await.map_err(|e| format!("Failed to read CONNECT response: {e}"))?;
if n == 0 {
break;
}
response.push(buf[0]);
}
let text = String::from_utf8_lossy(&response);
if text.starts_with("HTTP/1.1 200") || text.starts_with("HTTP/1.0 200") {
Ok(stream)
} else {
let status = text.lines().next().unwrap_or("invalid proxy response");
Err(format!("HTTP proxy CONNECT failed: {status}"))
}
}
async fn socks5_connect(
mut stream: TcpStream,
proxy: &ProxyEndpoint,
remote: &RemoteEndpoint,
) -> Result<TcpStream, String> {
let wants_auth = !proxy.username.is_empty() || !proxy.password.is_empty();
let methods: &[u8] = if wants_auth { &[0x00, 0x02] } else { &[0x00] };
let mut hello = vec![0x05, methods.len() as u8];
hello.extend_from_slice(methods);
stream.write_all(&hello).await.map_err(|e| format!("Failed to send SOCKS greeting: {e}"))?;
let mut method = [0_u8; 2];
stream.read_exact(&mut method).await.map_err(|e| format!("Failed to read SOCKS greeting: {e}"))?;
if method[0] != 0x05 {
return Err("Invalid SOCKS proxy version".to_string());
}
match method[1] {
0x00 => {}
0x02 => socks5_authenticate(&mut stream, proxy).await?,
0xff => return Err("SOCKS proxy rejected supported authentication methods".to_string()),
other => return Err(format!("SOCKS proxy selected unsupported auth method: {other}")),
}
let host = remote.host.as_bytes();
if host.len() > u8::MAX as usize {
return Err("Remote host is too long for SOCKS5 domain address".to_string());
}
let mut req = vec![0x05, 0x01, 0x00, 0x03, host.len() as u8];
req.extend_from_slice(host);
req.extend_from_slice(&remote.port.to_be_bytes());
stream.write_all(&req).await.map_err(|e| format!("Failed to send SOCKS connect request: {e}"))?;
let mut head = [0_u8; 4];
stream.read_exact(&mut head).await.map_err(|e| format!("Failed to read SOCKS connect response: {e}"))?;
if head[0] != 0x05 {
return Err("Invalid SOCKS connect response version".to_string());
}
if head[1] != 0x00 {
return Err(format!("SOCKS proxy connect failed with code {}", head[1]));
}
let addr_len = match head[3] {
0x01 => 4,
0x03 => {
let mut len = [0_u8; 1];
stream.read_exact(&mut len).await.map_err(|e| format!("Failed to read SOCKS bound address length: {e}"))?;
len[0] as usize
}
0x04 => 16,
other => return Err(format!("Unsupported SOCKS bound address type: {other}")),
};
let mut discard = vec![0_u8; addr_len + 2];
stream.read_exact(&mut discard).await.map_err(|e| format!("Failed to read SOCKS bound address: {e}"))?;
Ok(stream)
}
async fn socks5_authenticate(stream: &mut TcpStream, proxy: &ProxyEndpoint) -> Result<(), String> {
let username = proxy.username.as_bytes();
let password = proxy.password.as_bytes();
if username.len() > u8::MAX as usize || password.len() > u8::MAX as usize {
return Err("SOCKS username or password is too long".to_string());
}
let mut req = vec![0x01, username.len() as u8];
req.extend_from_slice(username);
req.push(password.len() as u8);
req.extend_from_slice(password);
stream.write_all(&req).await.map_err(|e| format!("Failed to send SOCKS authentication: {e}"))?;
let mut res = [0_u8; 2];
stream.read_exact(&mut res).await.map_err(|e| format!("Failed to read SOCKS authentication response: {e}"))?;
if res == [0x01, 0x00] {
Ok(())
} else {
Err("SOCKS proxy authentication failed".to_string())
}
}
crates/dbx-core/src/models/connection.rs
+58 -1
View File
@@ -38,6 +38,18 @@ pub struct ConnectionConfig {
#[serde(default = "default_ssh_connect_timeout_secs")]
pub ssh_connect_timeout_secs: u64,
#[serde(default)]
pub proxy_enabled: bool,
#[serde(default)]
pub proxy_type: ProxyType,
#[serde(default)]
pub proxy_host: String,
#[serde(default = "default_proxy_port")]
pub proxy_port: u16,
#[serde(default)]
pub proxy_username: String,
#[serde(default)]
pub proxy_password: String,
#[serde(default)]
pub ssl: bool,
#[serde(default)]
pub sysdba: bool,
@@ -60,6 +72,23 @@ pub fn default_ssh_connect_timeout_secs() -> u64 {
5
}
fn default_proxy_port() -> u16 {
1080
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum ProxyType {
Socks5,
Http,
}
impl Default for ProxyType {
fn default() -> Self {
Self::Socks5
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum DatabaseType {
@@ -372,7 +401,7 @@ fn bracket_ipv6(host: &str) -> String {
#[cfg(test)]
mod tests {
use super::{default_ssh_connect_timeout_secs, ConnectionConfig, DatabaseType};
use super::{default_ssh_connect_timeout_secs, ConnectionConfig, DatabaseType, ProxyType};
fn mysql_config(username: &str, password: &str, database: Option<&str>) -> ConnectionConfig {
ConnectionConfig {
@@ -397,6 +426,12 @@ mod tests {
ssh_key_passphrase: String::new(),
ssh_expose_lan: false,
ssh_connect_timeout_secs: default_ssh_connect_timeout_secs(),
proxy_enabled: false,
proxy_type: ProxyType::Socks5,
proxy_host: String::new(),
proxy_port: 1080,
proxy_username: String::new(),
proxy_password: String::new(),
ssl: false,
sysdba: false,
connection_string: None,
@@ -431,6 +466,28 @@ mod tests {
assert_eq!(config.effective_ssh_connect_timeout_secs(), default_ssh_connect_timeout_secs());
}
#[test]
fn proxy_fields_default_for_legacy_config() {
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
"id": "id",
"name": "name",
"db_type": "mysql",
"host": "10.1.2.3",
"port": 3306,
"username": "root",
"password": "",
"database": null
}))
.unwrap();
assert_eq!(config.proxy_enabled, false);
assert_eq!(config.proxy_type, ProxyType::Socks5);
assert_eq!(config.proxy_host, "");
assert_eq!(config.proxy_port, 1080);
assert_eq!(config.proxy_username, "");
assert_eq!(config.proxy_password, "");
}
#[test]
fn ssh_connect_timeout_zero_uses_default() {
let mut config = mysql_config("root", "", None);
crates/dbx-core/src/query.rs
+4
View File
@@ -205,6 +205,7 @@ pub async fn do_execute(
let client = pool.client();
let schema = schema.map(|s| s.to_string());
drop(connections);
log::info!("[query][oracle:lock:start] schema={:?} sql={}", schema, sql);
let client = match cancel_token.as_ref() {
Some(token) => tokio::select! {
biased;
@@ -213,6 +214,7 @@ pub async fn do_execute(
},
None => client.lock().await,
};
log::info!("[query][oracle:lock:done] schema={:?}", schema);
if let Some(schema) = schema {
wait_for_query(cancel_token, db::oracle_driver::execute_query_with_schema(&*client, &schema, sql))
.await
@@ -635,9 +637,11 @@ async fn exec_tx_none_inner(
) -> Result<db::QueryResult, String> {
let mut total_affected: u64 = 0;
for (i, sql) in statements.iter().enumerate() {
log::info!("[query][tx-none:statement:start] index={} sql={}", i + 1, sql);
match do_execute(state, pool_key, sql, schema, None).await {
Ok(result) => {
total_affected += result.affected_rows;
log::info!("[query][tx-none:statement:done] index={} affected_rows={}", i + 1, result.affected_rows);
}
Err(e) => {
log::warn!("Statement {} failed (no transaction support): {}", i + 1, e);
crates/dbx-core/src/storage.rs
+3
View File
@@ -397,6 +397,7 @@ impl Storage {
sanitized.password = String::new();
sanitized.ssh_password = String::new();
sanitized.ssh_key_passphrase = String::new();
sanitized.proxy_password = String::new();
sanitized.connection_string = None;
let json = serde_json::to_string(&sanitized).map_err(|e| e.to_string())?;
@@ -411,6 +412,7 @@ impl Storage {
persist_secret_in_tx(&mut tx, &config.id, "password", &config.password).await?;
persist_secret_in_tx(&mut tx, &config.id, "ssh_password", &config.ssh_password).await?;
persist_secret_in_tx(&mut tx, &config.id, "ssh_key_passphrase", &config.ssh_key_passphrase).await?;
persist_secret_in_tx(&mut tx, &config.id, "proxy_password", &config.proxy_password).await?;
if let Some(cs) = &config.connection_string {
persist_secret_in_tx(&mut tx, &config.id, "connection_string", cs).await?;
} else {
@@ -452,6 +454,7 @@ impl Storage {
config.password = self.get_secret(&id, "password").await?.unwrap_or_default();
config.ssh_password = self.get_secret(&id, "ssh_password").await?.unwrap_or_default();
config.ssh_key_passphrase = self.get_secret(&id, "ssh_key_passphrase").await?.unwrap_or_default();
config.proxy_password = self.get_secret(&id, "proxy_password").await?.unwrap_or_default();
config.connection_string = self.get_secret(&id, "connection_string").await?;
configs.push(config);
}