[SPARK-58776][K8S] Propagate spark.ssl.rpc.* passwords to executors

docs/security.md

@@ -183,6 +183,14 @@ Unlike the other SSL settings for the UI, the RPC SSL is *not* automatically ena
 `spark.ssl.enabled` is set. It must be explicitly enabled, to ensure a safe migration path for users
 upgrading Spark versions.
 
+On Kubernetes, the passwords used by RPC SSL are propagated to executor pods using environment
+variables, the same way as the authentication mechanics described under [Kubernetes](#kubernetes)
+above. As with the authentication secret, any user who can list pods in the namespace can read
+them, so access control rules should be properly set up or alternatively these passwords should
+be supplied using `spark.kubernetes.executor.secretKeyRef`, for example
+`spark.kubernetes.executor.secretKeyRef._SPARK_SSL_RPC_KEY_STORE_PASSWORD=<secret-name>:<key>`.
+Binding variables this way takes precedence.
+
 ## AES-based Encryption (Legacy)
 
 Spark supports AES-based encryption for RPC connections. For encryption to be enabled, RPC

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala

@@ -143,6 +143,20 @@ private[spark] class BasicExecutorFeatureStep(
         case _ => Nil
       }.getOrElse(Nil)
 
+      // SparkConf.isExecutorStartupConf withholds the spark.ssl.* passwords from the
+      // executor conf. Pass them through the environment, as the standalone worker
+      // does in CommandUtils. A name the user already binds, via
+      // spark.kubernetes.executor.secretKeyRef, spark.executorEnv or the pod template,
+      // is skipped and the user's value wins: buildEnvVars does not deduplicate and
+      // Kubernetes resolves a repeated name last-wins.
+      val userBoundEnvNames = kubernetesConf.secretEnvNamesToKeyRefs.keySet ++
+        kubernetesConf.environment.keySet ++
+        Option(pod.container).flatMap(c => Option(c.getEnv))
+          .map(_.asScala.map(_.getName).toSet).getOrElse(Set.empty)
+      val sslRpcPasswords = secMgr.getEnvironmentForSslRpcPasswords.filterNot {
+        case (name, _) => userBoundEnvNames.contains(name)
+      }.toSeq
+
       val userOpts = kubernetesConf.get(EXECUTOR_JAVA_OPTIONS).toSeq.flatMap { opts =>
         val subsOpts = Utils.substituteAppNExecIds(opts, kubernetesConf.appId,
           kubernetesConf.executorId)
@@ -187,6 +201,7 @@ private[spark] class BasicExecutorFeatureStep(
           ++ attributes
           ++ kubernetesConf.environment
           ++ sparkAuthSecret
+          ++ sslRpcPasswords
           ++ Seq(ENV_CLASSPATH -> kubernetesConf.get(EXECUTOR_CLASS_PATH).orNull)
           ++ allOpts) ++
       KubernetesUtils.buildEnvVarsWithFieldRef(

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala

@@ -24,7 +24,7 @@ import com.google.common.net.InternetDomainName
 import io.fabric8.kubernetes.api.model._
 import org.scalatest.BeforeAndAfter
 
-import org.apache.spark.{SecurityManager, SparkConf, SparkException, SparkFunSuite, SparkIllegalArgumentException}
+import org.apache.spark.{SecurityManager, SparkConf, SparkException, SparkFunSuite, SparkIllegalArgumentException, SSLOptions}
 import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesTestConf, SecretVolumeUtils, SparkPod}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
@@ -417,6 +417,118 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
     }
   }
 
+  test("SSL RPC password propagation") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.rpc.enabled", "true")
+      .set("spark.ssl.rpc.keyStorePassword", "keyStorePass")
+      .set("spark.ssl.rpc.keyPassword", "keyPass")
+      .set("spark.ssl.rpc.privateKeyPassword", "privateKeyPass")
+      .set("spark.ssl.rpc.trustStorePassword", "trustStorePass")
+
+    val step = new BasicExecutorFeatureStep(KubernetesTestConf.createExecutorConf(sparkConf = conf),
+      new SecurityManager(conf), defaultProfile)
+
+    val executor = step.configurePod(SparkPod.initialPod())
+    checkEnv(executor, conf, Map(
+      SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD -> "keyStorePass",
+      SSLOptions.ENV_RPC_SSL_KEY_PASSWORD -> "keyPass",
+      SSLOptions.ENV_RPC_SSL_PRIVATE_KEY_PASSWORD -> "privateKeyPass",
+      SSLOptions.ENV_RPC_SSL_TRUST_STORE_PASSWORD -> "trustStorePass"))
+  }
+
+  test("SSL RPC passwords shouldn't propagate if RPC SSL is disabled") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.rpc.enabled", "false")
+      .set("spark.ssl.rpc.keyStorePassword", "keyStorePass")
+      .set("spark.ssl.rpc.trustStorePassword", "trustStorePass")
+
+    val step = new BasicExecutorFeatureStep(KubernetesTestConf.createExecutorConf(sparkConf = conf),
+      new SecurityManager(conf), defaultProfile)
+
+    val executor = step.configurePod(SparkPod.initialPod())
+    SSLOptions.SPARK_RPC_SSL_PASSWORD_ENVS.foreach { env =>
+      assert(!KubernetesFeaturesTestUtils.containerHasEnvVar(executor.container, env))
+    }
+  }
+
+  test("SSL RPC passwords shouldn't override an explicit secretKeyRef") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.rpc.enabled", "true")
+      .set("spark.ssl.rpc.keyStorePassword", "keyStorePass")
+      .set("spark.ssl.rpc.trustStorePassword", "trustStorePass")
+
+    val step = new BasicExecutorFeatureStep(
+      KubernetesTestConf.createExecutorConf(
+        sparkConf = conf,
+        secretEnvNamesToKeyRefs = Map(
+          SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD -> "rpc-secret:keystore-password")),
+      new SecurityManager(conf), defaultProfile)
+
+    val executor = step.configurePod(SparkPod.initialPod())
+    checkEnv(executor, conf, Map(
+      SSLOptions.ENV_RPC_SSL_TRUST_STORE_PASSWORD -> "trustStorePass"))
+  }
+
+  test("SSL RPC passwords shouldn't override an explicit spark.executorEnv entry") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.rpc.enabled", "true")
+      .set("spark.ssl.rpc.keyStorePassword", "keyStorePass")
+      .set("spark.ssl.rpc.trustStorePassword", "trustStorePass")
+      .set(s"spark.executorEnv.${SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD}", "userKeyStorePass")
+
+    val step = new BasicExecutorFeatureStep(KubernetesTestConf.createExecutorConf(sparkConf = conf),
+      new SecurityManager(conf), defaultProfile)
+
+    val executor = step.configurePod(SparkPod.initialPod())
+    checkEnv(executor, conf, Map(
+      SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD -> "userKeyStorePass",
+      SSLOptions.ENV_RPC_SSL_TRUST_STORE_PASSWORD -> "trustStorePass"))
+  }
+
+  test("SSL RPC passwords shouldn't override an env var predefined on the pod template") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.rpc.enabled", "true")
+      .set("spark.ssl.rpc.keyStorePassword", "keyStorePass")
+      .set("spark.ssl.rpc.trustStorePassword", "trustStorePass")
+
+    val step = new BasicExecutorFeatureStep(KubernetesTestConf.createExecutorConf(sparkConf = conf),
+      new SecurityManager(conf), defaultProfile)
+
+    val templatePod = SparkPod.initialPod()
+    val templateContainer = new ContainerBuilder(templatePod.container)
+      .addNewEnv()
+        .withName(SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD)
+        .withNewValueFrom()
+          .withNewSecretKeyRef()
+            .withKey("keystore-password")
+            .withName("rpc-secret")
+            .endSecretKeyRef()
+          .endValueFrom()
+        .endEnv()
+      .build()
+
+    val executor = step.configurePod(SparkPod(templatePod.pod, templateContainer))
+    checkEnv(executor, conf, Map(
+      SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD -> null,
+      SSLOptions.ENV_RPC_SSL_TRUST_STORE_PASSWORD -> "trustStorePass"))
+  }
+
+  test("SSL RPC passwords inherited from the global spark.ssl.* namespace propagate") {
+    val conf = baseConf.clone()
+      .set("spark.ssl.enabled", "true")
+      .set("spark.ssl.keyStorePassword", "globalKeyStorePass")
+      .set("spark.ssl.trustStorePassword", "globalTrustStorePass")
+      .set("spark.ssl.rpc.enabled", "true")
+
+    val step = new BasicExecutorFeatureStep(KubernetesTestConf.createExecutorConf(sparkConf = conf),
+      new SecurityManager(conf), defaultProfile)
+
+    val executor = step.configurePod(SparkPod.initialPod())
+    checkEnv(executor, conf, Map(
+      SSLOptions.ENV_RPC_SSL_KEY_STORE_PASSWORD -> "globalKeyStorePass",
+      SSLOptions.ENV_RPC_SSL_TRUST_STORE_PASSWORD -> "globalTrustStorePass"))
+  }
+
   test("SPARK-32661 test executor offheap memory") {
     baseConf.set(MEMORY_OFFHEAP_ENABLED, true)
     baseConf.set("spark.memory.offHeap.size", "42m")
@@ -776,6 +888,10 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
       s"$ENV_JAVA_OPT_PREFIX${ind + extraJavaOptsStart}" -> opt
     }.toMap
 
+    val duplicateEnvNames = executorPod.container.getEnv.asScala
+      .groupBy(_.getName).filter(_._2.size > 1).keys
+    assert(duplicateEnvNames.isEmpty, s"duplicate env names: ${duplicateEnvNames.mkString(", ")}")
+
     val containerEnvs = executorPod.container.getEnv.asScala.map {
       x => (x.getName, x.getValue)
     }.toMap