fix(jdbc): route UI connections through transport endpoint

This commit is contained in:
wyeye authored and GitHub committed 2026-08-18 12:47:52 +08:00
1 parent 4d32bc4981
commit 3bb57b9c90
4 files changed
+272 -128

No files matched your search

+94 -61
View File
@@ -34,7 +34,12 @@ fn agent_jdbc_driver_class(config: &ConnectionConfig) -> &str {
}
}
pub fn agent_connect_params(config: &ConnectionConfig, host: &str, port: u16, database: &str) -> serde_json::Value {
pub fn agent_connect_params(
config: &ConnectionConfig,
host: &str,
port: u16,
database: &str,
) -> Result<serde_json::Value, String> {
agent_connect_params_with_role(config, host, port, database, AgentSessionRole::Workload)
}
@@ -44,7 +49,7 @@ pub fn agent_connect_params_with_role(
port: u16,
database: &str,
session_role: AgentSessionRole,
) -> serde_json::Value {
) -> Result<serde_json::Value, String> {
let agent_database = if config.db_type == DatabaseType::MongoDb {
mongo_agent_database(config, database)
} else if matches!(config.db_type, DatabaseType::Oracle | DatabaseType::OceanbaseOracle) {
@@ -62,9 +67,9 @@ pub fn agent_connect_params_with_role(
let connection_string = if config.db_type == DatabaseType::MongoDb {
config.connection_url_with_host(host, port)
} else if config.db_type == DatabaseType::Oracle {
oracle_jdbc_connection_string(config, host, port, database)
oracle_jdbc_connection_string(config, host, port, database)?
} else if config.db_type == DatabaseType::OceanbaseOracle {
oceanbase_oracle_jdbc_connection_string(config, host, port, database)
oceanbase_oracle_jdbc_connection_string(config, host, port, database)?
} else if matches!(
config.db_type,
DatabaseType::Kingbase | DatabaseType::Highgo | DatabaseType::Uxdb | DatabaseType::Vastbase
@@ -73,7 +78,7 @@ pub fn agent_connect_params_with_role(
} else if config.db_type == DatabaseType::SapHana {
sap_hana_jdbc_connection_string(config, host, port, database)
} else if matches!(config.db_type, DatabaseType::Trino | DatabaseType::PrestoSql) {
trino_like_jdbc_connection_string(config, host, port, database)
trino_like_jdbc_connection_string(config, host, port, database)?
} else if config.db_type == DatabaseType::H2 {
h2_agent_jdbc_connection_string(config)
} else {
@@ -117,7 +122,7 @@ pub fn agent_connect_params_with_role(
(config.effective_connect_timeout_secs() * 1000).max(ZOOKEEPER_MIN_CONNECTION_TIMEOUT_MS)
);
}
params
Ok(params)
}
fn oracle_uses_sysdba(config: &ConnectionConfig) -> bool {
@@ -330,7 +335,7 @@ pub fn oracle_alternate_connect_configs(config: &ConnectionConfig, err: &str) ->
let host = config.host.trim();
let port = config.port;
let current_url = oracle_jdbc_connection_string(config, host, port, database);
let current_url = oracle_jdbc_connection_string(config, host, port, database).unwrap_or_default();
let service_url = oracle_service_jdbc_url(host, port, database);
let sid_url = oracle_sid_jdbc_url(host, port, database);
let legacy_service_url = oracle_legacy_service_jdbc_url(host, port, database);
@@ -367,25 +372,30 @@ pub fn oracle_alternate_connect_configs(config: &ConnectionConfig, err: &str) ->
.collect()
}
fn oracle_jdbc_connection_string(config: &ConnectionConfig, host: &str, port: u16, database: &str) -> String {
fn oracle_jdbc_connection_string(
config: &ConnectionConfig,
host: &str,
port: u16,
database: &str,
) -> Result<String, String> {
if let Some(connection_string) = config.connection_string.as_deref().filter(|value| !value.trim().is_empty()) {
let connection_string = connection_string.trim();
if host == config.host && port == config.port {
return connection_string.to_string();
return Ok(connection_string.to_string());
}
return crate::models::connection::rewrite_jdbc_url_host(connection_string, host, port);
}
let database = database.trim();
if database.is_empty() {
return String::new();
return Ok(String::new());
}
if config.oracle_connection_type.as_deref() == Some("sid") {
Ok(if config.oracle_connection_type.as_deref() == Some("sid") {
oracle_sid_jdbc_url(host, port, database)
} else {
oracle_service_jdbc_url(host, port, database)
}
})
}
fn oracle_listener_error_can_retry(err: &str) -> bool {
@@ -413,27 +423,32 @@ fn oracle_descriptor_jdbc_url(host: &str, port: u16, database: &str, key: &str)
format!("jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST={host})(PORT={port}))(CONNECT_DATA=({key}={database})))")
}
fn oceanbase_oracle_jdbc_connection_string(config: &ConnectionConfig, host: &str, port: u16, database: &str) -> String {
fn oceanbase_oracle_jdbc_connection_string(
config: &ConnectionConfig,
host: &str,
port: u16,
database: &str,
) -> Result<String, String> {
if let Some(connection_string) = config.connection_string.as_deref().filter(|value| !value.trim().is_empty()) {
let connection_string = connection_string.trim();
let url = if host == config.host && port == config.port {
connection_string.to_string()
} else {
crate::models::connection::rewrite_jdbc_url_host(connection_string, host, port)
crate::models::connection::rewrite_jdbc_url_host(connection_string, host, port)?
};
if url_has_query_key(&url, OCEANBASE_ORACLE_COMPATIBLE_OJDBC_VERSION_KEY) {
return url;
return Ok(url);
}
return append_agent_url_params(url, Some(OCEANBASE_ORACLE_COMPATIBLE_OJDBC_VERSION_PARAM));
return Ok(append_agent_url_params(url, Some(OCEANBASE_ORACLE_COMPATIBLE_OJDBC_VERSION_PARAM)));
}
let database = database.trim();
if database.is_empty() {
return String::new();
return Ok(String::new());
}
let base = format!("jdbc:oceanbase://{host}:{port}/{database}");
append_agent_url_params(base, Some(&oceanbase_oracle_jdbc_params(config)))
Ok(append_agent_url_params(base, Some(&oceanbase_oracle_jdbc_params(config))))
}
fn oceanbase_oracle_jdbc_params(config: &ConnectionConfig) -> String {
@@ -529,7 +544,12 @@ fn sap_hana_jdbc_connection_string(config: &ConnectionConfig, host: &str, port:
}
}
pub fn trino_like_jdbc_connection_string(config: &ConnectionConfig, host: &str, port: u16, database: &str) -> String {
pub fn trino_like_jdbc_connection_string(
config: &ConnectionConfig,
host: &str,
port: u16,
database: &str,
) -> Result<String, String> {
let jdbc_scheme = match config.db_type {
DatabaseType::PrestoSql => "presto",
_ => "trino",
@@ -542,11 +562,12 @@ pub fn trino_like_jdbc_connection_string(config: &ConnectionConfig, host: &str,
.filter(|value| value.get(..jdbc_prefix.len()).is_some_and(|prefix| prefix.eq_ignore_ascii_case(&jdbc_prefix)))
.map(|connection_string| {
if host == config.host && port == config.port {
connection_string.to_string()
Ok(connection_string.to_string())
} else {
crate::models::connection::rewrite_jdbc_url_host(connection_string, host, port)
}
})
.transpose()?
.unwrap_or_else(|| {
let database = database.trim();
if database.is_empty() {
@@ -557,11 +578,7 @@ pub fn trino_like_jdbc_connection_string(config: &ConnectionConfig, host: &str,
});
let params = trino_agent_jdbc_params(config, &base);
if params.is_empty() {
base
} else {
append_agent_url_params(base, Some(&params))
}
Ok(if params.is_empty() { base } else { append_agent_url_params(base, Some(&params)) })
}
fn trino_agent_jdbc_params(config: &ConnectionConfig, base: &str) -> String {
@@ -724,7 +741,7 @@ mod tests {
#[test]
fn agent_connect_params_default_to_workload_session_role() {
let params = agent_connect_params(&config(DatabaseType::H2, Some("test")), "127.0.0.1", 9092, "test");
let params = agent_connect_params(&config(DatabaseType::H2, Some("test")), "127.0.0.1", 9092, "test").unwrap();
assert_eq!(params["sessionRole"], "workload");
assert_eq!(params["database_type"], "h2");
}
@@ -737,7 +754,8 @@ mod tests {
9092,
"test",
AgentSessionRole::Metadata,
);
)
.unwrap();
assert_eq!(params["sessionRole"], "metadata");
}
@@ -765,7 +783,7 @@ mod tests {
let mut cfg = config(DatabaseType::MongoDb, None);
cfg.connection_string = Some("mongodb://user:secret@127.0.0.1:27017/app_db?authSource=admin".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 27017, "");
let params = agent_connect_params(&cfg, "127.0.0.1", 27017, "").unwrap();
assert_eq!(params["database"], "app_db");
}
@@ -784,7 +802,7 @@ mod tests {
cfg.host = "/tmp/app.mv.db".to_string();
cfg.port = 0;
let params = agent_connect_params(&cfg, "/tmp/app.mv.db", 0, "");
let params = agent_connect_params(&cfg, "/tmp/app.mv.db", 0, "").unwrap();
assert_eq!(params["host"], "");
assert_eq!(params["port"], 0);
@@ -797,7 +815,7 @@ mod tests {
let mut cfg = config(DatabaseType::H2, None);
cfg.connection_string = Some("jdbc:h2:file:/tmp/app.mv.db;AUTO_SERVER=TRUE".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test");
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test").unwrap();
assert_eq!(params["host"], "");
assert_eq!(params["port"], 0);
@@ -810,7 +828,7 @@ mod tests {
let mut cfg = config(DatabaseType::H2, None);
cfg.connection_string = Some("jdbc:h2:split:28:C:/dbx-test/h2/sample-db;AUTO_SERVER=TRUE".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test");
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test").unwrap();
assert_eq!(
h2_file_path_from_jdbc_url(cfg.connection_string.as_deref().unwrap()).as_deref(),
@@ -828,7 +846,7 @@ mod tests {
cfg.host = "127.0.0.1".to_string();
cfg.port = 9092;
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test");
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test").unwrap();
assert_eq!(params["host"], "127.0.0.1");
assert_eq!(params["port"], 9092);
@@ -843,7 +861,7 @@ mod tests {
cfg.driver_profile = driver_profile.map(str::to_string);
cfg.jdbc_driver_class = Some("h2_embedded".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test");
let params = agent_connect_params(&cfg, "127.0.0.1", 9092, "test").unwrap();
assert_eq!(params["jdbc_driver_class"], "");
}
@@ -853,7 +871,7 @@ mod tests {
fn vastbase_agent_url_defaults_to_postgres_database_when_empty() {
let cfg = config(DatabaseType::Vastbase, Some(""));
let params = agent_connect_params(&cfg, "vastbase.example.com", 5432, "");
let params = agent_connect_params(&cfg, "vastbase.example.com", 5432, "").unwrap();
assert_eq!(params["database"], "postgres");
assert_eq!(params["connection_string"], "jdbc:vastbase://vastbase.example.com:5432/postgres");
@@ -863,7 +881,7 @@ mod tests {
fn uxdb_agent_params_use_vendor_jdbc_url() {
let cfg = config(DatabaseType::Uxdb, Some("uxdb"));
let params = agent_connect_params(&cfg, "uxdb.example.com", 52025, "uxdb");
let params = agent_connect_params(&cfg, "uxdb.example.com", 52025, "uxdb").unwrap();
assert_eq!(params["database"], "uxdb");
assert_eq!(params["connection_string"], "jdbc:uxdb://uxdb.example.com:52025/uxdb");
@@ -875,7 +893,7 @@ mod tests {
cfg.connection_string = Some("zk-1:2181,zk-2:2181/app".to_string());
cfg.connect_timeout_secs = 20;
let params = agent_connect_params(&cfg, "127.0.0.1", 2181, "");
let params = agent_connect_params(&cfg, "127.0.0.1", 2181, "").unwrap();
assert_eq!(params["connection_string"], "zk-1:2181,zk-2:2181/app");
assert_eq!(params["zookeeper_connect_string"], "zk-1:2181,zk-2:2181/app");
@@ -887,7 +905,7 @@ mod tests {
let mut cfg = config(DatabaseType::Mysql, Some("app"));
cfg.connect_timeout_secs = 45;
let params = agent_connect_params(&cfg, "mysql.example.com", 3306, "app");
let params = agent_connect_params(&cfg, "mysql.example.com", 3306, "app").unwrap();
assert_eq!(params["connect_timeout_secs"], 45);
}
@@ -896,7 +914,7 @@ mod tests {
fn zookeeper_agent_params_fall_back_to_host_port_connect_string() {
let cfg = config(DatabaseType::ZooKeeper, None);
let params = agent_connect_params(&cfg, "zk.local", 2281, "");
let params = agent_connect_params(&cfg, "zk.local", 2281, "").unwrap();
assert_eq!(params["connection_string"], "");
assert_eq!(params["zookeeper_connect_string"], "zk.local:2281");
@@ -945,11 +963,11 @@ mod tests {
let mut cfg = config(DatabaseType::Oracle, Some("ORCL"));
cfg.oracle_connection_type = Some("sid".to_string());
let sid = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCL");
let sid = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCL").unwrap();
assert_eq!(sid["connection_string"], "jdbc:oracle:thin:@oracle.example.com:1521:ORCL");
cfg.oracle_connection_type = Some("service_name".to_string());
let service = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCL");
let service = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCL").unwrap();
assert_eq!(service["connection_string"], "jdbc:oracle:thin:@//oracle.example.com:1521/ORCL");
}
@@ -973,7 +991,7 @@ mod tests {
cfg.username = "SYS".to_string();
cfg.oracle_connection_type = Some("service_name".to_string());
let params = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCLPDB1");
let params = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCLPDB1").unwrap();
assert_eq!(params["database"], "SYSDBA:ORCLPDB1");
assert_eq!(params["sysdba"], true);
@@ -986,7 +1004,7 @@ mod tests {
cfg.username = "system".to_string();
cfg.sysdba = true;
let params = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCLPDB1");
let params = agent_connect_params(&cfg, "oracle.example.com", 1521, "ORCLPDB1").unwrap();
assert_eq!(params["database"], "SYSDBA:ORCLPDB1");
assert_eq!(params["sysdba"], true);
@@ -998,7 +1016,7 @@ mod tests {
cfg.host = "oceanbase.example.com".to_string();
cfg.port = 2881;
let params = agent_connect_params(&cfg, "oceanbase.example.com", 2881, "sys");
let params = agent_connect_params(&cfg, "oceanbase.example.com", 2881, "sys").unwrap();
assert_eq!(params["database"], "sys");
assert_eq!(params["sysdba"], false);
@@ -1015,7 +1033,7 @@ mod tests {
cfg.port = 2881;
cfg.url_params = Some("useSSL=false".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys");
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys").unwrap();
assert_eq!(
params["connection_string"],
@@ -1030,7 +1048,7 @@ mod tests {
cfg.port = 2881;
cfg.url_params = Some("compatibleOjdbcVersion=6&useSSL=false".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys");
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys").unwrap();
assert_eq!(
params["connection_string"],
@@ -1045,7 +1063,7 @@ mod tests {
cfg.port = 2881;
cfg.connection_string = Some("jdbc:oceanbase://oceanbase.example.com:2881/sys?useSSL=false".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys");
let params = agent_connect_params(&cfg, "127.0.0.1", 12881, "sys").unwrap();
assert_eq!(
params["connection_string"],
@@ -1063,7 +1081,7 @@ mod tests {
.to_string(),
);
let params = agent_connect_params(&cfg, "127.0.0.1", 11521, "ORCL");
let params = agent_connect_params(&cfg, "127.0.0.1", 11521, "ORCL").unwrap();
assert_eq!(
params["connection_string"],
@@ -1081,7 +1099,7 @@ mod tests {
.to_string(),
);
let params = agent_connect_params(&cfg, "form-host.example.com", 1521, "ORCL");
let params = agent_connect_params(&cfg, "form-host.example.com", 1521, "ORCL").unwrap();
assert_eq!(
params["connection_string"],
@@ -1089,6 +1107,21 @@ mod tests {
);
}
#[test]
fn oracle_multi_address_descriptor_fails_closed_for_transport_endpoint() {
let mut cfg = config(DatabaseType::Oracle, Some("ORCL"));
cfg.host = "oracle.example.com".to_string();
cfg.port = 1521;
cfg.connection_string = Some(
"jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=db1.example.com)(PORT=1521))(ADDRESS=(PROTOCOL=TCP)(HOST=db2.example.com)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=ORCL)))"
.to_string(),
);
let err = agent_connect_params(&cfg, "127.0.0.1", 11521, "ORCL").unwrap_err();
assert!(err.contains("single ADDRESS entry"), "{err}");
}
#[test]
fn oracle_listener_errors_can_switch_descriptor() {
let mut cfg = config(DatabaseType::Oracle, Some("ORCL"));
@@ -1139,7 +1172,7 @@ mod tests {
let mut cfg = config(DatabaseType::SapHana, Some("TENANT1"));
cfg.url_params = Some("encrypt=true".to_string());
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1");
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1").unwrap();
assert_eq!(params["connection_string"], "jdbc:sap://hana.example.com:30013/?databaseName=TENANT1&encrypt=true");
}
@@ -1151,7 +1184,7 @@ mod tests {
cfg.jdbc_driver_class = Some(alias.to_string());
cfg.jdbc_driver_paths = vec!["/tmp/ngdbc.jar".to_string()];
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1");
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1").unwrap();
assert_eq!(params["jdbc_driver_class"], "");
assert_eq!(params["jdbc_driver_paths"], serde_json::json!(["/tmp/ngdbc.jar"]));
@@ -1163,7 +1196,7 @@ mod tests {
let mut cfg = config(DatabaseType::SapHana, Some("TENANT1"));
cfg.jdbc_driver_class = Some("com.example.CustomSapHanaDriver".to_string());
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1");
let params = agent_connect_params(&cfg, "hana.example.com", 30013, "TENANT1").unwrap();
assert_eq!(params["jdbc_driver_class"], "com.example.CustomSapHanaDriver");
}
@@ -1174,7 +1207,7 @@ mod tests {
cfg.jdbc_driver_class = Some("sap_hana".to_string());
cfg.jdbc_driver_paths = vec!["/tmp/custom-driver.jar".to_string()];
let params = agent_connect_params(&cfg, "mysql.example.com", 3306, "test");
let params = agent_connect_params(&cfg, "mysql.example.com", 3306, "test").unwrap();
assert_eq!(params["jdbc_driver_class"], "sap_hana");
assert_eq!(params["jdbc_driver_paths"], serde_json::json!(["/tmp/custom-driver.jar"]));
@@ -1185,7 +1218,7 @@ mod tests {
let mut cfg = config(DatabaseType::Jdbc, Some("test"));
cfg.jdbc_driver_class = Some("com.example.CustomDriver".to_string());
let params = agent_connect_params(&cfg, "jdbc.example.com", 1234, "test");
let params = agent_connect_params(&cfg, "jdbc.example.com", 1234, "test").unwrap();
assert_eq!(params["jdbc_driver_class"], "com.example.CustomDriver");
}
@@ -1196,7 +1229,7 @@ mod tests {
cfg.host = "trino.example.com".to_string();
cfg.port = 8080;
let params = agent_connect_params(&cfg, "trino.example.com", 8080, "hive");
let params = agent_connect_params(&cfg, "trino.example.com", 8080, "hive").unwrap();
assert_eq!(params["connection_string"], "jdbc:trino://trino.example.com:8080/hive");
assert_eq!(params["ssl"], false);
@@ -1208,7 +1241,7 @@ mod tests {
cfg.host = "presto.example.com".to_string();
cfg.port = 9090;
let params = agent_connect_params(&cfg, "presto.example.com", 9090, "hive/default");
let params = agent_connect_params(&cfg, "presto.example.com", 9090, "hive/default").unwrap();
assert_eq!(params["connection_string"], "jdbc:presto://presto.example.com:9090/hive/default");
assert_eq!(params["ssl"], false);
@@ -1221,7 +1254,7 @@ mod tests {
cfg.port = 9090;
cfg.connection_string = Some("jdbc:presto://presto.internal:9090/hive/default?source=dbx".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 19090, "hive/default");
let params = agent_connect_params(&cfg, "127.0.0.1", 19090, "hive/default").unwrap();
assert_eq!(params["connection_string"], "jdbc:presto://127.0.0.1:19090/hive/default?source=dbx");
}
@@ -1231,7 +1264,7 @@ mod tests {
let mut cfg = config(DatabaseType::Trino, Some("hive"));
cfg.ssl = true;
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive");
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive").unwrap();
assert_eq!(params["connection_string"], "jdbc:trino://trino.example.com:8443/hive?SSL=true");
}
@@ -1242,7 +1275,7 @@ mod tests {
cfg.ssl = true;
cfg.url_params = Some("ssl=true&SSLVerification=NONE".to_string());
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive");
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive").unwrap();
assert_eq!(
params["connection_string"],
@@ -1259,7 +1292,7 @@ mod tests {
.to_string(),
);
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive");
let params = agent_connect_params(&cfg, "trino.example.com", 8443, "hive").unwrap();
assert_eq!(
params["connection_string"],
@@ -1275,7 +1308,7 @@ mod tests {
cfg.ssl = true;
cfg.url_params = Some("SSLVerification=NONE".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 15443, "hive");
let params = agent_connect_params(&cfg, "127.0.0.1", 15443, "hive").unwrap();
assert_eq!(params["connection_string"], "jdbc:trino://127.0.0.1:15443/hive?SSLVerification=NONE&SSL=true");
}
@@ -1289,7 +1322,7 @@ mod tests {
cfg.connection_string = Some("jdbc:trino://trino.internal:8443/hive?SSL=true&source=dbx".to_string());
cfg.url_params = Some("SSLVerification=NONE".to_string());
let params = agent_connect_params(&cfg, "127.0.0.1", 15443, "hive");
let params = agent_connect_params(&cfg, "127.0.0.1", 15443, "hive").unwrap();
assert_eq!(
params["connection_string"],
+30 -26
View File
@@ -618,15 +618,19 @@ pub fn upsert_connection_url_param(params: Option<&str>, key: &str, value: &str)
parts.join("&")
}
pub fn prestosql_jdbc_config_for_endpoint(config: &ConnectionConfig, host: &str, port: u16) -> ConnectionConfig {
pub fn prestosql_jdbc_config_for_endpoint(
config: &ConnectionConfig,
host: &str,
port: u16,
) -> Result<ConnectionConfig, String> {
let mut jdbc_config = config.clone();
jdbc_config.connection_string =
Some(trino_like_jdbc_connection_string(config, host, port, config.effective_database().unwrap_or("")));
Some(trino_like_jdbc_connection_string(config, host, port, config.effective_database().unwrap_or(""))?);
jdbc_config.url_params = None;
if jdbc_config.jdbc_driver_class.as_deref().is_none_or(|value| value.trim().is_empty()) {
jdbc_config.jdbc_driver_class = Some(PRESTOSQL_JDBC_DRIVER_CLASS.to_string());
}
jdbc_config
Ok(jdbc_config)
}
pub fn gaussdb_uses_m_jdbc_driver(config: &ConnectionConfig) -> bool {
@@ -1215,7 +1219,7 @@ impl AppState {
if sqlserver_uses_legacy_driver(config) {
let legacy_config = sqlserver_legacy_agent_config(config);
let connect_params =
agent_connect_params(&legacy_config, host, port, legacy_config.effective_database().unwrap_or(""));
agent_connect_params(&legacy_config, host, port, legacy_config.effective_database().unwrap_or(""))?;
let mut client = self
.agent_manager
.spawn(&legacy_config.db_type, legacy_config.driver_profile.as_deref())
@@ -1259,7 +1263,7 @@ impl AppState {
if sqlserver_uses_legacy_driver(config) {
let legacy_config = sqlserver_legacy_agent_config(config);
let connect_params =
agent_connect_params(&legacy_config, host, port, legacy_config.effective_database().unwrap_or(""));
agent_connect_params(&legacy_config, host, port, legacy_config.effective_database().unwrap_or(""))?;
let mut client = self
.agent_manager
.spawn(&legacy_config.db_type, legacy_config.driver_profile.as_deref())
@@ -1950,7 +1954,7 @@ impl AppState {
DatabaseType::MongoDb => {
if mongo_uses_legacy_driver(&db_config) {
log::info!("Using configured MongoDB legacy driver for connection_id={connection_id}");
let connect_params = serde_json::json!({ "connection": agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or("")) });
let connect_params = serde_json::json!({ "connection": agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or(""))? });
let mut client = self.agent_manager.spawn(&DatabaseType::MongoDb, Some("mongodb-legacy")).await?;
client.connect(connect_params).await.map_err(|err| mongo_legacy_error_with_auth_hint(&err))?;
PoolKind::agent(client)
@@ -1993,7 +1997,7 @@ impl AppState {
};
if should_retry_mongo_with_legacy_driver(&native_err) {
log::info!("Native MongoDB driver failed ({native_err}), falling back to agent driver");
let connect_params = serde_json::json!({ "connection": agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or("")) });
let connect_params = serde_json::json!({ "connection": agent_connect_params(&db_config, &host, port, db_config.effective_database().unwrap_or(""))? });
let legacy_agent_key =
AgentManager::db_type_to_agent_key(&DatabaseType::MongoDb, Some("mongodb-legacy"))
.ok_or_else(|| "MongoDB (Legacy) Agent mapping is unavailable".to_string())?;
@@ -2150,7 +2154,7 @@ impl AppState {
port,
db_config.effective_database().unwrap_or(""),
session_role,
);
)?;
if db_config.db_type != DatabaseType::ZooKeeper {
let agent_session_id = uuid::Uuid::new_v4().simple().to_string();
let mut initial_result = self
@@ -2185,7 +2189,7 @@ impl AppState {
port,
db_config.effective_database().unwrap_or(""),
session_role,
),
)?,
agent_connect_timeout(&db_config),
)
.await;
@@ -2217,7 +2221,7 @@ impl AppState {
port,
db_config.effective_database().unwrap_or(""),
session_role,
),
)?,
Some(agent_connect_timeout(&db_config)),
)
.await?;
@@ -2240,7 +2244,7 @@ impl AppState {
port,
alternate_config.effective_database().unwrap_or(""),
session_role,
);
)?;
match self
.spawn_routed_shared_agent_client(
&alternate_config.db_type,
@@ -2312,7 +2316,7 @@ impl AppState {
port,
alternate_config.effective_database().unwrap_or(""),
session_role,
),
)?,
Some(agent_connect_timeout(&alternate_config)),
)
.await
@@ -2340,14 +2344,14 @@ impl AppState {
}
}
DatabaseType::PrestoSql => {
let jdbc_config = prestosql_jdbc_config_for_endpoint(&db_config, &host, port);
let jdbc_config = prestosql_jdbc_config_for_endpoint(&db_config, &host, port)?;
self.external_driver_pool("jdbc", &jdbc_config).await?
}
DatabaseType::Jdbc => {
let mut jdbc_config = db_config.clone();
if host != config.host || port != config.port {
if let Some(ref url) = jdbc_config.connection_string {
jdbc_config.connection_string = Some(rewrite_jdbc_url_host(url, &host, port));
jdbc_config.connection_string = Some(rewrite_jdbc_url_host(url, &host, port)?);
}
}
self.external_driver_pool("jdbc", &jdbc_config).await?
@@ -5612,7 +5616,7 @@ mod tests {
config.host = "presto.example.com".to_string();
config.port = 9090;
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "127.0.0.1", 19090);
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "127.0.0.1", 19090).unwrap();
assert_eq!(jdbc_config.connection_string.as_deref(), Some("jdbc:presto://127.0.0.1:19090/hive/default"));
assert_eq!(jdbc_config.jdbc_driver_class.as_deref(), Some(PRESTOSQL_JDBC_DRIVER_CLASS));
@@ -5625,7 +5629,7 @@ mod tests {
config.jdbc_driver_class = Some("custom.PrestoDriver".to_string());
config.jdbc_driver_paths = vec!["D:\\software\\jar\\presto-jdbc-350.jar".to_string()];
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "presto.example.com", 9090);
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "presto.example.com", 9090).unwrap();
assert_eq!(jdbc_config.jdbc_driver_class.as_deref(), Some("custom.PrestoDriver"));
assert_eq!(jdbc_config.jdbc_driver_paths, vec!["D:\\software\\jar\\presto-jdbc-350.jar"]);
@@ -5640,7 +5644,7 @@ mod tests {
"SSL=true&SSLKeyStorePassword=secret&SSLKeyStorePath=D:/keystore/presto/presto_keystore.jks".to_string(),
);
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "presto.example.com", 8443);
let jdbc_config = prestosql_jdbc_config_for_endpoint(&config, "presto.example.com", 8443).unwrap();
assert_eq!(
jdbc_config.connection_string.as_deref(),
@@ -5816,7 +5820,7 @@ mod tests {
config.password = "in4mix".to_string();
config.url_params = Some("INFORMIXSERVER=informix;CLIENT_LOCALE=en_US.utf8".to_string());
let params = agent_connect_params(&config, "172.26.128.159", 20013, "testdb");
let params = agent_connect_params(&config, "172.26.128.159", 20013, "testdb").unwrap();
assert_eq!(params["host"], "172.26.128.159");
assert_eq!(params["port"], 20013);
@@ -5832,7 +5836,7 @@ mod tests {
config.db_type = DatabaseType::SqlServer;
config.external_config = Some(serde_json::json!({ "portExplicit": true }));
let params = agent_connect_params(&config, r"db.example.com\SQLEXPRESS", 1433, "master");
let params = agent_connect_params(&config, r"db.example.com\SQLEXPRESS", 1433, "master").unwrap();
assert_eq!(params["port_explicit"], true);
}
@@ -5890,7 +5894,7 @@ mod tests {
config.password = "secret".to_string();
config.url_params = Some("authSource=admin&authMechanism=SCRAM-SHA-1".to_string());
let params = agent_connect_params(&config, "172.22.4.42", 27017, "RestCloud_V45PUB_Gateway");
let params = agent_connect_params(&config, "172.22.4.42", 27017, "RestCloud_V45PUB_Gateway").unwrap();
assert_eq!(params["connection_string"], "mongodb://mongouser:secret@172.22.4.42:27017/RestCloud%5FV45PUB%5FGateway?authSource=admin&authMechanism=SCRAM-SHA-1");
}
@@ -5902,7 +5906,7 @@ mod tests {
config.connection_string =
Some("mongodb://mongouser:secret@172.22.4.42:27017/RestCloud_V45PUB_Gateway?authSource=admin".to_string());
let params = agent_connect_params(&config, "172.22.4.42", 27017, "");
let params = agent_connect_params(&config, "172.22.4.42", 27017, "").unwrap();
assert_eq!(params["database"], "RestCloud_V45PUB_Gateway");
}
@@ -6000,7 +6004,7 @@ mod tests {
config.sysdba = true;
config.oracle_connection_type = Some("service_name".to_string());
let params = agent_connect_params(&config, "oracle.example.com", 1521, "ORCLPDB1");
let params = agent_connect_params(&config, "oracle.example.com", 1521, "ORCLPDB1").unwrap();
assert_eq!(params["database"], "SYSDBA:ORCLPDB1");
assert_eq!(params["sysdba"], true);
@@ -6043,7 +6047,7 @@ mod tests {
config.url_params = Some("sslmode=disable".to_string());
config.connection_string = Some(stale_connection_string.to_string());
let params = agent_connect_params(&config, host, port, "platform_face_freezer_jgj");
let params = agent_connect_params(&config, host, port, "platform_face_freezer_jgj").unwrap();
assert_eq!(params["database"], "platform_face_freezer_jgj");
assert_eq!(params["connection_string"], expected_connection_string);
@@ -6056,7 +6060,7 @@ mod tests {
config.db_type = DatabaseType::Oracle;
config.oracle_connection_type = Some("sid".to_string());
let params = agent_connect_params(&config, "127.0.0.1", 11521, "ORCL");
let params = agent_connect_params(&config, "127.0.0.1", 11521, "ORCL").unwrap();
assert_eq!(params["connection_string"], "jdbc:oracle:thin:@127.0.0.1:11521:ORCL");
}
@@ -6067,7 +6071,7 @@ mod tests {
config.db_type = DatabaseType::Oracle;
config.oracle_connection_type = None;
let params = agent_connect_params(&config, "127.0.0.1", 11521, "ORCL");
let params = agent_connect_params(&config, "127.0.0.1", 11521, "ORCL").unwrap();
assert_eq!(params["connection_string"], "jdbc:oracle:thin:@//127.0.0.1:11521/ORCL");
}
@@ -6128,7 +6132,7 @@ mod tests {
config.password = "secret".to_string();
config.url_params = Some("encrypt=true".to_string());
let params = agent_connect_params(&config, "hana.example.com", 30013, "TENANT1");
let params = agent_connect_params(&config, "hana.example.com", 30013, "TENANT1").unwrap();
assert_eq!(params["database"], "TENANT1");
assert_eq!(params["connection_string"], "jdbc:sap://hana.example.com:30013/?databaseName=TENANT1&encrypt=true");
+79 -16
View File
@@ -2260,7 +2260,7 @@ fn jdbc_transport_rest(url: &str) -> Option<&str> {
}
}
pub fn rewrite_jdbc_url_host(url: &str, new_host: &str, new_port: u16) -> String {
pub fn rewrite_jdbc_url_host(url: &str, new_host: &str, new_port: u16) -> Result<String, String> {
let normalized_url = url.to_ascii_uppercase();
let normalized_rest = jdbc_transport_rest(url).map(str::to_ascii_uppercase).unwrap_or_default();
if normalized_rest.starts_with("ORACLE:") && normalized_url.contains("(HOST=") && normalized_url.contains("(PORT=")
@@ -2269,11 +2269,41 @@ pub fn rewrite_jdbc_url_host(url: &str, new_host: &str, new_port: u16) -> String
}
let Some((old_host, old_port)) = parse_jdbc_host_port(url) else {
return url.to_string();
return Err(format!(
"Cannot route JDBC connection string through transport endpoint {new_host}:{new_port}: unsupported URL format"
));
};
if let Some(after_sqlserver) = normalized_rest.strip_prefix("SQLSERVER://") {
let authority = after_sqlserver.split(';').next().unwrap_or_default();
if authority.contains('\\') {
return Err(format!(
"Cannot route JDBC connection string through transport endpoint {new_host}:{new_port}: SQL Server named instances are not supported"
));
}
if normalized_url.contains(";FAILOVERPARTNER=") {
return Err(format!(
"Cannot route JDBC connection string through transport endpoint {new_host}:{new_port}: SQL Server failover partner addresses cannot be safely routed"
));
}
}
let old_authority = format!("{old_host}:{old_port}");
let new_authority = format!("{new_host}:{new_port}");
url.replacen(&old_authority, &new_authority, 1)
let new_authority = format!("{}:{new_port}", bracket_ipv6(new_host));
let rewritten = url.replacen(&old_authority, &new_authority, 1);
if rewritten == url {
return Err(format!(
"Cannot route JDBC connection string through transport endpoint {new_host}:{new_port}: endpoint {old_authority} is not present in the URL"
));
}
match parse_jdbc_host_port(&rewritten) {
Some((host, port)) if host.trim_matches(['[', ']']) == new_host.trim_matches(['[', ']']) && port == new_port => {
Ok(rewritten)
}
_ => Err(format!(
"Cannot route JDBC connection string through transport endpoint {new_host}:{new_port}: the rewritten URL does not resolve to a single local endpoint"
)),
}
}
fn oracle_descriptor_value(descriptor: &str, key: &str) -> Option<String> {
@@ -2284,9 +2314,21 @@ fn oracle_descriptor_value(descriptor: &str, key: &str) -> Option<String> {
Some(descriptor[value_start..value_end].trim().to_string())
}
fn rewrite_oracle_descriptor_host(url: &str, new_host: &str, new_port: u16) -> String {
fn rewrite_oracle_descriptor_host(url: &str, new_host: &str, new_port: u16) -> Result<String, String> {
let upper = url.to_ascii_uppercase();
let address_count = upper.matches("(ADDRESS=").count();
if address_count != 1 {
return Err(format!(
"Cannot route Oracle JDBC descriptor through transport endpoint {new_host}:{new_port}: descriptor must contain a single ADDRESS entry, found {address_count}"
));
}
if upper.matches("(HOST=").count() != 1 || upper.matches("(PORT=").count() != 1 {
return Err(format!(
"Cannot route Oracle JDBC descriptor through transport endpoint {new_host}:{new_port}: descriptor must contain exactly one HOST and one PORT"
));
}
let rewritten_host = replace_oracle_descriptor_value(url, "HOST", new_host);
replace_oracle_descriptor_value(&rewritten_host, "PORT", &new_port.to_string())
Ok(replace_oracle_descriptor_value(&rewritten_host, "PORT", &new_port.to_string()))
}
fn replace_oracle_descriptor_value(input: &str, key: &str, value: &str) -> String {
@@ -3965,7 +4007,7 @@ mod tests {
"jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=orahost)(PORT=1521))(CONNECT_DATA=(SERVICE_NAME=orcl)))";
assert_eq!(
super::rewrite_jdbc_url_host(url, "127.0.0.1", 11521),
super::rewrite_jdbc_url_host(url, "127.0.0.1", 11521).unwrap(),
"jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=11521))(CONNECT_DATA=(SERVICE_NAME=orcl)))"
);
}
@@ -3997,25 +4039,25 @@ mod tests {
#[test]
fn rewrite_jdbc_url_postgresql() {
let url = "jdbc:postgresql://myhost:5432/mydb";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321);
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321).unwrap();
assert_eq!(rewritten, "jdbc:postgresql://127.0.0.1:54321/mydb");
}
#[test]
fn rewrite_jdbcx_url_preserves_extension() {
let url = "jdbcx:script:mysql://db.example.com:3306/app";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 13306);
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 13306).unwrap();
assert_eq!(rewritten, "jdbcx:script:mysql://127.0.0.1:13306/app");
let sqlserver = "jdbcx:query:sqlserver://sql.example.com:1433;databaseName=app";
let rewritten = super::rewrite_jdbc_url_host(sqlserver, "127.0.0.1", 11433);
let rewritten = super::rewrite_jdbc_url_host(sqlserver, "127.0.0.1", 11433).unwrap();
assert_eq!(rewritten, "jdbcx:query:sqlserver://127.0.0.1:11433;databaseName=app");
}
#[test]
fn rewrite_jdbcx_oracle_descriptor() {
let url = "jdbcx:web:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=orahost)(PORT=1521))(CONNECT_DATA=(SERVICE_NAME=orcl)))";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 11521);
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 11521).unwrap();
assert_eq!(
rewritten,
"jdbcx:web:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT=11521))(CONNECT_DATA=(SERVICE_NAME=orcl)))"
@@ -4025,21 +4067,42 @@ mod tests {
#[test]
fn rewrite_jdbc_url_oracle() {
let url = "jdbc:oracle:thin:@orahost:1521:ORCL";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321);
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321).unwrap();
assert_eq!(rewritten, "jdbc:oracle:thin:@127.0.0.1:54321:ORCL");
}
#[test]
fn rewrite_jdbc_url_sqlserver() {
let url = "jdbc:sqlserver://mshost:1433;databaseName=master";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321);
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321).unwrap();
assert_eq!(rewritten, "jdbc:sqlserver://127.0.0.1:54321;databaseName=master");
}
#[test]
fn rewrite_jdbc_url_unparseable_returns_original() {
fn rewrite_jdbc_url_unparseable_errors() {
let url = "jdbc:custom:some-opaque-string";
let rewritten = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321);
assert_eq!(rewritten, url);
let err = super::rewrite_jdbc_url_host(url, "127.0.0.1", 54321).unwrap_err();
assert!(err.contains("unsupported URL format"), "{err}");
}
#[test]
fn rewrite_jdbc_url_host_rejects_sqlserver_named_instance() {
let url = r"jdbc:sqlserver://db\SQLEXPRESS:1433;databaseName=app";
let err = super::rewrite_jdbc_url_host(url, "127.0.0.1", 45678).unwrap_err();
assert!(err.to_lowercase().contains("named instance"), "{err}");
}
#[test]
fn rewrite_jdbc_url_host_rejects_sqlserver_failover_partner() {
let url = "jdbc:sqlserver://db1:1433;failoverPartner=db2:1433;databaseName=app";
let err = super::rewrite_jdbc_url_host(url, "127.0.0.1", 45678).unwrap_err();
assert!(err.contains("failover partner"), "{err}");
}
#[test]
fn rewrite_jdbc_url_host_rejects_multi_address_oracle_descriptor() {
let url = "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=db1)(PORT=1521))(ADDRESS=(PROTOCOL=TCP)(HOST=db2)(PORT=1521)))(CONNECT_DATA=(SERVICE_NAME=orcl)))";
let err = super::rewrite_jdbc_url_host(url, "127.0.0.1", 11521).unwrap_err();
assert!(err.contains("single ADDRESS entry"), "{err}");
}
}