[SPARK-58322][CORE] Verify executor driver identity via per-instance token - #57525
[SPARK-58322][CORE] Verify executor driver identity via per-instance token#57525wangyum wants to merge 1 commit into
Conversation
|
Thank you @wangyum! |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Two application-identity races can still allow an executor to register with the wrong driver. The inline comments describe the startup-cached missing-ID case and the config-fetch-versus-registration connection race.
wangyum
changed the title
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
The rewritten registration-time check fixes the earlier driver identity issues, but it introduces a YARN-client startup race that can reject correctly identified executors.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Four additional issues remain on the current head: the bootstrap fetch still exposes driver credentials before identity validation, registration can compare against an uninitialized fallback application ID, the message change breaks external-backend ABI and source compatibility, and the driver-swap test bypasses the production sender. Details inline.
| attributes, resources, resourceProfileId) => | ||
| if (executorDataMap.contains(executorId)) { | ||
| attributes, resources, resourceProfileId, appId) => | ||
| if (Option(appId).exists(_ != scheduler.applicationId())) { |
There was a problem hiding this comment.
[P1] Wait for the authoritative application ID before comparing
scheduler.applicationId() does not imply that the cluster-manager ID is initialized. In standalone mode, StandaloneSchedulerBackend.applicationId() returns an unrelated generated spark-application-* fallback until the asynchronous connected(appId) callback publishes the master's ID. The master sends RegisteredApplication and immediately calls schedule() over a different RPC path, so a legitimately launched executor can reach this handler first carrying the real master-assigned ID and be rejected against the fallback; those exits consume the executor retry budget and can fail applications configured with a low spark.deploy.maxExecutorRetries. Please expose an explicit, safely published ID-ready state and defer/retry registration until it is ready instead of comparing against the fallback.
There was a problem hiding this 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-scale timeout, and add a regression that delays ID publication beyond 300 ms.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| resources: Map[String, ResourceInformation], | ||
| resourceProfileId: Int) | ||
| resourceProfileId: Int, | ||
| appId: String = null) |
There was a problem hiding this comment.
[P2] Preserve the existing RegisterExecutor ABI
The default argument only helps newly compiled Scala call sites. Adding a ninth case-class field changes the generated constructor, apply, and unapply signatures: Java or precompiled external backends using the eight-argument form can fail linkage, and eight-field pattern matches no longer compile (the required BlockManagerSuite edits demonstrate the extractor break). A live external example is Armada's eight-field match. Since this PR explicitly intends compatibility with custom/external cluster managers, please retain the old RegisterExecutor shape and introduce a versioned identity-carrying message or equivalent compatibility path.
There was a problem hiding this comment.
add appId in attributes, then we do not need add a new field
| val ex = intercept[SparkException] { | ||
| backend.driverEndpoint.askSync[Boolean]( | ||
| RegisterExecutor("1", mockEndpointRef, mockAddress.host, 1, Map.empty, Map.empty, | ||
| Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID, fetchedAppId)) |
There was a problem hiding this comment.
[P2] Exercise the production sender in this regression
This test reads a synthetic config and then manually places fetchedAppId into RegisterExecutor; it never runs CoarseGrainedExecutorBackend.onStart or verifies that the launch arguments.appId reaches the only changed production sender. Removing or miswiring that sender still leaves every new test green. Please drive the executor-backend bootstrap/registration path and capture the emitted RegisterExecutor, asserting that it contains the launch application ID. If the test keeps claiming an exact port-reuse regression, it should also shut down A and have B bind A's released address.
wangyum
changed the title
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Five issues remain on this head: three P1 security/correctness problems, including five failing existing CI tests, and two P2 compatibility/test gaps. Details inline.
| } | ||
|
|
||
| private def appIdMismatch(executorAppId: Option[String]): Option[SparkException] = { | ||
| val resolvedDriverAppId = scheduler.backend.realApplicationId() |
There was a problem hiding this comment.
[P1] Do not dereference the scheduler backend before checking executor identity
TaskSchedulerImpl.backend starts as null, and existing mocked/custom schedulers do not necessarily populate it, but this lookup runs unconditionally even when the executor supplied no application ID. The current core CI job already reports five existing failures across CoarseGrainedSchedulerBackendSuite, StandaloneDynamicAllocationSuite, and HeartbeatReceiverSuite, all caused by NullPointerException at this line; I also reproduced extra resources from executor locally. Please check whether an executor ID exists before resolving driver identity and obtain it from the actual enclosing backend instead of dereferencing the nullable scheduler backlink.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| val resolvedDriverAppId = scheduler.backend.realApplicationId() | ||
| for { | ||
| execAppId <- executorAppId | ||
| driverAppId <- resolvedDriverAppId |
There was a problem hiding this comment.
[P1] Do not accept an executor while the driver application ID is unavailable
When realApplicationId() returns None, this comprehension reports no mismatch, so both RetrieveSparkAppConfig and RegisterExecutor accept any concrete executor application ID. This is a real production window: standalone exposes its driver endpoint before asynchronous connected(appId), and YARN client calls super.start() before submitApplication()/bindToYarn(). An executor from old application A reaching reused-port driver B during that interval can receive B's I/O-encryption key/delegation credentials and register permanently before B's identity is published; nothing revalidates it afterward. The new fallback test actually asserts that "any-app-id" is accepted. Please defer both bootstrap and registration until the authoritative driver ID is available, then validate the executor.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| context.reply(reply) | ||
| case RetrieveSparkAppConfig(resourceProfileId, appId) => | ||
| // Validate identity before returning bootstrap credentials. | ||
| appIdMismatch(Option(appId)) match { |
There was a problem hiding this comment.
[P1] Require executor identity before returning bootstrap credentials
Option(appId) converts the compatibility default null into None, after which the success branch returns SparkAppConfig including Spark properties, the I/O-encryption key, and Hadoop delegation credentials. This is not hypothetical: RayDP still sends the one-argument request, so recompiling it against this change supplies null and leaves wrong-driver credential disclosure intact; registration without an app-ID attribute is likewise explicitly accepted. RPC authentication is disabled by default. Please require a verified application ID for credential-bearing bootstrap requests, or provide a compatibility path with equivalent authentication rather than silently failing open.
There was a problem hiding this 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.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| case class RetrieveSparkAppConfig(resourceProfileId: Int) extends CoarseGrainedClusterMessage | ||
| case class RetrieveSparkAppConfig( | ||
| resourceProfileId: Int, | ||
| appId: String = null) |
There was a problem hiding this comment.
[P2] Preserve the existing RetrieveSparkAppConfig ABI
The default argument preserves only newly compiled Scala construction syntax. Adding this field removes the existing JVM (int) constructor and companion apply(int), and changes the one-field extractor; javap confirms the new class exposes only (int, String). A real Spark-Proxy Java integration directly invokes new RetrieveSparkAppConfig(0), which now fails even after recompilation, while previously compiled RayDP artifacts fail linkage. This moves the earlier RegisterExecutor compatibility problem to the bootstrap message. Please retain the original one-field request and introduce a separately versioned identity-carrying request, or otherwise preserve the old constructor, factory, and extractor.
| env.executorBackend = Option(this) | ||
| ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls, | ||
| extractAttributes, _resources, resourceProfile.id)) | ||
| extractAttributes + ("spark.app.id" -> env.conf.getAppId), |
There was a problem hiding this comment.
[P2] Add a regression test for the production registration sender
Every added core regression manually constructs RegisterExecutor; none runs CoarseGrainedExecutorBackend.onStart or captures the registration message produced here. The new Kubernetes test only captures RetrieveSparkAppConfig and deliberately throws from its backend factory before registration. Therefore deleting or miswiring this "spark.app.id" insertion leaves all eight new SPARK-58322 core tests green while reopening the post-bootstrap driver-swap vulnerability. Please add a fake-driver/executor-backend test that captures the actual RegisterExecutor emitted by onStart and verifies its attributes contain the launch application ID.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Three new P1 issues are inline; two existing P1 discussions include updated evidence.
| "and RPC authentication is not enabled (spark.authenticate=false). " + | ||
| "RPC authentication must be enabled for this legacy request.")) | ||
| } else { | ||
| replySparkAppConfig(resourceProfileId, context) |
There was a problem hiding this 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.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| val resolvedDriverAppId = this.realApplicationId() | ||
| val execAppId = executorAppId.filter(_.nonEmpty) | ||
| (execAppId, resolvedDriverAppId) match { | ||
| case (Some(exec), Some(driver)) if exec != driver => |
There was a problem hiding this 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.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient:
| "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") |
There was a problem hiding this 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.
There was a problem hiding this comment.
We use a UUID here instead of an appid, which makes things much more convenient, we do not need to add new config:
| context.reply(reply) | ||
| case RetrieveSparkAppConfig(resourceProfileId, appId) => | ||
| // Validate identity before returning bootstrap credentials. | ||
| appIdMismatch(Option(appId)) match { |
There was a problem hiding this 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.
| attributes, resources, resourceProfileId) => | ||
| if (executorDataMap.contains(executorId)) { | ||
| attributes, resources, resourceProfileId, appId) => | ||
| if (Option(appId).exists(_ != scheduler.applicationId())) { |
There was a problem hiding this 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-scale timeout, and add a regression that delays ID publication beyond 300 ms.
wangyum
changed the title
| env.executorBackend = Option(this) | ||
| ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores, extractLogUrls, | ||
| extractAttributes, _resources, resourceProfile.id)) | ||
| extractAttributes ++ Map(EXECUTOR_DRIVER_INSTANCE_TOKEN -> getDriverInstanceToken), |
There was a problem hiding this comment.
[P1] Keep the driver instance token out of public executor attributes
Putting the bearer UUID in RegisterExecutor.attributes makes it public as soon as the first executor registers. DriverEndpoint copies the same attributes into ExecutorData and publishes SparkListenerExecutorAdded; EventLoggingListener.onExecutorAdded writes that event without redaction, JsonProtocol.executorInfoToJson serializes its Attributes map verbatim, and AppStatusListener exposes the same map through /api/v1/applications/<app-id>/executors. The normal spark.redaction.regex protects environment updates but is not applied to executor attributes, and UI ACLs are disabled by default. Anyone able to read the executor API or event log can replay the token in RetrieveSparkAppConfigWithIdentity to obtain the application's Hadoop delegation credentials/I/O-encryption key or register a fake executor. Remove the token from the attributes immediately after validation, before creating ExecutorData or publishing the event, and add event-log/REST non-disclosure coverage.
| // Propagate the token as an environment variable to executor processes. | ||
| // setExecutorEnv covers YARN and Kubernetes; sc.executorEnvs is updated directly for Standalone. | ||
| conf.setExecutorEnv(EXECUTOR_DRIVER_INSTANCE_TOKEN, driverInstanceToken) | ||
| scheduler.sc.executorEnvs(EXECUTOR_DRIVER_INSTANCE_TOKEN) = driverInstanceToken |
There was a problem hiding this comment.
[P1] Do not disclose driver tokens through Standalone master state
Adding the bearer token to sc.executorEnvs causes StandaloneSchedulerBackend.start to embed it in ApplicationDescription.command.environment (StandaloneSchedulerBackend.scala:115-135). The Standalone master retains that complete serializable description and returns every application's unredacted ApplicationInfo.desc.command.environment to any caller of RequestMasterState (Master.scala:505-509). docs/security.md:44-50 explicitly documents Standalone applications and daemons sharing the same RPC authentication secret, so application A can query the master for application B's raw token and immediately replay it against B's credential-bearing bootstrap or registration endpoint. SparkConf/UI redaction does not protect this RPC response, and fixing the separate executor-attribute leak does not close this earlier exposure. Avoid distributing per-application bearer secrets through globally readable master state, and add a regression covering two applications sharing the cluster RPC secret.
| case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls, | ||
| attributes, resources, resourceProfileId) => | ||
| if (executorDataMap.contains(executorId)) { | ||
| val mismatch = verifyDriverInstanceToken(attributes.get(EXECUTOR_DRIVER_INSTANCE_TOKEN)) |
There was a problem hiding this comment.
[P1] Update all existing registration senders before requiring the token
This new check rejects existing in-tree RegisterExecutor(..., attributes = Map.empty, ...) callers that were not updated in HeartbeatReceiverSuite.scala:180-185 and StandaloneDynamicAllocationSuite.scala:507-508,630-632. The exact-head Core JUnit artifacts confirm 11 failing tests: one in HeartbeatReceiverSuite and ten in StandaloneDynamicAllocationSuite, with the underlying exception SparkException: Executor did not supply a driver instance token. at this exact line. Nine standalone cases fail in syncExecutors, and the excluded-host regression now receives the missing-token exception instead of the expected IllegalStateException. Populate the matching driver token in every existing synthetic registration and restore the required Core check.
| MockitoAnnotations.openMocks(this).close() | ||
| when(taskScheduler.sc).thenReturn(sc) | ||
| when(sc.conf).thenReturn(sparkConf) | ||
| when(sc.executorEnvs).thenReturn(new scala.collection.mutable.HashMap[String, String]) |
There was a problem hiding this comment.
[P2] Stub the isolated Kubernetes SparkContext environment too
This new setup only stubs executorEnvs on the shared sc mock. The existing SPARK-56238: applicationId() is stable across calls when spark.app.id is not set test below creates a separate localSc = mock(classOf[SparkContext]) at lines 324-340 without stubbing localSc.executorEnvs. Constructing its KubernetesClusterSchedulerBackend now immediately executes scheduler.sc.executorEnvs(EXECUTOR_DRIVER_INSTANCE_TOKEN) = driverInstanceToken in the base constructor; Mockito returns null for that separate mock, so the test throws NullPointerException before it reaches either application-ID assertion. Add the same mutable-map stub for localSc so the existing Kubernetes regression continues to run.
…ariable
Replace SparkConf-based token propagation with environment variable:
- Remove EXECUTOR_DRIVER_INSTANCE_TOKEN config builder and
isExecutorStartupConf entry
- Add ENV_EXECUTOR_DRIVER_INSTANCE_TOKEN constant and
getDriverInstanceToken helper reading from System.getenv
- Use conf.setExecutorEnv + sc.executorEnvs for propagation
- Revert sparkProperties filter (token no longer in SparkConf)
Generated-by: GLM 5.2.
fix
[SPARK-58322][CORE] Simplify executor identity verification to token-only
Remove redundant appId verification, readiness retry logic, and
DriverAppIdNotReadyException. The per-driver instance token alone
provides the security property: it is generated before the driver RPC
endpoint is created, so there is no startup race window.
- Rename config to spark.executor.driverInstanceToken for redaction
coverage (matches default regex 'token')
- Add .version("4.3.0").withBindingPolicy(NOT_APPLICABLE)
- Simplify RetrieveSparkAppConfigWithIdentity to 2 fields
- Remove realApplicationId() from SchedulerBackend and all overrides
- Remove registration retry executor and fetchSparkAppConfig retry loop
- Remove EXECUTOR_APP_ID_ATTR, DriverAppIdNotReadyException
- Revert @volatile/visibility changes on Yarn/StandaloneSchedulerBackend
Net: -212 lines of production code and tests
[SPARK-58322][CORE] Bound identity readiness retries by RPC timeout
Use one monotonic deadline derived from spark.rpc.askTimeout for bootstrap
and registration retries, rather than multiplying the RPC timeout by a retry
count. Retry delays are capped by the deadline's remaining time. Extend
regressions to hold the driver identity unavailable beyond the prior 300 ms
limit.
[SPARK-58322][CORE] Bind executors to a driver instance token
Generate a random token for every coarse-grained driver instance and pass it
as executor startup configuration across Standalone, YARN, and Kubernetes.
Require the token together with the application ID for bootstrap and
registration, preventing driver swaps even when an app ID or RPC secret is
reused. Reject the legacy credential-bearing request, derive startup retry
limits from the RPC timeout, and add missing config binding policy and tests.
[SPARK-58322][CORE] Retry executor identity checks while driver starts
Reject disabled identity verification without RPC authentication, so legacy
bootstrap requests cannot expose credentials unauthenticated. Introduce a
transient driver-ID-not-ready error and retry both bootstrap config fetches
and executor registration before failing. Share bootstrap retry logic with
the Kubernetes executor backend and cover both retry paths with tests.
[SPARK-58322][CORE] PMC review: volatile, config switch, SECURITY.md, code cleanup
- Add @volatile to YarnSchedulerBackend.appId for cross-thread visibility
- Fix realApplicationId() Scaladoc to say 'reject' not 'skip', add @SInCE 4.3.0
- Fix error message: clarify cluster manager retries, not the executor
- Add spark.executor.identityVerification.enabled (default true) for
backward compatibility with external cluster managers
- Extract EXECUTOR_APP_ID_ATTR constant in CoarseGrainedClusterMessages
- Make StandaloneSchedulerBackend.appId private[scheduler] to replace
fragile reflection in tests
- Extract replySparkAppConfig helper to eliminate code duplication
- Update SECURITY.md with executor identity verification threat model
[SPARK-58322][CORE] Address review: fail-closed, preserve ABI, add production sender test
- P1: Fix NPE by using this.realApplicationId() instead of scheduler.backend
- P1: Fail closed when realApplicationId() returns None (reject both
config fetch and registration instead of silently accepting)
- P1: Require verified appId for credential-bearing RetrieveSparkAppConfig;
gate the old one-field message on RPC auth being enabled
- P2: Preserve RetrieveSparkAppConfig ABI by reverting to one-field and
adding a new RetrieveSparkAppConfigWithIdentity message
- P2: Add production sender test exercising CoarseGrainedExecutorBackend.onStart
to verify RegisterExecutor carries spark.app.id in attributes
[SPARK-58322][CORE] Validate executor app ID before returning credentials and at registration
This patch implements a two-layer application-identity check to prevent
an executor from registering with the wrong driver:
Layer 1: RetrieveSparkAppConfig now carries executor appId; the driver
validates it before returning I/O encryption key and Hadoop delegation
tokens.
Layer 2: RegisterExecutor carries appId in its attributes map (preserving
the 8-field ABI for external backends); the driver validates it at
registration time as defense-in-depth against driver swap after config
fetch.
A new SchedulerBackend.realApplicationId() method replaces the fragile
"spark-application-*" string-prefix heuristic. StandaloneSchedulerBackend
and YarnSchedulerBackend override it to return None before their
respective cluster-assigned IDs are available (connected() / bindToYarn()),
so the check is cleanly skipped during the startup race window.
Added 11 tests: config-fetch rejection/acceptance/null-compat, registration
rejection/acceptance/no-attr-compat, driver swap, standalone fallback,
YARN client race, production CoarseGrainedExecutorBackend.run() sender,
and Kubernetes executor backend sender.
[SPARK-58322][CORE] Clean up SPARK-58322 test: fix indentation, add try/finally
[SPARK-58322][CORE] Verify executor app ID matches driver at registration
Add appId to RegisterExecutor message and validate it in DriverEndpoint
against scheduler.sc.applicationId. This binds the identity check to the
registration connection itself, preventing the config-fetch-vs-registration
race where a wrong driver could bind the same port between config retrieval
and registration.
The check replaces the executor-side verifyAppId which ran on a transient
RPC connection that was shut down before registration. The driver-side check
is authoritative since scheduler.sc.applicationId is always set before any
executor can register.
Null appId is accepted for backward compatibility with custom/external
cluster managers that construct RegisterExecutor without appId.
verify appid
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Two additional P1 security gaps remain on the current head: Kubernetes Pod metadata exposes the driver token, and Spark Connect/SQL session configuration exposes it independently. Details inline.
| private[spark] val driverInstanceToken = UUID.randomUUID().toString | ||
| // Propagate the token as an environment variable to executor processes (YARN, Kubernetes). | ||
| // Standalone mode adds it to Command.environment directly in StandaloneSchedulerBackend. | ||
| conf.setExecutorEnv(EXECUTOR_DRIVER_INSTANCE_TOKEN, driverInstanceToken) |
There was a problem hiding this comment.
[P1] Keep the driver bearer token out of Kubernetes Pod specifications
setExecutorEnv sends this UUID through KubernetesExecutorConf.environment into BasicExecutorFeatureStep, where KubernetesUtils.buildEnvVars writes it as a literal EnvVar.value beside SPARK_DRIVER_URL. The driver's service account needs Pod get/list/watch permissions, and executors inherit that service account by default, so an executor or another application sharing the namespace can read application B's executor Pod and recover both B's driver address and token. With the default spark.authenticate=false, it can replay RetrieveSparkAppConfigWithIdentity to obtain B's I/O encryption key and Hadoop delegation tokens, or submit RegisterExecutor; thus this proposed per-driver identity check provides no isolation in that deployment. The existing documentation about Pod exposure of _SPARK_AUTH_SECRET applies when RPC authentication is enabled and does not cover the default unauthenticated case. Please distribute the token through an application-isolated channel instead of a literal Pod environment value and add a generated-Pod non-disclosure regression.
| private lazy val sparkProperties = scheduler.sc.conf.getAll | ||
| .filter { case (k, _) => k.startsWith("spark.") } | ||
| .filter { case (k, _) => | ||
| k.startsWith("spark.") && k != s"spark.executorEnv.$EXECUTOR_DRIVER_INSTANCE_TOKEN" |
There was a problem hiding this comment.
[P1] Keep the driver bearer token out of user-visible session configuration
Filtering the SparkAppConfig reply does not remove the token already stored in scheduler.sc.conf. SparkSession copies sparkContext.conf.getAll into every session's SQLConf, and SparkConnectConfigHandler.handleGet/handleGetAll return those values without redaction, so any Spark Connect client can directly call spark.conf.get("spark.executorEnv.SPARK_EXECUTOR_DRIVER_INSTANCE_TOKEN") and recover the raw per-driver bearer token. SQL-only Thrift/JDBC clients can also retrieve it with SET spark.sql.legacy.setCommandRejectsSparkCoreConfs=false; SET spark.redaction.regex=^$; SET spark.executorEnv.SPARK_EXECUTOR_DRIVER_INSTANCE_TOKEN;; I reproduced this sequence and confirmed it changes only session-local redaction. When the driver RPC endpoint is reachable with default authentication disabled, that token permits credential-bearing bootstrap requests or executor registration, despite Spark Connect's documented isolation from driver/static configuration. Please keep this bearer secret out of shared SparkConf/SQL session state, or prevent its extraction across every remote configuration boundary, and add Connect plus SQL non-disclosure regressions.
3 participants
What changes were proposed in this pull request?
This PR introduces a per-driver-instance token that the driver validates 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 after a port-reuse / driver-swap scenario.
Changes:
Token generation (
CoarseGrainedSchedulerBackend): The driver generates a randomUUID.randomUUIDtoken in its constructor, before the RPC endpoint is created (no startup race). The token is propagated to executor processes as the environment variableSPARK_EXECUTOR_DRIVER_INSTANCE_TOKENvia bothconf.setExecutorEnv()(covers YARN and Kubernetes) andscheduler.sc.executorEnvs()(covers Standalone, sinceexecutorEnvsis snapshotted before the backend constructor runs).New RPC message (
CoarseGrainedClusterMessage):RetrieveSparkAppConfigWithIdentity(resourceProfileId, driverInstanceToken)— the executor sends this instead of the legacyRetrieveSparkAppConfigso the driver can verify identity before returning credentials. The legacyRetrieveSparkAppConfigis rejected fail-closed with an actionable error message.Token verification at registration (
CoarseGrainedSchedulerBackend):RegisterExecutorattributes now carry the token; the driver validates it viaverifyDriverInstanceToken()and rejects on mismatch or missing token.Executor-side token reader (
CoarseGrainedExecutorBackend):getDriverInstanceTokenreads the token fromSystem.getenvwith aSystem.getPropertyfallback for unit tests. The token is sent in bothRetrieveSparkAppConfigWithIdentityandRegisterExecutor.Kubernetes executor (
KubernetesExecutorBackend): UsesRetrieveSparkAppConfigWithIdentitywith the token fromCoarseGrainedExecutorBackend.getDriverInstanceToken.Token filtered from SparkAppConfig (
CoarseGrainedSchedulerBackend):sparkPropertiesfilters outspark.executorEnv.SPARK_EXECUTOR_DRIVER_INSTANCE_TOKENso the token never appears in theSparkAppConfigRPC response.Documentation: Added "Executor Identity Verification" section to both
SECURITY.mdanddocs/security.md.Why are the changes needed?
When a driver process dies but its RPC port is reused by another driver on the same host, newly launched executors for the original application can connect to the wrong driver. Without identity verification, the wrong driver returns its bootstrap credentials and accepts the executor registration, risking data corruption.
spark.authenticateRPC authentication is not sufficient because deployments can share an RPC authentication secret across applications, so authentication alone is not application-scoped. The per-driver-instance token closes this gap.Does this PR introduce any user-facing change?
The legacy
RetrieveSparkAppConfigRPC message is now rejected by the driver. This only affects internal RPC between Spark processes — no public API change. Executors created from older Spark versions that sendRetrieveSparkAppConfigwill fail to start with a clear error message directing them to useRetrieveSparkAppConfigWithIdentity. Since executor and driver versions must match within a Spark application, this is not expected to impact users in practice.How was this patch tested?
Added unit tests covering:
CoarseGrainedSchedulerBackendSuite(7 tests): token generation and propagation via bothsetExecutorEnvandexecutorEnvs; token filtered fromsparkProperties;RegisterExecutoraccepted with correct token, rejected with wrong token, rejected with missing token;RetrieveSparkAppConfigWithIdentityaccepted with correct token, rejected with wrong token; legacyRetrieveSparkAppConfigrejected.CoarseGrainedExecutorBackendSuite(2 tests):RegisterExecutorcarries the token in attributes when token is set via system property; production sender path includes token.KubernetesClusterSchedulerBackendBackendSuite(1 test): token propagated to executor environment variables in K8s mode.Was this patch authored or co-authored using generative AI tooling?
Generated-by: GLM 5.2.