Merge branch 'feature/ssl-support' into feature/percent_random

* feature/ssl-support:
  Added support for custom cipher algorithms (needed when limited ciphers are allowed)

Author: pravinbhat

src/main/scala/datastax/astra/migrate/AbstractJob.scala

@@ -26,6 +26,7 @@ class AbstractJob extends App {
   val sourceTrustStoreType = sc.getConf.get("spark.migrate.source.trustStore.type", "JKS")
   val sourceKeyStorePath = sc.getConf.get("spark.migrate.source.keyStore.path", "")
   val sourceKeyStorePassword = sc.getConf.get("spark.migrate.source.keyStore.password", "")
+  val sourceEnabledAlgorithms = sc.getConf.get("spark.migrate.source.enabledAlgorithms", "")
 
   val destinationIsAstra = sc.getConf.get("spark.migrate.destination.isAstra", "true")
   val destinationScbPath = sc.getConf.get("spark.migrate.destination.scb", "")
@@ -38,17 +39,18 @@ class AbstractJob extends App {
   val destinationTrustStoreType = sc.getConf.get("spark.migrate.destination.trustStore.type", "JKS")
   val destinationKeyStorePath = sc.getConf.get("spark.migrate.destination.keyStore.path", "")
   val destinationKeyStorePassword = sc.getConf.get("spark.migrate.destination.keyStore.password", "")
+  val destinationEnabledAlgorithms = sc.getConf.get("spark.migrate.destination.enabledAlgorithms", "")
 
   val minPartition = new BigInteger(sc.getConf.get("spark.migrate.source.minPartition"))
   val maxPartition = new BigInteger(sc.getConf.get("spark.migrate.source.maxPartition"))
 
   val splitSize = sc.getConf.get("spark.migrate.splitSize", "10000")
 
   var sourceConnection = getConnection(true, sourceIsAstra, sourceScbPath, sourceHost, sourceUsername, sourcePassword, sourceReadConsistencyLevel,
-    sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword);
+    sourceTrustStorePath, sourceTrustStorePassword, sourceTrustStoreType, sourceKeyStorePath, sourceKeyStorePassword, sourceEnabledAlgorithms);
 
   var destinationConnection = getConnection(false, destinationIsAstra, destinationScbPath, destinationHost, destinationUsername, destinationPassword, destinationReadConsistencyLevel,
-    destinationTrustStorePath, destinationTrustStorePassword, destinationTrustStoreType, destinationKeyStorePath, destinationKeyStorePassword);
+    destinationTrustStorePath, destinationTrustStorePassword, destinationTrustStoreType, destinationKeyStorePath, destinationKeyStorePassword, destinationEnabledAlgorithms);
 
   protected def exitSpark() = {
     spark.stop()
@@ -57,7 +59,7 @@ class AbstractJob extends App {
 
   private def getConnection(isSource: Boolean, isAstra: String, scbPath: String, host: String, username: String, password: String, readConsistencyLevel: String,
                             trustStorePath: String, trustStorePassword: String, trustStoreType: String,
-                            keyStorePath: String, keyStorePassword: String): CassandraConnector = {
+                            keyStorePath: String, keyStorePassword: String, enabledAlgorithms: String): CassandraConnector = {
     var connType: String = "Source"
     if (!isSource) {
       connType = "Destination"
@@ -74,17 +76,24 @@ class AbstractJob extends App {
     } else if (null != trustStorePath && !trustStorePath.trim.isEmpty) {
       abstractLogger.info(connType + ": Connected to Cassandra (or DSE) with SSL!");
 
+      // Use defaults when not provided
+      var enabledAlgorithmsVar = enabledAlgorithms
+      if (enabledAlgorithms == null || enabledAlgorithms.trim.isEmpty) {
+        enabledAlgorithmsVar = "TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA"
+      }
+
       return CassandraConnector(sc.getConf
         .set("spark.cassandra.auth.username", username)
         .set("spark.cassandra.auth.password", password)
         .set("spark.cassandra.input.consistency.level", readConsistencyLevel)
         .set("spark.cassandra.connection.host", host)
         .set("spark.cassandra.connection.ssl.enabled", "true")
+        .set("spark.cassandra.connection.ssl.enabledAlgorithms", enabledAlgorithmsVar)
         .set("spark.cassandra.connection.ssl.trustStore.password", trustStorePassword)
         .set("spark.cassandra.connection.ssl.trustStore.path", trustStorePath)
         .set("spark.cassandra.connection.ssl.keyStore.password", keyStorePassword)
         .set("spark.cassandra.connection.ssl.keyStore.path", keyStorePath)
-        .set("spark.cassandra.connection.ssl.trustStore.type", trustStoreType) 
+        .set("spark.cassandra.connection.ssl.trustStore.type", trustStoreType)
         .set("spark.cassandra.connection.ssl.clientAuth.enabled", "true")
       )
     } else {

src/resources/sparkConf.properties

@@ -49,13 +49,15 @@ spark.migrate.source.maxWriteTimeStampFilter                    9223372036854775
 #spark.migrate.source.trustStore.type                            JKS
 #spark.migrate.source.keyStore.path
 #spark.migrate.source.keyStore.password
+#spark.migrate.source.enabledAlgorithms                          TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA
 
 ####################### ONLY USE if SSL clientAuth is enabled on destination Cassandra/DSE #############################
 #spark.migrate.destination.trustStore.path
 #spark.migrate.destination.trustStore.password
 #spark.migrate.destination.trustStore.type                       JKS
 #spark.migrate.destination.keyStore.path
 #spark.migrate.destination.keyStore.password
+#spark.migrate.destination.enabledAlgorithms                     TLS_RSA_WITH_AES_128_CBC_SHA,TLS_RSA_WITH_AES_256_CBC_SHA
 
 ########################################################################################################################
 # Following are the supported data types and their corresponding [Cassandra data-types]