add unit test and docs

Change-Id: I38146ee45a4565295fa6fd297f591c368d6b250a

Author: jerryshao

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

@@ -328,7 +328,6 @@ private[spark] class SecurityManager(
         .orElse(Option(secretKey))
         .orElse(Option(sparkConf.getenv(ENV_AUTH_SECRET)))
         .orElse(sparkConf.getOption(SPARK_AUTH_SECRET_CONF))
-        .orElse(Option(hadoopConf.getPassword(SPARK_AUTH_SECRET_CONF)).map(new String(_)))
         .getOrElse {
           throw new IllegalArgumentException(
             s"A secret key must be specified via the $SPARK_AUTH_SECRET_CONF config")

core/src/test/scala/org/apache/spark/SSLOptionsSuite.scala

@@ -18,9 +18,11 @@
 package org.apache.spark
 
 import java.io.File
+import java.util.UUID
 import javax.net.ssl.SSLContext
 
 import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.alias.{CredentialProvider, CredentialProviderFactory}
 import org.scalatest.BeforeAndAfterAll
 
 import org.apache.spark.util.SparkConfWithEnv
@@ -154,4 +156,60 @@ class SSLOptionsSuite extends SparkFunSuite with BeforeAndAfterAll {
     assert(opts.trustStore === Some(new File("val2")))
   }
 
+  test("get password from Hadoop credential provider") {
+    val keyStorePath = new File(this.getClass.getResource("/keystore").toURI).getAbsolutePath
+    val trustStorePath = new File(this.getClass.getResource("/truststore").toURI).getAbsolutePath
+
+    val conf = new SparkConf
+    val hadoopConf = new Configuration()
+    val tmpPath = s"localjceks://file${sys.props("java.io.tmpdir")}/test-" +
+      s"${UUID.randomUUID().toString}.jceks"
+    val provider = createCredentialProvider(tmpPath, hadoopConf)
+
+    conf.set("spark.ssl.enabled", "true")
+    conf.set("spark.ssl.keyStore", keyStorePath)
+    storePassword(provider, "spark.ssl.keyStorePassword", "password")
+    storePassword(provider, "spark.ssl.keyPassword", "password")
+    conf.set("spark.ssl.trustStore", trustStorePath)
+    storePassword(provider, "spark.ssl.trustStorePassword", "password")
+    conf.set("spark.ssl.enabledAlgorithms",
+      "TLS_RSA_WITH_AES_128_CBC_SHA, TLS_RSA_WITH_AES_256_CBC_SHA")
+    conf.set("spark.ssl.protocol", "SSLv3")
+
+    val defaultOpts = SSLOptions.parse(conf, hadoopConf, "spark.ssl", defaults = None)
+    val opts = SSLOptions.parse(conf, hadoopConf, "spark.ssl.ui", defaults = Some(defaultOpts))
+
+    assert(opts.enabled === true)
+    assert(opts.trustStore.isDefined === true)
+    assert(opts.trustStore.get.getName === "truststore")
+    assert(opts.trustStore.get.getAbsolutePath === trustStorePath)
+    assert(opts.keyStore.isDefined === true)
+    assert(opts.keyStore.get.getName === "keystore")
+    assert(opts.keyStore.get.getAbsolutePath === keyStorePath)
+    assert(opts.trustStorePassword === Some("password"))
+    assert(opts.keyStorePassword === Some("password"))
+    assert(opts.keyPassword === Some("password"))
+    assert(opts.protocol === Some("SSLv3"))
+    assert(opts.enabledAlgorithms ===
+      Set("TLS_RSA_WITH_AES_128_CBC_SHA", "TLS_RSA_WITH_AES_256_CBC_SHA"))
+  }
+
+  private def createCredentialProvider(tmpPath: String, conf: Configuration): CredentialProvider = {
+    conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, tmpPath)
+
+    val provider = CredentialProviderFactory.getProviders(conf).get(0)
+    if (provider == null) {
+      throw new IllegalStateException(s"Fail to get credential provider with path $tmpPath")
+    }
+
+    provider
+  }
+
+  private def storePassword(
+      provider: CredentialProvider,
+      passwordKey: String,
+      password: String): Unit = {
+    provider.createCredentialEntry(passwordKey, password.toCharArray)
+    provider.flush()
+  }
 }

docs/security.md

@@ -177,7 +177,7 @@ ACLs can be configured for either users or groups. Configuration entries accept
 lists as input, meaning multiple users or groups can be given the desired privileges. This can be
 used if you run on a shared cluster and have a set of administrators or developers who need to
 monitor applications they may not have started themselves. A wildcard (`*`) added to specific ACL
-means that all users will have the respective pivilege. By default, only the user submitting the
+means that all users will have the respective privilege. By default, only the user submitting the
 application is added to the ACLs.
 
 Group membership is established by using a configurable group mapping provider. The mapper is
@@ -446,6 +446,18 @@ replaced with one of the above namespaces.
   </tr>
 </table>
 
+Spark also supports retrieving `${ns}.keyPassword`, `${ns}.keyStorePassword` and `${ns}.trustStorePassword` from
+[Hadoop Credential Providers](https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html).
+User could store password into credential file and make it accessible by different components, like:
+
+```
+hadoop credential create spark.ssl.keyPassword -value password \
+    -provider jceks://[email protected]:9001/user/backup/ssl.jceks
+```
+
+In the meantime, adding configuration "hadoop.security.credential.provider.path=jceks://[email protected]:9001/user/backup/ssl.jceks"
+into Spark's Hadoop configuration to make it aware of credential provider path.
+
 ## Preparing the key stores
 
 Key stores can be generated by `keytool` program. The reference documentation for this tool for