This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 749972239a05 [SPARK-59923][CORE] Warn when OIDC credential propagation
is enabled without RPC encryption
749972239a05 is described below
commit 749972239a05d1c021e703305805d990198aa1a4
Author: Kousuke Saruta <[email protected]>
AuthorDate: Fri Oct 2 08:20:51 2026 -0700
[SPARK-59923][CORE] Warn when OIDC credential propagation is enabled
without RPC encryption
### What changes were proposed in this pull request?
When OIDC credential propagation is enabled
(`spark.security.oidc.enabled=true`) but Spark's RPC channel encryption is not
configured, Spark now logs a warning at startup.
The check is added to `UserCredentialManager.create()`, which runs once on
the driver when the scheduler backend starts. It reuses the same "RPC
encryption is enabled" criteria as `HadoopDelegationTokenManager`'s
direct-credential-provider path:
- `spark.ssl.rpc.enabled=true`, or
- `spark.authenticate=true` together with one of
`spark.network.crypto.enabled` / `spark.authenticate.enableSaslEncryption`.
Unlike the `directCredentialProviders` path (which always fails fast via
`require()`), OIDC credential propagation warns by default. Operators can opt
into fail-fast behavior by setting
`spark.security.oidc.requireRpcEncryption=true`, which makes Spark refuse to
start unless RPC encryption is configured (the default is `false`, preserving
the SPIP's warn-by-default behavior). The predicate and the remediation message
are shared with `HadoopDelegationTokenManager` via `SecurityManager` [...]
`docs/security.md` is updated to document both the warning and the new
`spark.security.oidc.requireRpcEncryption` config.
### Why are the changes needed?
OIDC-derived service credentials are transmitted to executors over Spark's
RPC channel (via
`UpdateUserCredentials` broadcasts and `TaskDescription` delivery). If RPC
encryption is not configured, those credentials travel in cleartext.
The SPIP (design document) specifies this behavior:
> Enabling credential propagation without RPC encryption logs a warning by
default. A stricter
> mode (`spark.security.credentials.requireEncryption=true`) that fails
application startup unless
> RPC encryption is enabled may be introduced based on community feedback.
This PR implements both the warning and that stricter mode. The stricter
mode is exposed as `spark.security.oidc.requireRpcEncryption` (default `false`)
rather than the SPIP's tentative `spark.security.credentials.requireEncryption`
name, to stay within the existing `spark.security.oidc.*` namespace for this
feature (`spark.security.credentials.*` is the Kerberos /
direct-credential-provider namespace). The design and the documentation already
assumed the warning, but the implementati [...]
### Does this PR introduce _any_ user-facing change?
Yes. A warning is now logged at driver startup when OIDC credential
propagation is enabled without RPC encryption configured. By default startup is
not blocked. Additionally, a new config
`spark.security.oidc.requireRpcEncryption` (default `false`) is introduced:
when set to `true`, Spark fails startup (throws `IllegalArgumentException`) if
OIDC credential propagation is enabled without RPC encryption, instead of only
warning. This also fails startup in local mode, which is consistent [...]
### How was this patch tested?
New and updated unit tests:
- `SecurityManagerSuite`: `isRpcEncryptionEnabled` truth-table coverage for
the shared predicate (nothing configured / SSL / auth+crypto / auth+SASL / SSL
together with auth+crypto / SSL with crypto only / auth-only-without-encryption
/ crypto-only-without-auth / SASL-only-without-auth).
- `UserCredentialManagerSuite`: `checkRpcEncryption` emits / does not emit
the warning (default mode); fails fast with `IllegalArgumentException` when
`spark.security.oidc.requireRpcEncryption` is true and RPC encryption is
missing, and does not fail when encryption is configured; and `create()` emits
the warning (or fails fast under the enforce flag) when OIDC is enabled without
encryption but does nothing when OIDC is disabled (guards the wiring). A
further test pins the check order [...]
- `NonKerberosCredentialsSuite`: the direct-credential-provider `require()`
still rejects a config with no RPC encryption and accepts SASL / SSL RPC
encryption, now exercising the shared `SecurityManager.isRpcEncryptionEnabled`
predicate.
`SecurityManagerSuite`, `UserCredentialManagerSuite`, and
`NonKerberosCredentialsSuite` pass.
`core/scalastyle` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Kiro CLI / Claude
Closes #59185 from sarutak/oidc-propagation/rpc-encryption-warning.
Authored-by: Kousuke Saruta <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 5adc30a74c2da2c7194e5d965da029d418b38020)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scala/org/apache/spark/SecurityManager.scala | 51 +++++++++-
.../security/HadoopDelegationTokenManager.scala | 13 +--
.../deploy/security/UserCredentialManager.scala | 49 +++++++++-
.../org/apache/spark/internal/config/package.scala | 11 +++
.../org/apache/spark/SecurityManagerSuite.scala | 68 ++++++++++++++
.../security/UserCredentialManagerSuite.scala | 104 ++++++++++++++++++++-
docs/security.md | 20 +++-
7 files changed, 302 insertions(+), 14 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SecurityManager.scala
b/core/src/main/scala/org/apache/spark/SecurityManager.scala
index 1461677219bc..ddfeae900fca 100644
--- a/core/src/main/scala/org/apache/spark/SecurityManager.scala
+++ b/core/src/main/scala/org/apache/spark/SecurityManager.scala
@@ -86,8 +86,7 @@ private[spark] class SecurityManager(
private var secretKey: String = _
- private val sslRpcEnabled = sparkConf.getBoolean(
- "spark.ssl.rpc.enabled", false)
+ private val sslRpcEnabled = sparkConf.getBoolean(SSL_RPC_ENABLED_CONF, false)
logInfo(log"SecurityManager: authentication ${MDC(LogKeys.AUTH_ENABLED,
if (authOn) "enabled" else "disabled")}" +
@@ -275,6 +274,12 @@ private[spark] class SecurityManager(
/**
* Checks whether network encryption should be enabled.
+ *
+ * This reports only whether AES/SASL *network* encryption is in effect, and
treats SSL RPC as
+ * *disabling* it (AES/SASL and SSL RPC are mutually exclusive transports).
To ask instead "is
+ * the RPC channel encrypted at all?" (treating SSL RPC as encryption and
requiring
+ * `spark.authenticate` for the AES/SASL path), use
[[SecurityManager.isRpcEncryptionEnabled]].
+ *
* @return Whether to enable encryption when connecting to services that
support it.
*/
def isEncryptionEnabled(): Boolean = {
@@ -454,4 +459,46 @@ private[spark] object SecurityManager {
// key used to store the spark secret in the Hadoop UGI
val SECRET_LOOKUP_KEY = new Text("sparkCookie")
+
+ // `spark.ssl.rpc.enabled` has no Scala ConfigEntry (it is read directly in
TransportConf), so
+ // its key is defined here as a single source of truth for the code that
needs to read it.
+ private[spark] val SSL_RPC_ENABLED_CONF = "spark.ssl.rpc.enabled"
+
+ /**
+ * Returns true if Spark's RPC channel is encrypted by any supported
mechanism: SSL-based RPC
+ * encryption (`spark.ssl.rpc.enabled`), or authentication
(`spark.authenticate`) enabled
+ * together with either AES-based encryption
(`spark.network.crypto.enabled`) or the separate
+ * (deprecated) SASL-based encryption
(`spark.authenticate.enableSaslEncryption`).
+ *
+ * This is intentionally distinct from the instance method
+ * [[SecurityManager.isEncryptionEnabled]]:
+ * - `isEncryptionEnabled` reports whether AES/SASL *network* encryption is
in effect and
+ * treats SSL RPC as *disabling* it (returning false when
`spark.ssl.rpc.enabled` is set),
+ * because AES/SASL and SSL RPC are mutually exclusive transports. It
also does not require
+ * `spark.authenticate`, so it can return true for a configuration that
does not actually
+ * encrypt the channel.
+ * - This predicate instead answers "is the RPC channel encrypted at all?",
treating SSL RPC
+ * as encryption and requiring `spark.authenticate` for the AES/SASL
path. It must not be
+ * replaced by `isEncryptionEnabled`.
+ *
+ * It is shared by the two credential-propagation paths so they cannot drift
apart: the
+ * direct-credential-provider path in `HadoopDelegationTokenManager` fails
startup via
+ * `require()`, while OIDC credential propagation warns by default and fails
startup only when
+ * `spark.security.oidc.requireRpcEncryption` is true.
+ */
+ private[spark] def isRpcEncryptionEnabled(conf: SparkConf): Boolean = {
+ conf.getBoolean(SSL_RPC_ENABLED_CONF, false) ||
+ (conf.get(NETWORK_AUTH_ENABLED) &&
+ (conf.get(Network.NETWORK_CRYPTO_ENABLED) ||
conf.get(SASL_ENCRYPTION_ENABLED)))
+ }
+
+ /**
+ * Human-readable remediation describing how to enable RPC channel
encryption. Shared by the
+ * OIDC credential-propagation warning and the direct-credential-provider
`require()` message so
+ * the two stay in sync with [[isRpcEncryptionEnabled]].
+ */
+ private[spark] val rpcEncryptionRemediation: String =
+ s"Enable RPC encryption via $SSL_RPC_ENABLED_CONF=true, or
${NETWORK_AUTH_ENABLED.key}=true " +
+ s"together with one of ${Network.NETWORK_CRYPTO_ENABLED.key} / " +
+ s"${SASL_ENCRYPTION_ENABLED.key}."
}
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 b98a71b13cba..dec0f2b31a07 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
@@ -30,12 +30,11 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
-import org.apache.spark.SparkConf
+import org.apache.spark.{SecurityManager, SparkConf}
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
@@ -79,14 +78,10 @@ private[spark] class HadoopDelegationTokenManager(
"Both principal and keytab must be defined, or neither.")
if (sparkConf.get(DIRECT_CREDENTIAL_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,
+ require(SecurityManager.isRpcEncryptionEnabled(sparkConf),
"RPC channel encryption must be enabled when " +
- "spark.security.directCredentialProviders.enabled is true: either " +
- "spark.ssl.rpc.enabled=true, or spark.authenticate=true together with " +
- "spark.network.crypto.enabled or
spark.authenticate.enableSaslEncryption. " +
+ s"${DIRECT_CREDENTIAL_PROVIDERS_ENABLED.key} is true. " +
+ s"${SecurityManager.rpcEncryptionRemediation} " +
"Credential tokens must not be transmitted over unencrypted channels.")
}
diff --git
a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
index 9d6a72a48884..3f7ceea9c4bb 100644
---
a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
@@ -28,7 +28,7 @@ import scala.collection.mutable
import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
-import org.apache.spark.SparkConf
+import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
@@ -513,12 +513,59 @@ private[spark] object UserCredentialManager extends
Logging {
s"${SECURITY_OIDC_ENABLED.key} is true")
}
+ // Credentials are transmitted to executors over Spark's RPC channel (via
+ // UpdateUserCredentials broadcasts and TaskDescription delivery). If
RPC encryption is not
+ // configured, those credentials travel in cleartext. Per the SPIP,
enabling credential
+ // propagation without RPC encryption logs a warning by default;
operators can opt into a
+ // fail-fast mode via spark.security.oidc.requireRpcEncryption. This
check runs after the
+ // required-config checks above so a missing token file is not masked by
an encryption
+ // warning/error about an unrelated setting.
+ checkRpcEncryption(sparkConf)
+
val tokenIngestor = new FileTokenIngestor(Paths.get(tokenFile))
Some(new UserCredentialManager(
sparkConf, tokenIngestor, onCredentialsUpdate, selectionLoader))
}
}
+ /**
+ * Checks RPC channel encryption when OIDC credential propagation is
enabled, since credentials
+ * are otherwise transmitted over the RPC channel in cleartext.
+ *
+ * By default this only logs a warning, per the SPIP's warn-by-default
behavior. When
+ * `spark.security.oidc.requireRpcEncryption` is true, it instead fails fast
with an
+ * `IllegalArgumentException`, mirroring [[HadoopDelegationTokenManager]]'s
+ * direct-credential-provider path.
+ *
+ * The encryption predicate and the remediation hint are shared with
+ * [[HadoopDelegationTokenManager]]'s `require()` via [[SecurityManager]],
so the warning, the
+ * fail-fast message here, and the direct-provider `require()` message
cannot drift apart.
+ */
+ private[security] def checkRpcEncryption(sparkConf: SparkConf): Unit = {
+ if (!SecurityManager.isRpcEncryptionEnabled(sparkConf)) {
+ if (sparkConf.get(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION)) {
+ throw new IllegalArgumentException(
+ s"OIDC credential propagation (${SECURITY_OIDC_ENABLED.key}=true)
requires RPC channel " +
+ s"encryption when
${SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key}=true, but it is not " +
+ s"configured. ${SecurityManager.rpcEncryptionRemediation}")
+ }
+ // The remediation hint is carried under MDC(LogKeys.MESSAGE, ...)
rather than
+ // MDC(LogKeys.REASON, ...): REASON is used elsewhere for the reason an
operation failed, so
+ // keeping the hint out of it leaves the structured `reason` field clean
for real failure
+ // reasons. We say credentials "may be" (not "will be") transmitted in
cleartext because the
+ // channel could still be protected outside Spark (e.g. a service mesh
or a private network),
+ // which Spark cannot detect. We also point operators at the fail-fast
mode so those who want
+ // Spark to enforce encryption (rather than warn) know how to opt in.
+ logWarning(log"OIDC credential propagation is enabled " +
+ log"(${MDC(LogKeys.CONFIG, SECURITY_OIDC_ENABLED.key)}=true) but RPC
channel encryption " +
+ log"is not configured. Credentials may be transmitted to executors
over an unencrypted " +
+ log"channel. " + log"${MDC(LogKeys.MESSAGE,
SecurityManager.rpcEncryptionRemediation)}" +
+ log" This is a warning by default; set " +
+ log"${MDC(LogKeys.CONFIG2,
SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key)}=true to fail fast " +
+ log"instead.")
+ }
+ }
+
/**
* Selection phase of OIDC credential propagation, run early during
`SparkContext`
* initialization (before the driver's Hadoop `Configuration` is
materialized).
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 0d57cfaf9452..90f3e08f367e 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
@@ -1746,6 +1746,17 @@ package object config {
.checkValue(_ > 0, "The minimum renewal interval must be a positive time
value.")
.createWithDefaultString("30s")
+ private[spark] val SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION =
+ ConfigBuilder("spark.security.oidc.requireRpcEncryption")
+ .doc("When OIDC credential propagation is enabled, whether to require
RPC channel " +
+ "encryption. When true, Spark refuses to start if RPC encryption is
not configured, " +
+ "instead of only logging a warning. Defaults to false to preserve the
warn-by-default " +
+ "behavior described in the SPIP.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
private[spark] val DIRECT_CREDENTIAL_PROVIDERS_ENABLED =
ConfigBuilder("spark.security.directCredentialProviders.enabled")
.doc(
diff --git a/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
b/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
index f4ecc540155d..b9ec0726074a 100644
--- a/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
@@ -513,5 +513,73 @@ class SecurityManagerSuite extends SparkFunSuite with
ResetSystemProperties {
private def encodeFileAsBase64(secretFile: File) = {
Base64.getEncoder.encodeToString(Files.readAllBytes(secretFile.toPath))
}
+
+ // ========== SecurityManager.isRpcEncryptionEnabled (shared RPC-encryption
predicate) ==========
+ //
+ // This predicate is distinct from the instance method
isEncryptionEnabled(): it treats SSL RPC
+ // as encryption (rather than as disabling AES/SASL) and requires
spark.authenticate for the
+ // AES/SASL path. The cases below pin that contract, including the
combinations that the OIDC
+ // credential-propagation warning and the direct-credential-provider
require() both depend on.
+
+ test("isRpcEncryptionEnabled: false when nothing is configured") {
+ assert(!SecurityManager.isRpcEncryptionEnabled(new SparkConf(loadDefaults
= false)))
+ }
+
+ test("isRpcEncryptionEnabled: true with SSL RPC encryption") {
+ val conf = new SparkConf(loadDefaults =
false).set(SecurityManager.SSL_RPC_ENABLED_CONF, "true")
+ assert(SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: true with authentication + network crypto") {
+ val conf = new SparkConf(loadDefaults = false)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(Network.NETWORK_CRYPTO_ENABLED, true)
+ assert(SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: true with authentication + SASL encryption") {
+ val conf = new SparkConf(loadDefaults = false)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(SASL_ENCRYPTION_ENABLED, true)
+ assert(SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: true with SSL RPC encryption together with
authentication + " +
+ "network crypto") {
+ // SSL RPC is a sufficient condition on its own; combining it with the AES
path must stay true.
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SecurityManager.SSL_RPC_ENABLED_CONF, "true")
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(Network.NETWORK_CRYPTO_ENABLED, true)
+ assert(SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: true with SSL RPC encryption and network
crypto but no auth") {
+ // SSL RPC alone encrypts the channel; the missing spark.authenticate only
gates the AES/SASL
+ // path, so the disjunction is still true via SSL.
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SecurityManager.SSL_RPC_ENABLED_CONF, "true")
+ .set(Network.NETWORK_CRYPTO_ENABLED, true)
+ assert(SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: false when authentication is enabled without
encryption") {
+ // Authentication alone (without crypto/SASL encryption) does not encrypt
the channel.
+ val conf = new SparkConf(loadDefaults = false).set(NETWORK_AUTH_ENABLED,
true)
+ assert(!SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: false when network crypto is enabled without
authentication") {
+ // spark.network.crypto.enabled only takes effect with
spark.authenticate=true; on its own it
+ // does not encrypt anything, so it must not be treated as RPC encryption.
(This is also where
+ // isEncryptionEnabled() would differ: it returns true here even though
nothing is encrypted.)
+ val conf = new SparkConf(loadDefaults =
false).set(Network.NETWORK_CRYPTO_ENABLED, true)
+ assert(!SecurityManager.isRpcEncryptionEnabled(conf))
+ }
+
+ test("isRpcEncryptionEnabled: false when SASL encryption is enabled without
authentication") {
+ val conf = new SparkConf(loadDefaults =
false).set(SASL_ENCRYPTION_ENABLED, true)
+ assert(!SecurityManager.isRpcEncryptionEnabled(conf))
+ }
}
diff --git
a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
index e8ff807e7173..a1e49481ed33 100644
---
a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
+++
b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
@@ -28,7 +28,7 @@ import scala.jdk.CollectionConverters._
import org.scalatest.concurrent.Eventually.{eventually, timeout}
-import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
import org.apache.spark.internal.config._
import org.apache.spark.security._
@@ -858,4 +858,106 @@ class UserCredentialManagerSuite extends SparkFunSuite {
AnotherFakeCredentialProvider.throwOnProperties = false
}
}
+
+ // ========== RPC encryption warning (OIDC wiring) ==========
+ //
+ // The RPC-encryption predicate itself
(SecurityManager.isRpcEncryptionEnabled) is unit-tested in
+ // SecurityManagerSuite. Here we only verify the OIDC-specific wiring: that
the warning fires (and
+ // does not fire) for the right configurations. Assertions match on a stable
fragment plus the
+ // WARN level rather than the full message text, so rewording the message
does not break them.
+
+ // A fragment that stays stable across rewording: the OIDC config key is
always named verbatim.
+ private val oidcWarningKeyFragment = SECURITY_OIDC_ENABLED.key
+
+ private def hasOidcEncryptionWarning(appender: LogAppender): Boolean =
+ appender.loggingEvents.exists { e =>
+ e.getLevel == org.apache.logging.log4j.Level.WARN &&
+ e.getMessage.getFormattedMessage.contains(oidcWarningKeyFragment) &&
+ e.getMessage.getFormattedMessage.contains("encryption")
+ }
+
+ test("checkRpcEncryption: warns when encryption is not configured") {
+ val appender = new LogAppender(oidcWarningKeyFragment)
+ withLogAppender(appender) {
+ UserCredentialManager.checkRpcEncryption(new SparkConf(loadDefaults =
false))
+ }
+ assert(hasOidcEncryptionWarning(appender))
+ }
+
+ test("checkRpcEncryption: does not warn when encryption is configured") {
+ val appender = new LogAppender(oidcWarningKeyFragment)
+ withLogAppender(appender) {
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SecurityManager.SSL_RPC_ENABLED_CONF, "true")
+ UserCredentialManager.checkRpcEncryption(conf)
+ }
+ assert(!hasOidcEncryptionWarning(appender))
+ }
+
+ test("checkRpcEncryption: fails fast when requireRpcEncryption is true
without encryption") {
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION, true)
+ val e = intercept[IllegalArgumentException] {
+ UserCredentialManager.checkRpcEncryption(conf)
+ }
+ assert(e.getMessage.contains(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key))
+ assert(e.getMessage.contains("RPC channel"))
+ }
+
+ test("checkRpcEncryption: does not fail when requireRpcEncryption is true
with encryption") {
+ val appender = new LogAppender(oidcWarningKeyFragment)
+ withLogAppender(appender) {
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION, true)
+ .set(SecurityManager.SSL_RPC_ENABLED_CONF, "true")
+ UserCredentialManager.checkRpcEncryption(conf) // must not throw
+ }
+ // When encryption is configured, neither the warning nor the exception
fires.
+ assert(!hasOidcEncryptionWarning(appender))
+ }
+
+ test("create warns when OIDC is enabled without RPC encryption") {
+ // Guards the wiring: create() must invoke the encryption warning for an
enabled config.
+ val appender = new LogAppender(oidcWarningKeyFragment)
+ withLogAppender(appender) {
+ val conf = createSparkConf() // OIDC enabled, no RPC encryption
configured
+ UserCredentialManager.create(
+ conf, (_, _) => (), Some(new CredentialProviderLoader()))
+ }
+ assert(hasOidcEncryptionWarning(appender))
+ }
+
+ test("create fails fast when OIDC requires RPC encryption but it is not
configured") {
+ // Guards the wiring: create() must enforce when requireRpcEncryption is
set.
+ val conf = createSparkConf().set(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION,
true)
+ val e = intercept[IllegalArgumentException] {
+ UserCredentialManager.create(conf, (_, _) => (), Some(new
CredentialProviderLoader()))
+ }
+ assert(e.getMessage.contains(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key))
+ }
+
+ test("create: a missing token file is reported before the RPC-encryption
enforce error") {
+ // Pins the check order: with requireRpcEncryption=true and no RPC
encryption, both the
+ // token-file check and checkRpcEncryption would throw. The
required-config checks must run
+ // first, so the token-file error (not the encryption enforce error)
surfaces. Moving
+ // checkRpcEncryption before the token-file check would fail this test.
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION, true)
+ // Deliberately not setting SECURITY_OIDC_IDENTITY_TOKEN_FILE and no RPC
encryption.
+ val e = intercept[IllegalArgumentException] {
+ UserCredentialManager.create(conf, (_, _) => (), Some(new
CredentialProviderLoader()))
+ }
+ assert(e.getMessage.contains(SECURITY_OIDC_IDENTITY_TOKEN_FILE.key))
+ assert(!e.getMessage.contains(SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key))
+ }
+
+ test("create does not warn when OIDC is disabled") {
+ val appender = new LogAppender(oidcWarningKeyFragment)
+ withLogAppender(appender) {
+ val conf = new SparkConf(loadDefaults =
false).set(SECURITY_OIDC_ENABLED, false)
+ UserCredentialManager.create(conf, (_, _) => (), Some(new
CredentialProviderLoader()))
+ }
+ assert(!hasOidcEncryptionWarning(appender))
+ }
}
diff --git a/docs/security.md b/docs/security.md
index 2affb692603c..b921598b6d6b 100644
--- a/docs/security.md
+++ b/docs/security.md
@@ -1222,7 +1222,12 @@ delegation tokens rather than the original
ticket-granting ticket.
* Service credentials are not written to shuffle files, event logs, or
checkpoints.
* Credentials are carried over Spark's RPC channels and therefore rely on
Spark's existing RPC
encryption configuration. Operators are strongly encouraged to enable RPC
encryption whenever
- credential propagation is enabled. See [Network
Encryption](#network-encryption) for details.
+ credential propagation is enabled. If credential propagation is enabled
while RPC encryption is
+ not configured, Spark logs a warning at startup, since credentials would
otherwise be transmitted
+ to executors over an unencrypted channel. To turn this into a hard
requirement, set
+ `spark.security.oidc.requireRpcEncryption=true`, which makes Spark refuse to
start unless RPC
+ encryption is configured. See [Network Encryption](#network-encryption) for
+ details.
## Custom CredentialProvider
@@ -1313,6 +1318,19 @@ each provider.
</td>
<td>4.4.0</td>
</tr>
+<tr>
+ <td><code>spark.security.oidc.requireRpcEncryption</code></td>
+ <td><code>false</code></td>
+ <td>
+ When OIDC credential propagation is enabled, whether to require RPC
channel encryption. By
+ default (<code>false</code>), Spark only logs a warning if credential
propagation is enabled
+ without RPC encryption, since credentials would then travel over the RPC
channel in cleartext.
+ When set to <code>true</code>, Spark instead refuses to start unless RPC
encryption is
+ configured. Leave it <code>false</code> if the channel is protected
outside Spark (for
+ example, a service mesh or a private network).
+ </td>
+ <td>4.4.0</td>
+</tr>
<tr>
<td><code>spark.security.oidc.identityToken.file</code></td>
<td>(none)</td>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
