SPARK-3883: Fixes after the review

conf/ssl.conf.template

@@ -1,20 +1,3 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#    http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
 # Spark SSL settings
 
 # ssl.enabled               true
@@ -24,4 +7,4 @@
 # ssl.trustStore            /path/to/your/trustStore
 # ssl.trustStorePassword    password
 # ssl.enabledAlgorithms     [TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA]
-# ssl.protocol              SSLv3
+# ssl.protocol              TLSv1.2

core/src/main/scala/org/apache/spark/SSLOptions.scala

@@ -18,11 +18,9 @@
 package org.apache.spark
 
 import java.io.{FileReader, File}
-import java.nio.file.Paths
 import java.util.Properties
 
 import com.typesafe.config.{Config, ConfigFactory, ConfigValueFactory}
-import org.apache.commons.io.FilenameUtils
 import org.eclipse.jetty.util.ssl.SslContextFactory
 
 import scala.util.Try
@@ -138,23 +136,14 @@ object SSLOptions extends Logging {
    * as a base directory to resolve relative paths to keystore and truststore.
    */
   def parse(conf: SparkConf, ns: String): SSLOptions = {
-    val configFilePath = conf.getOption("sslConfigurationFileLocation")
-
-    def makeFile(pathString: String): File = {
-      val path = Paths.get(pathString)
-
-      if (path.isAbsolute || configFilePath.isEmpty) {
-        path.toFile
-      } else {
-        new File(FilenameUtils.concat(new File(configFilePath.get).getParent, pathString))
-      }
-    }
+    val parentDir = conf.getOption("sslConfigurationFileLocation").map(new File(_).getParentFile)
+      .getOrElse(new File(".")).toPath
 
     val enabled = conf.getBoolean(s"$ns.enabled", defaultValue = false)
-    val keyStore = Try(conf.get(s"$ns.keyStore")).toOption.map(makeFile)
+    val keyStore = Try(conf.get(s"$ns.keyStore")).toOption.map(parentDir.resolve(_).toFile)
     val keyStorePassword = Try(conf.get(s"$ns.keyStorePassword")).toOption
     val keyPassword = Try(conf.get(s"$ns.keyPassword")).toOption
-    val trustStore = Try(conf.get(s"$ns.trustStore")).toOption.map(makeFile)
+    val trustStore = Try(conf.get(s"$ns.trustStore")).toOption.map(parentDir.resolve(_).toFile)
     val trustStorePassword = Try(conf.get(s"$ns.trustStorePassword")).toOption
     val protocol = Try(conf.get(s"$ns.protocol")).toOption
     val enabledAlgorithms = Try(conf.get(s"$ns.enabledAlgorithms")).toOption

core/src/main/scala/org/apache/spark/SecurityManager.scala

@@ -22,7 +22,7 @@ import java.security.KeyStore
 import java.security.cert.X509Certificate
 import javax.net.ssl._
 
-import org.apache.commons.io.FileUtils
+import com.google.common.io.Files
 import org.apache.hadoop.io.Text
 
 import org.apache.spark.deploy.SparkHadoopUtil
@@ -196,13 +196,13 @@ private[spark] class SecurityManager(sparkConf: SparkConf) extends Logging {
     )
   }
 
-  private[spark] val sslOptions = SSLOptions.parse(ns = "ssl", conf = sparkConf)
+  private[spark] val sslOptions = SSLOptions.parse(sparkConf, "ssl")
 
   private[spark] val (sslSocketFactory, hostnameVerifier) = if (sslOptions.enabled) {
     val trustStoreManagers =
       for (trustStore <- sslOptions.trustStore) yield {
         val ks = KeyStore.getInstance(KeyStore.getDefaultType)
-        ks.load(FileUtils.openInputStream(sslOptions.trustStore.get),
+        ks.load(Files.asByteSource(sslOptions.trustStore.get).openStream(),
           sslOptions.trustStorePassword.get.toCharArray)
 
         val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
@@ -230,7 +230,6 @@ private[spark] class SecurityManager(sparkConf: SparkConf) extends Logging {
 
     (Some(sslContext.getSocketFactory), Some(hostVerifier))
   } else {
-    val sslContext = SSLContext.getDefault
     (None, None)
   }
 

core/src/test/resources/bad-ssl.conf

@@ -24,4 +24,4 @@ ssl.keyPassword           password
 ssl.trustStore            truststore
 ssl.trustStorePassword    password
 ssl.enabledAlgorithms     [TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA]
-ssl.protocol              SSLv3
+ssl.protocol              TLSv1.2

core/src/test/resources/good-ssl.conf

@@ -24,4 +24,4 @@ ssl.keyPassword           password
 ssl.trustStore            truststore
 ssl.trustStorePassword    password
 ssl.enabledAlgorithms     [TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA]
-ssl.protocol              SSLv3
+ssl.protocol              TLSv1.2