sunchao commented on code in PR #57525:
URL: https://github.com/apache/spark/pull/57525#discussion_r3714069804
##########
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala:
##########
@@ -104,7 +104,8 @@ private[spark] class CoarseGrainedExecutorBackend(
driver = Some(ref)
env.executorBackend = Option(this)
ref.ask[Boolean](RegisterExecutor(executorId, self, hostname, cores,
extractLogUrls,
- extractAttributes, _resources, resourceProfile.id))
+ extractAttributes ++ Map(EXECUTOR_DRIVER_INSTANCE_TOKEN ->
getDriverInstanceToken),
Review 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 cov
erage.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -62,6 +63,11 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
// Total number of executors that are currently registered
protected val totalRegisteredExecutors = new AtomicInteger(0)
protected val conf = scheduler.sc.conf
+ private val driverInstanceToken = UUID.randomUUID().toString
+ // 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
Review 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 cl
uster RPC secret.
##########
resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala:
##########
@@ -119,6 +120,7 @@ class KubernetesClusterSchedulerBackendSuite extends
SparkFunSuite with BeforeAn
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])
Review 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.
##########
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala:
##########
@@ -244,11 +263,27 @@ class CoarseGrainedSchedulerBackend(scheduler:
TaskSchedulerImpl, val rpcEnv: Rp
logError(log"Received unexpected message. ${MDC(ERROR, e)}")
}
+ private def replySparkAppConfig(resourceProfileId: Int, context:
RpcCallContext): Unit = {
+ val rp =
scheduler.sc.resourceProfileManager.resourceProfileFromId(resourceProfileId)
+ val reply = SparkAppConfig(
+ sparkProperties,
+ SparkEnv.get.securityManager.getIOEncryptionKey(),
+ Option(delegationTokens.get()),
+ rp,
+ currentLogLevel)
+ context.reply(reply)
+ }
+
override def receiveAndReply(context: RpcCallContext):
PartialFunction[Any, Unit] = {
case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
attributes, resources, resourceProfileId) =>
- if (executorDataMap.contains(executorId)) {
+ val mismatch =
verifyDriverInstanceToken(attributes.get(EXECUTOR_DRIVER_INSTANCE_TOKEN))
Review 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.
--
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]
