dongjoon-hyun commented on code in PR #59185:
URL: https://github.com/apache/spark/pull/59185#discussion_r4158581876
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -519,6 +527,44 @@ private[spark] object UserCredentialManager extends
Logging {
}
}
+ /**
+ * Returns true if Spark's RPC channel encryption is configured: either
SSL-based RPC encryption
+ * (`spark.ssl.rpc.enabled`), or authentication (`spark.authenticate`)
together with either
+ * AES-based encryption (`spark.network.crypto.enabled`) or the separate
(deprecated) SASL-based
+ * encryption (`spark.authenticate.enableSaslEncryption`).
+ *
+ * This is shared with [[HadoopDelegationTokenManager]]'s
direct-credential-provider check so the
+ * two credential-propagation paths cannot drift apart: that path fails
startup via `require()`,
+ * while OIDC propagation only warns (see [[warnIfRpcEncryptionDisabled]]).
+ */
+ private[security] def isRpcEncryptionEnabled(sparkConf: SparkConf): Boolean
= {
+ sparkConf.getBoolean("spark.ssl.rpc.enabled", false) ||
+ (sparkConf.get(NETWORK_AUTH_ENABLED) &&
+ (sparkConf.get(NETWORK_CRYPTO_ENABLED) ||
sparkConf.get(SASL_ENCRYPTION_ENABLED)))
+ }
+
+ /**
+ * Logs a warning if OIDC credential propagation is enabled but RPC channel
encryption is not
+ * configured, since credentials would then be transmitted over the RPC
channel in cleartext.
+ * Unlike [[HadoopDelegationTokenManager]]'s direct-credential-provider path
(which fails fast),
+ * this only warns, per the SPIP's default behavior.
+ */
+ private[security] def warnIfRpcEncryptionDisabled(sparkConf: SparkConf):
Unit = {
+ if (!isRpcEncryptionEnabled(sparkConf)) {
+ // Reference config keys via `.key` (rather than hardcoded strings) so
the message stays in
+ // sync if a key is renamed. `spark.ssl.rpc.enabled` has no Scala
ConfigEntry (it is read
+ // directly in TransportConf), so it is spelled out literally here and in
+ // isRpcEncryptionEnabled.
+ val remediation = s"Enable RPC encryption via
spark.ssl.rpc.enabled=true, or " +
Review Comment:
Two small things on the message construction:
- `s"Enable RPC encryption via spark.ssl.rpc.enabled=true, or "` has no
interpolated value, so the `s` prefix is not needed.
- The comment above says keys are referenced via `.key` to stay in sync, but
`spark.ssl.rpc.enabled` is spelled out here and again in
`isRpcEncryptionEnabled`. A single private constant used by both would keep
them in sync and make the explanatory comment unnecessary.
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -519,6 +527,44 @@ private[spark] object UserCredentialManager extends
Logging {
}
}
+ /**
+ * Returns true if Spark's RPC channel encryption is configured: either
SSL-based RPC encryption
+ * (`spark.ssl.rpc.enabled`), or authentication (`spark.authenticate`)
together with either
+ * AES-based encryption (`spark.network.crypto.enabled`) or the separate
(deprecated) SASL-based
+ * encryption (`spark.authenticate.enableSaslEncryption`).
+ *
+ * This is shared with [[HadoopDelegationTokenManager]]'s
direct-credential-provider check so the
+ * two credential-propagation paths cannot drift apart: that path fails
startup via `require()`,
+ * while OIDC propagation only warns (see [[warnIfRpcEncryptionDisabled]]).
+ */
+ private[security] def isRpcEncryptionEnabled(sparkConf: SparkConf): Boolean
= {
Review Comment:
This is now a third copy of the "is RPC encrypted" decision.
`SecurityManager.isEncryptionEnabled()` and `TransportConf` already read
`spark.ssl.rpc.enabled` on their own, and this PR also makes
`HadoopDelegationTokenManager` (the Kerberos / direct-provider path) depend on
the OIDC-specific `UserCredentialManager` object just to reach the predicate.
If the rule changes later (for example a new transport-encryption option),
these copies can drift apart, and a missed warning here means credentials go
out in cleartext with no signal. Could the predicate live somewhere neutral
(e.g. `SecurityManager` or a util object), or reuse the existing check, so both
credential paths depend on that rather than on each other?
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -519,6 +527,44 @@ private[spark] object UserCredentialManager extends
Logging {
}
}
+ /**
+ * Returns true if Spark's RPC channel encryption is configured: either
SSL-based RPC encryption
+ * (`spark.ssl.rpc.enabled`), or authentication (`spark.authenticate`)
together with either
+ * AES-based encryption (`spark.network.crypto.enabled`) or the separate
(deprecated) SASL-based
+ * encryption (`spark.authenticate.enableSaslEncryption`).
+ *
+ * This is shared with [[HadoopDelegationTokenManager]]'s
direct-credential-provider check so the
+ * two credential-propagation paths cannot drift apart: that path fails
startup via `require()`,
+ * while OIDC propagation only warns (see [[warnIfRpcEncryptionDisabled]]).
+ */
+ private[security] def isRpcEncryptionEnabled(sparkConf: SparkConf): Boolean
= {
+ sparkConf.getBoolean("spark.ssl.rpc.enabled", false) ||
+ (sparkConf.get(NETWORK_AUTH_ENABLED) &&
+ (sparkConf.get(NETWORK_CRYPTO_ENABLED) ||
sparkConf.get(SASL_ENCRYPTION_ENABLED)))
+ }
+
+ /**
+ * Logs a warning if OIDC credential propagation is enabled but RPC channel
encryption is not
+ * configured, since credentials would then be transmitted over the RPC
channel in cleartext.
+ * Unlike [[HadoopDelegationTokenManager]]'s direct-credential-provider path
(which fails fast),
+ * this only warns, per the SPIP's default behavior.
+ */
+ private[security] def warnIfRpcEncryptionDisabled(sparkConf: SparkConf):
Unit = {
+ if (!isRpcEncryptionEnabled(sparkConf)) {
+ // Reference config keys via `.key` (rather than hardcoded strings) so
the message stays in
+ // sync if a key is renamed. `spark.ssl.rpc.enabled` has no Scala
ConfigEntry (it is read
+ // directly in TransportConf), so it is spelled out literally here and in
+ // isRpcEncryptionEnabled.
+ val remediation = s"Enable RPC encryption via
spark.ssl.rpc.enabled=true, or " +
+ s"${NETWORK_AUTH_ENABLED.key}=true together with one of " +
+ s"${NETWORK_CRYPTO_ENABLED.key} / ${SASL_ENCRYPTION_ENABLED.key}."
+ 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 will be transmitted to executors
over an unencrypted " +
+ log"channel. ${MDC(LogKeys.REASON, remediation)}")
Review Comment:
`LogKeys.REASON` is used elsewhere for the reason an operation failed. Here
it carries a multi-sentence remediation hint, so with structured logging the
`reason` field for this warning contains config-key advice, which will be mixed
in with real failure reasons when filtering or aggregating. Plain `log"..."`
text for the hint (or separate MDC values for the keys) would avoid that.
Also, the message says credentials "will be transmitted ... over an
unencrypted channel", but Spark cannot know that when transport is protected
outside Spark (service mesh mTLS, private network), and there is currently no
way to silence the warning in that case. Wording such as "may be transmitted"
would be more accurate, and it could point at the stricter mode mentioned in
the SPIP.
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -79,10 +78,7 @@ 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(UserCredentialManager.isRpcEncryptionEnabled(sparkConf),
Review Comment:
The predicate moved, but the error message right below still lists the key
combinations in its own wording, and the OIDC warning builds its text
separately again. If `isRpcEncryptionEnabled` is extended later, the
`require()` message and the warning text no longer match what the check
accepts. Building both messages from one place next to the predicate would
avoid that.
##########
core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala:
##########
@@ -858,4 +859,90 @@ class UserCredentialManagerSuite extends SparkFunSuite {
AnotherFakeCredentialProvider.throwOnProperties = false
}
}
+
+ // ========== RPC encryption warning ==========
+
+ private val encryptionWarningFragment = "RPC channel encryption is not
configured"
Review Comment:
The log-based tests match on this exact fragment, which spans two
concatenated `log""` pieces in the main code, so rewording the message breaks
all three tests even though behavior is unchanged. Consider asserting on a
stable part (for example the OIDC config key, or the log level plus a shorter
fragment).
The `isRpcEncryptionEnabled` truth table also omits a few combinations:
`spark.ssl.rpc.enabled` together with auth + crypto, and
`spark.ssl.rpc.enabled` combined with only crypto. And since
`HadoopDelegationTokenManager`'s `require()` now goes through the shared
predicate, a small test that it still rejects/accepts the same configs would
guard the refactor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
