feat(mssql-cdc): add ssl support for sqlserver-cdc (#18912)

risingwavelabs · Oct 16, 2024 · 7920e3d · 7920e3d
1 parent 360fe8a
commit 7920e3d
Show file tree

Hide file tree

Showing 8 changed files with 21 additions and 5 deletions.
diff --git a/...ctor-service/src/main/java/com/risingwave/connector/source/common/DbzConnectorConfig.java b/...ctor-service/src/main/java/com/risingwave/connector/source/common/DbzConnectorConfig.java
@@ -58,6 +58,7 @@ public class DbzConnectorConfig {
 
     /* Sql Server configs */
     public static final String SQL_SERVER_SCHEMA_NAME = "schema.name";
+    public static final String SQL_SERVER_ENCRYPT = "database.encrypt";
 
     /* RisingWave configs */
     private static final String DBZ_CONFIG_FILE = "debezium.properties";

diff --git a/...ctor-service/src/main/java/com/risingwave/connector/source/common/SqlServerValidator.java b/...ctor-service/src/main/java/com/risingwave/connector/source/common/SqlServerValidator.java
@@ -50,8 +50,12 @@ public SqlServerValidator(
         var dbName = userProps.get(DbzConnectorConfig.DB_NAME);
         var user = userProps.get(DbzConnectorConfig.USER);
         var password = userProps.get(DbzConnectorConfig.PASSWORD);
+        var encrypt =
+                Boolean.parseBoolean(
+                        userProps.getOrDefault(DbzConnectorConfig.SQL_SERVER_ENCRYPT, "false"));
 
         var jdbcUrl = ValidatorUtils.getJdbcUrl(SourceTypeE.SQL_SERVER, dbHost, dbPort, dbName);
+        jdbcUrl = jdbcUrl + ";encrypt=" + encrypt + ";trustServerCertificate=true";
         this.jdbcConnection = DriverManager.getConnection(jdbcUrl, user, password);
 
         this.dbName = dbName;

diff --git a/...onnector-service/src/main/java/com/risingwave/connector/source/common/ValidatorUtils.java b/...onnector-service/src/main/java/com/risingwave/connector/source/common/ValidatorUtils.java
@@ -68,8 +68,7 @@ public static String getJdbcUrl(
                 return String.format("jdbc:postgresql://%s:%s/%s", host, port, database);
             case SQL_SERVER:
                 return String.format(
-                        "jdbc:sqlserver://%s:%s;databaseName=%s;encrypt=false",
-                        host, port, database);
+                        "jdbc:sqlserver://%s:%s;databaseName=%s", host, port, database);
             default:
                 throw ValidatorUtils.invalidArgument("Unknown source type: " + sourceType);
         }

diff --git a/java/connector-node/risingwave-connector-service/src/main/resources/sql_server.properties b/java/connector-node/risingwave-connector-service/src/main/resources/sql_server.properties
@@ -21,4 +21,5 @@ name=${hostname}:${port}:${database.name}.${schema.name}.${table.name:-RW_CDC_Sh
 # In sharing cdc mode, transaction metadata will be enabled in frontend.
 # For sql server, it's always false actually.
 provide.transaction.metadata=${transactional:-false}
-database.encrypt=false
+database.encrypt=${database.encrypt:-false}
+database.trustServerCertificate=true
diff --git a/src/connector/src/source/cdc/external/mod.rs b/src/connector/src/source/cdc/external/mod.rs
@@ -243,6 +243,11 @@ pub struct ExternalTableConfig {
     #[serde(rename = "ssl.root.cert")]
     #[serde(alias = "debezium.database.sslrootcert")]
     pub ssl_root_cert: Option<String>,
+
+    /// `encrypt` specifies whether connect to SQL Server using SSL.
+    /// Only "true" means using SSL. All other values are treated as "false".
+    #[serde(rename = "database.encrypt", default = "Default::default")]
+    pub encrypt: String,
 }
 
 impl ExternalTableConfig {

diff --git a/src/connector/src/source/cdc/external/mysql.rs b/src/connector/src/source/cdc/external/mysql.rs
@@ -598,6 +598,7 @@ mod tests {
             table: "part".to_string(),
             ssl_mode: Default::default(),
             ssl_root_cert: None,
+            encrypt: "false".to_string(),
         };
 
         let table = MySqlExternalTable::connect(config).await.unwrap();

diff --git a/src/connector/src/source/cdc/external/postgres.rs b/src/connector/src/source/cdc/external/postgres.rs
@@ -540,6 +540,7 @@ mod tests {
             table: "mytest".to_string(),
             ssl_mode: Default::default(),
             ssl_root_cert: None,
+            encrypt: "false".to_string(),
         };
 
         let table = PostgresExternalTable::connect(config).await.unwrap();

diff --git a/src/connector/src/source/cdc/external/sql_server.rs b/src/connector/src/source/cdc/external/sql_server.rs
@@ -90,8 +90,10 @@ impl SqlServerExternalTable {
             &config.username,
             &config.password,
         ));
-        // TODO(kexiang): add ssl support
         // TODO(kexiang): use trust_cert_ca, trust_cert is not secure
+        if config.encrypt == "true" {
+            client_config.encryption(tiberius::EncryptionLevel::Required);
+        }
         client_config.trust_cert();
 
         let mut client = SqlServerClient::new_with_config(client_config).await?;
@@ -282,8 +284,10 @@ impl SqlServerExternalTableReader {
             &config.username,
             &config.password,
         ));
-        // TODO(kexiang): add ssl support
         // TODO(kexiang): use trust_cert_ca, trust_cert is not secure
+        if config.encrypt == "true" {
+            client_config.encryption(tiberius::EncryptionLevel::Required);
+        }
         client_config.trust_cert();
 
         let client = SqlServerClient::new_with_config(client_config).await?;