sunchao commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3706641145
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -140,6 +140,20 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
ThreadUtils.newDaemonSingleThreadScheduledExecutor("cleanup-decommission-execs")
}
+ private def appIdMismatch(executorAppId: Option[String]):
Option[SparkException] = {
+ val resolvedDriverAppId = this.realApplicationId()
+ val execAppId = executorAppId.filter(_.nonEmpty)
+ (execAppId, resolvedDriverAppId) match {
+ case (Some(exec), Some(driver)) if exec != driver =>
Review Comment:
[P1] Do not treat reusable application IDs as unique application-instance
identities
This comparison assumes that equal application IDs identify the same
driver/application instance, but Spark Standalone explicitly supports
`spark.master.useAppNameAsAppId.enabled=true`. With that documented option,
`Master.createApplication` derives the ID only from the normalized application
name, so successive runs of the same job receive the same nonempty ID;
configurable application-ID patterns and modulo can also repeat IDs. If an
executor from the old run connects after a replacement driver for the same job
binds the released address, both the bootstrap check and the registration check
accept it, exposing the replacement driver's encryption key/delegation tokens
and permitting wrong-instance execution. Please bind the checks to a
per-application-instance/driver nonce, or explicitly reject configurations that
cannot provide unique IDs, and add a same-name driver-swap regression.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -353,14 +393,37 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
triggeredByExecutor = true))
case RetrieveSparkAppConfig(resourceProfileId) =>
- val rp =
scheduler.sc.resourceProfileManager.resourceProfileFromId(resourceProfileId)
- val reply = SparkAppConfig(
- sparkProperties,
- SparkEnv.get.securityManager.getIOEncryptionKey(),
- Option(delegationTokens.get()),
- rp,
- currentLogLevel)
- context.reply(reply)
+ if (!SparkEnv.get.securityManager.isAuthenticationEnabled()) {
+ // External callers that still use the original one-field message do
+ // not supply an appId for validation. Only return credentials (I/O
+ // encryption key, delegation tokens) when RPC authentication is
+ // enabled -- otherwise the identity is unverifiable and credential
+ // disclosure to a wrong driver is possible.
+ context.sendFailure(new SparkException("Executor did not supply an
application ID " +
+ "and RPC authentication is not enabled (spark.authenticate=false).
" +
+ "RPC authentication must be enabled for this legacy request."))
+ } else {
+ replySparkAppConfig(resourceProfileId, context)
Review Comment:
[P1] Require application-scoped authentication before returning legacy
bootstrap credentials
This branch treats `spark.authenticate=true` as proof that the caller
belongs to this application and returns `SparkAppConfig`, including the I/O
encryption key and Hadoop delegation credentials, without validating any
application ID. However, `docs/security.md:44-50` explicitly states that
standalone and other deployments can share the same configured
`spark.authenticate.secret` across all applications and daemons;
`SecurityManager.getSecretKey(appId)` also ignores `appId`. Consequently, an
executor from application A can authenticate to replacement driver B with their
shared secret, send the legacy `RetrieveSparkAppConfig`, and obtain B's
bootstrap secrets. RayDP still uses that exact legacy request, and the new
authenticated-legacy test currently asserts this vulnerable behavior. Please
require application identity for credential-bearing legacy requests unless
authentication is demonstrably scoped to one application, and add a
cross-application shared-secret regression test.
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -2695,6 +2695,20 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val EXECUTOR_IDENTITY_VERIFICATION_ENABLED =
+ ConfigBuilder("spark.executor.identityVerification.enabled")
+ .doc("When enabled, the driver validates the executor's application ID
before " +
+ "returning bootstrap credentials (I/O encryption key, Hadoop
delegation " +
+ "tokens) and before accepting executor registration. This prevents an
" +
+ "executor from connecting to the wrong driver in a port-reuse
scenario. " +
+ "External cluster managers that do not supply an application ID in the
" +
+ "identity-carrying RPC messages can set this to false for backward " +
+ "compatibility. The legacy bootstrap RPC always requires RPC " +
+ "authentication (spark.authenticate) to return credentials.")
+ .version("4.3.0")
Review Comment:
[P1] Declare the required binding policy for the new configuration
This new `ConfigBuilder` does not set a binding policy, so existing
`SparkConfigBindingPolicySuite.Config enforcement for bindingPolicy`
deterministically fails. The exact-head test-results check reports
`spark.executor.identityVerification.enabled` as its sole failing
configuration, and the Hive job is red for this reason. Since executor identity
verification does not affect SQL view/UDF/procedure resolution, add
`.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)` before `.booleanConf`;
do not add the setting to the frozen exceptions list.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -352,15 +365,20 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
adjustTargetNumExecutors = false,
triggeredByExecutor = true))
- case RetrieveSparkAppConfig(resourceProfileId) =>
- val rp =
scheduler.sc.resourceProfileManager.resourceProfileFromId(resourceProfileId)
- val reply = SparkAppConfig(
- sparkProperties,
- SparkEnv.get.securityManager.getIOEncryptionKey(),
- Option(delegationTokens.get()),
- rp,
- currentLogLevel)
- context.reply(reply)
+ case RetrieveSparkAppConfig(resourceProfileId, appId) =>
+ // Validate identity before returning bootstrap credentials.
+ appIdMismatch(Option(appId)) match {
Review Comment:
[P1] This issue remains partially unfixed on
`d47091b0f25ac40cbf9c7a0fb353b17cb9d5d726`: the new code rejects missing or
empty application IDs for `RetrieveSparkAppConfigWithIdentity`, but
`RegisterExecutor` still calls
`appIdMismatch(attributes.get(EXECUTOR_APP_ID_ATTR))`. An absent attribute
becomes `None`; an empty attribute is filtered to `None`; and the wildcard
branch reports no mismatch, including when the driver's application ID is
unavailable. The new `accept RegisterExecutor without appId in attributes for
backward compatibility` test explicitly proves that registration succeeds with
verification enabled and RPC authentication disabled. Thus a legacy/external
executor can still register with the wrong driver after a post-bootstrap port
swap, bypassing the second protection layer. Please reject missing/empty
registration identities unless the connection has genuinely application-scoped
authentication, and change the acceptance test into a rejection regression.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -247,8 +247,12 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
- attributes, resources, resourceProfileId) =>
- if (executorDataMap.contains(executorId)) {
+ attributes, resources, resourceProfileId, appId) =>
+ if (Option(appId).exists(_ != scheduler.applicationId())) {
Review Comment:
[P1] This race still survives the new retry commit
`d47091b0f25ac40cbf9c7a0fb353b17cb9d5d726`. Both bootstrap and registration
hard-code `MAX_APP_ID_RETRIES = 3` with `APP_ID_RETRY_DELAY_MS = 100`, so an
executor gives up after approximately 300 ms. The standalone master sends
`RegisteredApplication` asynchronously and immediately schedules executors;
`StandaloneAppClient` processes that callback on the shared RPC dispatcher,
while the driver endpoint rejecting executor requests runs on a dedicated loop.
Shared-dispatcher backlog, network delay, or a JVM pause can therefore keep the
driver's ID unpublished past all four attempts even though
`StandaloneAppClient` itself allows 20-second registration windows. The
resulting executor exits still count toward `spark.deploy.maxExecutorRetries`,
allowing a startup burst to fail the whole application. The added tests cover
only one immediately recovered rejection. Please wait for actual driver
readiness or use a configurable startup/RPC-s
cale timeout, and add a regression that delays ID publication beyond 300 ms.
--
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]
