diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala index 58bbd7477bd07..fb7a62e48389c 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala @@ -34,6 +34,7 @@ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED import org.apache.spark.rpc.RpcEndpointRef import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens import org.apache.spark.security.HadoopDelegationTokenProvider @@ -76,28 +77,52 @@ private[spark] class HadoopDelegationTokenManager( require((principal == null) == (keytab == null), "Both principal and keytab must be defined, or neither.") + if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + val hasEncryption = sparkConf.getBoolean("spark.ssl.rpc.enabled", false) || + (sparkConf.get(NETWORK_AUTH_ENABLED) && + (sparkConf.get(NETWORK_CRYPTO_ENABLED) || sparkConf.get(SASL_ENCRYPTION_ENABLED))) + require(hasEncryption, + "RPC channel encryption must be enabled when " + + "spark.security.credentials.directProviders.enabled is true: either " + + "spark.ssl.rpc.enabled=true, or spark.authenticate=true together with " + + "spark.network.crypto.enabled or spark.authenticate.enableSaslEncryption. " + + "Credential tokens must not be transmitted over unencrypted channels.") + } + private val delegationTokenProviders = loadProviders() + if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.isEmpty) { + logWarning("spark.security.credentials.directProviders.enabled is true but " + + "no HadoopDelegationTokenProvider implementations were discovered. " + + "Ensure provider JARs are on the classpath with META-INF/services registration.") + } logDebug("Using the following builtin delegation token providers: " + s"${delegationTokenProviders.keys.mkString(", ")}.") private var renewalExecutor: ScheduledExecutorService = _ + private def hasKerberosCredentials: Boolean = + sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { + case "keytab" => principal != null + case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials() + case _ => false + } + /** @return Whether delegation token renewal is enabled. */ - def renewalEnabled: Boolean = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { - case "keytab" => principal != null - case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials() - case _ => false + def renewalEnabled: Boolean = { + hasKerberosCredentials || + (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty) } /** - * Start the token renewer. Requires a principal and keytab. Upon start, the renewer will - * obtain delegation tokens for all configured services and send them to the driver, and - * set up tasks to periodically get fresh tokens as needed. + * Start the token renewer. Upon start, the renewer will obtain delegation tokens for all + * configured services and send them to the driver, and set up tasks to periodically get + * fresh tokens as needed. * - * This method requires that a keytab has been provided to Spark, and will try to keep the - * logged in user's TGT valid while this manager is active. + * When Kerberos credentials are available, the manager keeps the TGT renewed and calls + * providers inside a privileged context. When only direct providers are configured, providers + * are called without Kerberos authentication. * - * @return New set of delegation tokens created for the configured principal. + * @return New set of delegation tokens created for the configured services. */ def start(): Array[Byte] = { require(renewalEnabled, "Token renewal must be enabled to start the renewer.") @@ -141,39 +166,58 @@ private[spark] class HadoopDelegationTokenManager( val hasKerberosCreds = principal != null || Option(currentUser.getRealUser()).getOrElse(currentUser).hasKerberosCredentials() - // Delegation tokens can only be obtained if the real user has Kerberos credentials, so - // skip creation when those are not available. if (hasKerberosCreds) { val freshUGI = doLogin() freshUGI.doAs(new PrivilegedExceptionAction[Unit]() { override def run(): Unit = { - val (newTokens, _) = obtainDelegationTokens() + val (newTokens, _, _) = obtainDelegationTokens() creds.addAll(newTokens) } }) if (!currentUser.equals(freshUGI)) { FileSystem.closeAllForUGI(freshUGI) } + } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + val (newTokens, _, _) = obtainDelegationTokens(isolateFailures = true) + creds.addAll(newTokens) } } /** * Fetch new delegation tokens for configured services. * - * @return 2-tuple (credentials with new tokens, time by which the tokens must be renewed) + * @param isolateFailures When true, catch per-provider exceptions and continue with remaining + * providers. When false, exceptions propagate to the caller. + * @return 3-tuple (credentials with new tokens, time by which the tokens must be renewed, + * number of providers that failed). The failure count is always 0 when + * isolateFailures is false (failures propagate instead). */ - private def obtainDelegationTokens(): (Credentials, Long) = { + private def obtainDelegationTokens( + isolateFailures: Boolean = false): (Credentials, Long, Int) = { val creds = new Credentials() + var failureCount = 0 val nextRenewal = delegationTokenProviders.values.flatMap { provider => if (provider.delegationTokensRequired(sparkConf, hadoopConf)) { - provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) + if (isolateFailures) { + try { + provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) + } catch { + case e: Exception => + logWarning(log"Failed to obtain credentials from " + + log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e) + failureCount += 1 + None + } + } else { + provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) + } } else { logDebug(s"Service ${provider.serviceName} does not require a token." + s" Check your configuration to see if security is disabled or not.") None } }.foldLeft(Long.MaxValue)(math.min) - (creds, nextRenewal) + (creds, nextRenewal, failureCount) } // Visible for testing. @@ -199,8 +243,7 @@ private[spark] class HadoopDelegationTokenManager( */ private def updateTokensTask(): Array[Byte] = { try { - val freshUGI = doLogin() - val creds = obtainTokensAndScheduleRenewal(freshUGI) + val creds = obtainTokensAndScheduleRenewal() val tokens = SparkHadoopUtil.get.serialize(creds) logInfo("Updating delegation tokens.") @@ -226,24 +269,44 @@ private[spark] class HadoopDelegationTokenManager( * * @return Credentials containing the new tokens. */ - private def obtainTokensAndScheduleRenewal(ugi: UserGroupInformation): Credentials = { - ugi.doAs(new PrivilegedExceptionAction[Credentials]() { - override def run(): Credentials = { - val (creds, nextRenewal) = obtainDelegationTokens() - - // Calculate the time when new credentials should be created, based on the configured - // ratio. - val now = System.currentTimeMillis - val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) - val delay = (ratio * (nextRenewal - now)).toLong - logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + - log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + - log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + - log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}") - scheduleRenewal(delay) - creds + private def obtainTokensAndScheduleRenewal(): Credentials = { + val (creds, nextRenewal, failureCount) = if (hasKerberosCredentials) { + val freshUGI = doLogin() + val currentUser = UserGroupInformation.getCurrentUser() + val result = freshUGI.doAs( + new PrivilegedExceptionAction[(Credentials, Long, Int)]() { + override def run(): (Credentials, Long, Int) = { + obtainDelegationTokens() + } + }) + if (!currentUser.equals(freshUGI)) { + FileSystem.closeAllForUGI(freshUGI) } - }) + result + } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + obtainDelegationTokens(isolateFailures = true) + } else { + // Reachable when ccache-based Kerberos credentials expired after start(). + // Throw so that updateTokensTask() reschedules a retry instead of silently + // distributing empty credentials and never renewing again. + throw new IllegalStateException("Cannot obtain delegation tokens: no Kerberos " + + "credentials are available and direct credential providers are not enabled.") + } + + // Scheduling does not require a privileged context (only token acquisition does). + val now = System.currentTimeMillis + val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) + val delay = if (failureCount > 0 && nextRenewal == Long.MaxValue) { + TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT)) + } else { + (ratio * (nextRenewal - now)).toLong + } + logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + + log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + + log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + + log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}") + scheduleRenewal(delay) + creds } private def doLogin(): UserGroupInformation = { diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 02e0fc9f00895..e01599e7d63f1 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1730,6 +1730,18 @@ package object config { .checkValue(_ > 0, "The minimum renewal interval must be a positive time value.") .createWithDefaultString("30s") + private[spark] val CREDENTIALS_DIRECT_PROVIDERS_ENABLED = + ConfigBuilder("spark.security.credentials.directProviders.enabled") + .doc( + "When true, enables delegation token collection and renewal without Kerberos. " + + "Providers are called directly (without doLogin/doAs) and participate in the " + + "same renewal and distribution lifecycle as Kerberos delegation token providers. " + + "Providers that require Kerberos self-gate via delegationTokensRequired.") + .version("4.3.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) + .booleanConf + .createWithDefault(false) + private[spark] val SHUFFLE_SORT_INIT_BUFFER_SIZE = ConfigBuilder("spark.shuffle.sort.initialBufferSize") .internal() diff --git a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala index cbc32f900abfb..3fdbe4c25017a 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala @@ -33,7 +33,7 @@ private[spark] trait SupportsDelegationToken { /** * Create the delegation token manager to be used for the application. This method is called * once during the start of the scheduler backend (so after the object has already been - * fully constructed), only if security is enabled in the Hadoop configuration. + * fully constructed), when security is enabled or direct credential providers are configured. */ protected def createTokenManager(): Option[HadoopDelegationTokenManager] = None @@ -42,8 +42,14 @@ private[spark] trait SupportsDelegationToken { */ protected def updateDelegationTokens(tokens: Array[Byte]): Unit + /** + * Whether the token manager should be started. Returns true if Hadoop security is enabled + * or if direct credential providers are configured. + */ + protected def tokenManagerRequired(): Boolean = UserGroupInformation.isSecurityEnabled + protected def setupTokenManager(): Unit = { - if (UserGroupInformation.isSecurityEnabled) { + if (tokenManagerRequired()) { delegationTokenManager = createTokenManager() delegationTokenManager.foreach { dtm => val ugi = UserGroupInformation.getCurrentUser diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index e4bdcd9fa7a37..1665e5a7a6a47 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -618,6 +618,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp override def start(): Unit = { setupTokenManager() + if (conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) { + logWarning("spark.security.credentials.directProviders.enabled is set but " + + "this cluster manager does not support credential distribution. " + + "No tokens will be collected or renewed.") + } + } + + override protected def tokenManagerRequired(): Boolean = { + super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) } protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint() diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index c3a1a60a0384f..88f628f367288 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -129,6 +129,10 @@ private[spark] class LocalSchedulerBackend( Some(new HadoopDelegationTokenManager(conf, scheduler.sc.hadoopConfiguration, localEndpoint)) } + override protected def tokenManagerRequired(): Boolean = { + super.tokenManagerRequired() || conf.get(config.CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + } + override protected def updateDelegationTokens(tokens: Array[Byte]): Unit = { SparkHadoopUtil.get.addDelegationTokens(tokens, conf) } diff --git a/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider b/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider index ed3908e95e4cb..830729beaa19a 100644 --- a/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider +++ b/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider @@ -16,3 +16,6 @@ # org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider +org.apache.spark.deploy.security.TestNonKerberosTokenProvider +org.apache.spark.deploy.security.TestDisabledProvider +org.apache.spark.deploy.security.TestFailingProvider diff --git a/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala new file mode 100644 index 0000000000000..64914340d4eee --- /dev/null +++ b/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala @@ -0,0 +1,186 @@ +/* + * 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. + */ + +package org.apache.spark.deploy.security + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.io.Text +import org.apache.hadoop.security.Credentials +import org.mockito.ArgumentCaptor +import org.mockito.Mockito.{mock, verify} + +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.SparkHadoopUtil +import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED +import org.apache.spark.rpc.RpcEndpointRef +import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens +import org.apache.spark.security.HadoopDelegationTokenProvider + +private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-direct" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = + sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + creds.addSecretKey(new Text("test.direct.credential"), "test-token".getBytes) + Some(System.currentTimeMillis() + 3600000L) + } +} + +private class TestDisabledProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-disabled" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = false + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + // scalastyle:off throwerror + throw new AssertionError("Should not be called when delegationTokensRequired is false") + // scalastyle:on throwerror + } +} + +private class TestFailingProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-failing" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = + sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + throw new RuntimeException("Simulated provider failure") + } +} + +class NonKerberosCredentialsSuite extends SparkFunSuite { + private val hadoopConf = new Configuration() + + private def baseConf: SparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(NETWORK_AUTH_ENABLED, true) + .set(NETWORK_CRYPTO_ENABLED, true) + + test("renewalEnabled returns true when config is enabled") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("renewalEnabled returns false when config is disabled and no Kerberos") { + val sparkConf = new SparkConf(false) + .set(NETWORK_CRYPTO_ENABLED, true) + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(!manager.renewalEnabled) + } + + test("providers are called without Kerberos when config is enabled") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } + + test("providers with delegationTokensRequired=false are not called") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + } + + test("provider failure does not prevent other providers from running") { + val sparkConf = baseConf + .set("spark.security.credentials.test-failing.enabled", "true") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } + + test("individual provider can be disabled via per-service config") { + val sparkConf = baseConf + .set("spark.security.credentials.test-direct.enabled", "false") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + + assert(!manager.isProviderLoaded("test-direct")) + } + + test("fails if no RPC encryption is enabled") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + + val e = intercept[IllegalArgumentException] { + new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + } + assert(e.getMessage.contains("RPC channel encryption")) + } + + test("accepts SASL encryption as sufficient") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(NETWORK_AUTH_ENABLED, true) + .set(SASL_ENCRYPTION_ENABLED, true) + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("accepts SSL RPC encryption as sufficient") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set("spark.ssl.rpc.enabled", "true") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("start() obtains tokens and sends UpdateDelegationTokens to schedulerRef") { + val mockRef = mock(classOf[RpcEndpointRef]) + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, mockRef) + + try { + val tokens = manager.start() + assert(tokens != null) + + val captor = ArgumentCaptor.forClass(classOf[Any]) + verify(mockRef).send(captor.capture()) + val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens] + + val creds = SparkHadoopUtil.get.deserialize(msg.tokens) + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } finally { + manager.stop() + } + } +} diff --git a/docs/security.md b/docs/security.md index 7c0d1b08875e1..3915f09f4e8d7 100644 --- a/docs/security.md +++ b/docs/security.md @@ -981,6 +981,23 @@ The following options provides finer-grained control for this feature: 2.3.0 + + spark.security.credentials.directProviders.enabled + false + + When true, enables credential collection and renewal without Kerberos. Providers + registered via HadoopDelegationTokenProvider are called directly (without + doLogin/doAs) and participate in the same renewal and + distribution lifecycle as Kerberos delegation token providers. Providers that require + Kerberos self-gate via their delegationTokensRequired method. + Requires RPC channel encryption: either spark.ssl.rpc.enabled=true, + or spark.authenticate=true together with + spark.network.crypto.enabled or + spark.authenticate.enableSaslEncryption. + Supported on YARN, Kubernetes, and local mode. Not supported on standalone clusters. + + 4.3.0 + spark.kerberos.access.hadoopFileSystems (none)