Skip to content

[SPARK-58592][CORE] Redact secrets in Standalone Master RequestMasterState RPC #57796

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wangyum wants to merge 3 commits into
base: master
Choose a base branch
from
+113 −8
Open

[SPARK-58592][CORE] Redact secrets in Standalone Master RequestMasterState RPC #57796

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a1d1fd4
[SPARK-58592][CORE] Redact secrets in persisted Master recovery state
wangyum Aug 5, 2026
9f3ab2c
[SPARK-58592][CORE] Redact secrets in Standalone Master RequestMaster…
wangyum Aug 6, 2026
2a420a2
[SPARK-58592][CORE] Deduplicate redaction logic in Command.redactedCopy
wangyum Aug 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions core/src/main/scala/org/apache/spark/deploy/Command.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,18 @@ package org.apache.spark.deploy

import scala.collection.Map

import org.apache.spark.SparkConf
import org.apache.spark.util.Utils

private[spark] case class Command(
mainClass: String,
arguments: Seq[String],
environment: Map[String, String],
classPathEntries: Seq[String],
libraryPathEntries: Seq[String],
javaOpts: Seq[String]) {

private[deploy] def redactedCopy(conf: SparkConf): Command = copy(
environment = Utils.redact(conf, environment.toSeq).toMap,
javaOpts = Utils.redactCommandLineArgs(conf, javaOpts))
}
24 changes: 24 additions & 0 deletions core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package org.apache.spark.deploy

import scala.collection.immutable.List

import org.apache.spark.SparkConf
import org.apache.spark.deploy.ExecutorState.ExecutorState
import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo, WorkerInfo}
import org.apache.spark.deploy.master.DriverState.DriverState
Expand Down Expand Up @@ -277,6 +278,29 @@ private[deploy] object DeployMessages {

def uri: String = "spark://" + host + ":" + port
def restUri: Option[String] = restPort.map { p => "spark://" + host + ":" + p }

// Must be called before sending the response so writeReplace redacts secrets.
// If unset, writeReplace returns this object unredacted.
@transient private var _conf: SparkConf = _

private[deploy] def withConf(conf: SparkConf): this.type = {
_conf = conf
this
}

private def writeReplace(): Any = {
if (_conf == null) {
this
} else {
MasterStateResponse(
host, port, restPort, workers,
activeApps.map(_.redactedCopy(_conf)),
completedApps.map(_.redactedCopy(_conf)),
activeDrivers.map(_.redactedCopy(_conf)),
completedDrivers.map(_.redactedCopy(_conf)),
status)
}
}
}

// WorkerWebUI to Worker
Expand Down
7 changes: 1 addition & 6 deletions core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, WorkerStateR
import org.apache.spark.deploy.master._
import org.apache.spark.deploy.worker.ExecutorRunner
import org.apache.spark.resource.{ResourceInformation, ResourceRequirement}
import org.apache.spark.util.Utils

private[deploy] object JsonProtocol {

Expand Down Expand Up @@ -130,11 +129,7 @@ private[deploy] object JsonProtocol {
* For compatibility also returns the deprecated `memoryperslave` & `resourcesperslave` fields.
*/
def writeApplicationDescription(obj: ApplicationDescription, conf: SparkConf): JObject = {
val redactedEnvironment = Utils.redact(conf, obj.command.environment.toSeq).toMap
val redactedJavaOpts = Utils.redactCommandLineArgs(conf, obj.command.javaOpts)
val redactedCommand = obj.command.copy(
environment = redactedEnvironment,
javaOpts = redactedJavaOpts)
val redactedCommand = obj.command.redactedCopy(conf)
("name" -> obj.name) ~
("cores" -> obj.maxCores.getOrElse(0)) ~
("memoryperexecutor" -> obj.memoryPerExecutorMB) ~
Expand Down
6 changes: 6 additions & 0 deletions core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import java.util.Date
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.SparkConf
import org.apache.spark.deploy.ApplicationDescription
import org.apache.spark.resource.{ResourceInformation, ResourceProfile, ResourceUtils}
import org.apache.spark.resource.ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID
Expand Down Expand Up @@ -204,4 +205,9 @@ private[spark] class ApplicationInfo(
System.currentTimeMillis() - startTime
}
}

private[deploy] def redactedCopy(conf: SparkConf): ApplicationInfo = {
val redactedDesc = desc.copy(command = desc.command.redactedCopy(conf))
new ApplicationInfo(startTime, id, redactedDesc, submitDate, driver, defaultCores)
}
Comment thread
wangyum marked this conversation as resolved.
}

@uros-b uros-b Aug 5, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Functional regression in HA recovery for supervised drivers and executors. After a master crash and recovery, the objects deserialized from disk are already-redacted copies: desc.command.environment and desc.command.javaOpts contain Utils.REDACTION_REPLACEMENT_TEXT in place of real secrets.

The recovery path in Master.completeRecovery() calls relaunchDriver(d) for every supervised driver whose worker is gone, and relaunchDriver calls createDriver(driver.desc); where driver.desc is the recovered, permanently-redacted DriverDescription. The resulting DriverInfo (even with withConf(conf) attached) has a redacted command, so the worker receives LaunchDriver with REDACTION_REPLACEMENT_TEXT in env vars. Any driver that requires env-var secrets (e.g. HADOOP_CREDSTORE_PASSWORD, AWS_ACCESS_KEY) will fail to authenticate after master-crash recovery. The same applies to executor re-launches: when a recovered application needs new executors, launchExecutor sends exec.application.desc (the redacted ApplicationDescription) to the worker, which builds the executor process with the redacted environment. The PersistenceEngineSuite test explicitly asserts recoveredApp.desc.command.environment("PASSWORD") == Utils.REDACTION_REPLACEMENT_TEXT, confirming the regression.

The fix should store an out-of-band, separate redacted copy for persistence (e.g., serialize a lightweight ApplicationDescription/DriverDescription snapshot with redacted fields) rather than having the live deserialized object carry permanently-redacted state.

@wangyum wangyum Aug 6, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you @uros-b, I reduced the scope to RPC only, and this only affects Spark Standalone mode. ApplicationInfo/DriverInfo are Standalone/Master-specific classes; YARN and Kubernetes do not use them or have an equivalent RPC.

@wangyum wangyum Aug 6, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

May be persistence handling for sensitive fields:

  • When spark.authenticate=true: consider encrypting the persisted data instead of redacting it.
  • When spark.authenticate=false: no change to existing behavior.

8 changes: 8 additions & 0 deletions core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package org.apache.spark.deploy.master

import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.deploy.DriverDescription
import org.apache.spark.resource.ResourceInformation
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -55,4 +56,11 @@ private[deploy] class DriverInfo(
def withResources(r: Map[String, ResourceInformation]): Unit = _resources = r

def resources: Map[String, ResourceInformation] = _resources

private[deploy] def redactedCopy(conf: SparkConf): DriverInfo = {
val redactedDesc = desc.copy(command = desc.command.redactedCopy(conf))
val copy = new DriverInfo(startTime, id, redactedDesc, submitDate)
copy.withResources(_resources)
copy
}
Comment thread
wangyum marked this conversation as resolved.
}
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/deploy/master/Master.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -506,7 +506,7 @@ private[deploy] class Master(
context.reply(MasterStateResponse(
address.host, address.port, restServerBoundPort,
workers.toArray, apps.toArray, completedApps.toArray,
drivers.toArray, completedDrivers.toArray, state))
drivers.toArray, completedDrivers.toArray, state).withConf(conf))

case RequestReadyz =>
context.reply(state != RecoveryState.STANDBY)
Expand Down
67 changes: 66 additions & 1 deletion core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,9 @@ import org.json4s.jackson.JsonMethods

import org.apache.spark.{JsonTestUtils, SparkConf, SparkFunSuite}
import org.apache.spark.deploy.DeployMessages.{MasterStateResponse, WorkerStateResponse}
import org.apache.spark.deploy.master.{ApplicationInfo, RecoveryState, WorkerInfo}
import org.apache.spark.deploy.master.{ApplicationInfo, DriverInfo, RecoveryState, WorkerInfo}
import org.apache.spark.deploy.worker.ExecutorRunner
import org.apache.spark.serializer.JavaSerializer
import org.apache.spark.util.Utils

class JsonProtocolSuite extends SparkFunSuite with JsonTestUtils {
Expand Down Expand Up @@ -138,6 +139,70 @@ class JsonProtocolSuite extends SparkFunSuite with JsonTestUtils {
assert(commandStr.contains("-Xmx2g"))
}

test("SPARK-58592: redactedCopy redacts secrets in ApplicationInfo and DriverInfo") {
val conf = new SparkConf()
val secretEnv = Map(
"PASSWORD" -> "topsecret",
"JAVA_HOME" -> "/usr/lib/jvm/default")
val secretJavaOpts = Seq(
"-Dspark.executorEnv.TOKEN=env-token",
"-Xmx2g")
val cmd = Command("mainClass", List("arg1"), secretEnv, Seq(), Seq(), secretJavaOpts)

val appDesc = ApplicationDescription("name", Some(4), cmd, "appUiUrl", defaultResourceProfile)
val appInfo = new ApplicationInfo(0, "app-1", appDesc, new Date(0), null, Int.MaxValue)
val redactedApp = appInfo.redactedCopy(conf)
assert(!redactedApp.desc.command.environment.contains("topsecret"))
assert(redactedApp.desc.command.environment("PASSWORD") == Utils.REDACTION_REPLACEMENT_TEXT)
assert(redactedApp.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default")
assert(!redactedApp.desc.command.javaOpts.contains("env-token"))
assert(redactedApp.desc.command.javaOpts.contains("-Xmx2g"))

val driverDesc = DriverDescription("hdfs://some.jar", 100, 3, false, cmd)
val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new Date(0))
val redactedDriver = driverInfo.redactedCopy(conf)
assert(!redactedDriver.desc.command.environment.contains("topsecret"))
assert(redactedDriver.desc.command.environment("PASSWORD") == Utils.REDACTION_REPLACEMENT_TEXT)
assert(redactedDriver.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default")
assert(!redactedDriver.desc.command.javaOpts.contains("env-token"))
assert(redactedDriver.desc.command.javaOpts.contains("-Xmx2g"))
}

test("SPARK-58592: writeReplace redacts secrets during RPC serialization") {
val conf = new SparkConf()
val secretEnv = Map("PASSWORD" -> "topsecret", "JAVA_HOME" -> "/usr/lib/jvm/default")
val secretJavaOpts = Seq("-Dspark.executorEnv.TOKEN=env-token", "-Xmx2g")
val cmd = Command("mainClass", List("arg1"), secretEnv, Seq(), Seq(), secretJavaOpts)
val appDesc = ApplicationDescription("name", Some(4), cmd, "appUiUrl", defaultResourceProfile)
val appInfo = new ApplicationInfo(0, "app-1", appDesc, new Date(0), null, Int.MaxValue)
val driverDesc = DriverDescription("hdfs://some.jar", 100, 3, false, cmd)
val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new Date(0))

val stateResponse = new MasterStateResponse(
"host", 8080, None, Array.empty[WorkerInfo], Array(appInfo),
Array.empty[ApplicationInfo], Array(driverInfo),
Array.empty[DriverInfo], RecoveryState.ALIVE).withConf(conf)

val serializer = new JavaSerializer(conf).newInstance()
val serialized = serializer.serialize(stateResponse)
val deserialized = serializer.deserialize[MasterStateResponse](serialized)

val deserializedApp = deserialized.activeApps.head
assert(!deserializedApp.desc.command.environment.contains("topsecret"))
assert(deserializedApp.desc.command.environment("PASSWORD") == Utils.REDACTION_REPLACEMENT_TEXT)
assert(deserializedApp.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default")
assert(!deserializedApp.desc.command.javaOpts.contains("env-token"))
assert(deserializedApp.desc.command.javaOpts.contains("-Xmx2g"))

val deserializedDriver = deserialized.activeDrivers.head
assert(!deserializedDriver.desc.command.environment.contains("topsecret"))
assert(deserializedDriver.desc.command.environment("PASSWORD") ==
Utils.REDACTION_REPLACEMENT_TEXT)
assert(deserializedDriver.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default")
assert(!deserializedDriver.desc.command.javaOpts.contains("env-token"))
assert(deserializedDriver.desc.command.javaOpts.contains("-Xmx2g"))
}

test("SPARK-46883: writeClusterUtilization") {
val workers = Array(createWorkerInfo(), createWorkerInfo())
val activeApps = Array(createAppInfo())
Expand Down