sarutak commented on code in PR #59185:
URL: https://github.com/apache/spark/pull/59185#discussion_r4163326709
##########
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:
Thanks, I've addressed them:
- The remediation no longer goes into `MDC(LogKeys.REASON, ...)`. `REASON`
is used elsewhere for the reason an operation failed, so I moved the hint to
`LogKeys.MESSAGE` to keep the structured `reason` field clean for real failure
reasons, with a comment explaining why.
- Changed "will be transmitted ... over an unencrypted channel" to "may be
transmitted", since the channel can be protected outside Spark (service mesh
mTLS, a private network) and Spark can't detect that.
- As you suggested, the warning now points at the SPIP's stricter mode: it
ends with "This is a warning by default; per the SPIP, a stricter fail-fast
mode that refuses to start without RPC encryption may be introduced later." So
an operator who wants Spark to *enforce* encryption rather than warn knows
that's the intended path.
On silencing / enforcing: I kept it warn-by-default without a dedicated
opt-out, matching the SPIP (warn by default; the stricter fail-fast mode is the
planned follow-up). If you'd like a knob in this PR, here's the shape I'd
propose. Does this look right to you?
Rather than a "silence the warning" flag, I'd lean toward an enforce flag,
which also covers the silence case (operators who protect the channel
externally just leave it off) and mirrors the sibling
direct-credential-provider path, which already fails fast via `require()`:
```scala
// in internal/config/package.scala, next to the other spark.security.oidc.*
entries
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
SPIP's " +
"warn-by-default behavior.")
.version("4.4.0")
.booleanConf
.createWithDefault(false)
```
Then `warnIfRpcEncryptionDisabled` becomes a check that either throws or
warns, reusing the same shared predicate and remediation so the two messages
still can't drift:
```scala
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 requires RPC channel encryption when "
+
s"${SECURITY_OIDC_REQUIRE_RPC_ENCRYPTION.key}=true. " +
SecurityManager.rpcEncryptionRemediation)
} else {
logWarning(...) // the current warning, unchanged
}
}
}
```
This stays within the SPIP (default is still warn), gives operators a way to
opt into the stricter behavior now, and keeps the enforce message aligned with
both the warning and the direct-provider `require()`. If you think it better
that I hold it for the SPIP's dedicated follow-up instead, I'm fine with that
too; just let me know.
--
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]
