From a1d1fd4e42b76624ec2d9f27d22375f353dfb82b Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Wed, 5 Aug 2026 23:34:08 +0800 Subject: [PATCH 1/3] [SPARK-58592][CORE] Redact secrets in persisted Master recovery state ApplicationInfo/DriverInfo are Java-serialized as-is by PersistenceEngine (ZooKeeper/filesystem/RocksDB) for Master HA recovery, so secrets in an app or driver's Command (env vars, -D java opts) end up in plaintext in ZK znodes or the recovery directory. Tag each instance with the SparkConf at creation time and redact via writeReplace so the redacted copy is what actually gets serialized, without changing the in-memory object used by the running Master. --- .../spark/deploy/master/ApplicationInfo.scala | 20 +++++++ .../spark/deploy/master/DriverInfo.scala | 22 +++++++ .../apache/spark/deploy/master/Master.scala | 4 +- .../spark/deploy/JsonProtocolSuite.scala | 60 ++++++++++++++++++- .../master/PersistenceEngineSuite.scala | 51 ++++++++++++++++ 5 files changed, 154 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala index e66933b84af55..bff2f7b7466d2 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala @@ -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 @@ -204,4 +205,23 @@ private[spark] class ApplicationInfo( System.currentTimeMillis() - startTime } } + + private[deploy] def redactedCopy(conf: SparkConf): ApplicationInfo = { + val redactedCommand = desc.command.copy( + environment = Utils.redact(conf, desc.command.environment.toSeq).toMap, + javaOpts = Utils.redactCommandLineArgs(conf, desc.command.javaOpts)) + val redactedDesc = desc.copy(command = redactedCommand) + new ApplicationInfo(startTime, id, redactedDesc, submitDate, driver, defaultCores) + } + + @transient private var _conf: SparkConf = _ + + private[deploy] def withConf(conf: SparkConf): this.type = { + _conf = conf + this + } + + private def writeReplace(): AnyRef = { + if (_conf == null) this else redactedCopy(_conf) + } } diff --git a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala index 252e7048ba105..5644747bd61d1 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala @@ -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 @@ -55,4 +56,25 @@ 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 redactedCommand = desc.command.copy( + environment = Utils.redact(conf, desc.command.environment.toSeq).toMap, + javaOpts = Utils.redactCommandLineArgs(conf, desc.command.javaOpts)) + val redactedDesc = desc.copy(command = redactedCommand) + val copy = new DriverInfo(startTime, id, redactedDesc, submitDate) + copy.withResources(_resources) + copy + } + + @transient private var _conf: SparkConf = _ + + private[deploy] def withConf(conf: SparkConf): this.type = { + _conf = conf + this + } + + private def writeReplace(): AnyRef = { + if (_conf == null) this else redactedCopy(_conf) + } } diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index e95c4bd8c6222..ca43918916fb7 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -1136,7 +1136,7 @@ private[deploy] class Master( } else { newApplicationId(date) } - new ApplicationInfo(now, appId, desc, date, driver, defaultCores) + new ApplicationInfo(now, appId, desc, date, driver, defaultCores).withConf(conf) } private[master] def registerApplication(app: ApplicationInfo): Unit = { @@ -1349,7 +1349,7 @@ private[deploy] class Master( val now = System.currentTimeMillis() val date = new Date(now) val id = newDriverId(date) - new DriverInfo(now, id, maybeUpdateAppName(desc, id), date) + new DriverInfo(now, id, maybeUpdateAppName(desc, id), date).withConf(conf) } private def launchDriver(worker: WorkerInfo, driver: DriverInfo): Unit = { diff --git a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala index 6d2c663a2588e..6a0c41c707b11 100644 --- a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala @@ -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 { @@ -138,6 +139,63 @@ 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 when Java-serialized for persistence") { + 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 serializer = new JavaSerializer(conf).newInstance() + + val appDesc = ApplicationDescription("name", Some(4), cmd, "appUiUrl", defaultResourceProfile) + val appInfo = new ApplicationInfo( + 0, "app-1", appDesc, new Date(0), null, Int.MaxValue).withConf(conf) + val deserializedApp = serializer.deserialize[ApplicationInfo](serializer.serialize(appInfo)) + 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 driverDesc = DriverDescription("hdfs://some.jar", 100, 3, false, cmd) + val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new Date(0)).withConf(conf) + val deserializedDriver = serializer.deserialize[DriverInfo](serializer.serialize(driverInfo)) + 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()) diff --git a/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala index 6839afdeeff8e..fdef4e51800ed 100644 --- a/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala @@ -25,6 +25,8 @@ import java.util.concurrent.ThreadLocalRandom import org.apache.curator.test.TestingServer import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} +import org.apache.spark.deploy.{ApplicationDescription, Command, DriverDescription} +import org.apache.spark.deploy.DeployTestUtils.defaultResourceProfile import org.apache.spark.internal.config.Deploy.ZOOKEEPER_URL import org.apache.spark.io.CompressionCodec import org.apache.spark.rpc.{RpcEndpoint, RpcEnv} @@ -42,6 +44,55 @@ class PersistenceEngineSuite extends SparkFunSuite { } } + test("SPARK-58592: FileSystemPersistenceEngine redacts secrets in ApplicationInfo and " + + "DriverInfo") { + withTempDir { dir => + val conf = new SparkConf() + val serializer = new JavaSerializer(conf) + val engine = new FileSystemPersistenceEngine(dir.getAbsolutePath, serializer) + try { + 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 java.util.Date(0), null, Int.MaxValue).withConf(conf) + engine.addApplication(appInfo) + + val driverDesc = new DriverDescription("hdfs://some.jar", 100, 3, false, cmd) + val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new java.util.Date(0)) + .withConf(conf) + engine.addDriver(driverInfo) + + // The plaintext secrets must not be present in the bytes actually written to disk. + Seq("app_app-1", "driver_driver-1").foreach { fileName => + val bytes = Files.readAllBytes(Paths.get(dir.getAbsolutePath, fileName)) + val contents = new String(bytes, java.nio.charset.StandardCharsets.ISO_8859_1) + assert(!contents.contains("topsecret")) + assert(!contents.contains("env-token")) + } + + val recoveredApp = engine.read[ApplicationInfo]("app_").head + assert(recoveredApp.desc.command.environment("PASSWORD") == + Utils.REDACTION_REPLACEMENT_TEXT) + assert(recoveredApp.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default") + assert(!recoveredApp.desc.command.javaOpts.contains("env-token")) + assert(recoveredApp.desc.command.javaOpts.contains("-Xmx2g")) + + val recoveredDriver = engine.read[DriverInfo]("driver_").head + assert(recoveredDriver.desc.command.environment("PASSWORD") == + Utils.REDACTION_REPLACEMENT_TEXT) + assert(recoveredDriver.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default") + assert(!recoveredDriver.desc.command.javaOpts.contains("env-token")) + assert(recoveredDriver.desc.command.javaOpts.contains("-Xmx2g")) + } finally { + engine.close() + } + } + } + test("SPARK-46258: RocksDBPersistenceEngine") { withTempDir { dir => val conf = new SparkConf() From 9f3ab2c5a311ca263e316a26644a12c4e1f87589 Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Thu, 6 Aug 2026 12:04:29 +0800 Subject: [PATCH 2/3] [SPARK-58592][CORE] Redact secrets in Standalone Master RequestMasterState RPC Replace writeReplace on ApplicationInfo/DriverInfo with writeReplace on MasterStateResponse only. This redacts secrets (Command.environment, javaOpts) during cross-process RPC serialization via Utils.redact, while leaving PersistenceEngine serialization untouched - avoiding the HA recovery regression where redacted copies would be persisted and deserialized with REDACTION_REPLACEMENT_TEXT instead of real secrets. - Remove writeReplace/withConf/_conf from ApplicationInfo and DriverInfo - Keep redactedCopy on both classes - Add writeReplace/withConf to MasterStateResponse - Call withConf(conf) only in RequestMasterState handler - Revert createApplication/createDriver to not call withConf - Remove PersistenceEngineSuite test (persistence not touched) - Update JsonProtocolSuite tests for RPC serialization path Generated-by: GLM 5.2. --- .../apache/spark/deploy/DeployMessage.scala | 24 +++++++++ .../spark/deploy/master/ApplicationInfo.scala | 11 ---- .../spark/deploy/master/DriverInfo.scala | 11 ---- .../apache/spark/deploy/master/Master.scala | 6 +-- .../spark/deploy/JsonProtocolSuite.scala | 23 ++++++--- .../master/PersistenceEngineSuite.scala | 51 ------------------- 6 files changed, 42 insertions(+), 84 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala index 12e031711aa2a..d31cabb8f937f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala +++ b/core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala @@ -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 @@ -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 diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala index bff2f7b7466d2..cdd205318530f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala @@ -213,15 +213,4 @@ private[spark] class ApplicationInfo( val redactedDesc = desc.copy(command = redactedCommand) new ApplicationInfo(startTime, id, redactedDesc, submitDate, driver, defaultCores) } - - @transient private var _conf: SparkConf = _ - - private[deploy] def withConf(conf: SparkConf): this.type = { - _conf = conf - this - } - - private def writeReplace(): AnyRef = { - if (_conf == null) this else redactedCopy(_conf) - } } diff --git a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala index 5644747bd61d1..bf8a3a8cc49ea 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala @@ -66,15 +66,4 @@ private[deploy] class DriverInfo( copy.withResources(_resources) copy } - - @transient private var _conf: SparkConf = _ - - private[deploy] def withConf(conf: SparkConf): this.type = { - _conf = conf - this - } - - private def writeReplace(): AnyRef = { - if (_conf == null) this else redactedCopy(_conf) - } } diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index ca43918916fb7..8a9c3d25e040f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -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) @@ -1136,7 +1136,7 @@ private[deploy] class Master( } else { newApplicationId(date) } - new ApplicationInfo(now, appId, desc, date, driver, defaultCores).withConf(conf) + new ApplicationInfo(now, appId, desc, date, driver, defaultCores) } private[master] def registerApplication(app: ApplicationInfo): Unit = { @@ -1349,7 +1349,7 @@ private[deploy] class Master( val now = System.currentTimeMillis() val date = new Date(now) val id = newDriverId(date) - new DriverInfo(now, id, maybeUpdateAppName(desc, id), date).withConf(conf) + new DriverInfo(now, id, maybeUpdateAppName(desc, id), date) } private def launchDriver(worker: WorkerInfo, driver: DriverInfo): Unit = { diff --git a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala index 6a0c41c707b11..fa570c1d9c553 100644 --- a/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/JsonProtocolSuite.scala @@ -168,26 +168,33 @@ class JsonProtocolSuite extends SparkFunSuite with JsonTestUtils { assert(redactedDriver.desc.command.javaOpts.contains("-Xmx2g")) } - test("SPARK-58592: writeReplace redacts secrets when Java-serialized for persistence") { + 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 appDesc = ApplicationDescription("name", Some(4), cmd, "appUiUrl", defaultResourceProfile) - val appInfo = new ApplicationInfo( - 0, "app-1", appDesc, new Date(0), null, Int.MaxValue).withConf(conf) - val deserializedApp = serializer.deserialize[ApplicationInfo](serializer.serialize(appInfo)) + 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 driverDesc = DriverDescription("hdfs://some.jar", 100, 3, false, cmd) - val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new Date(0)).withConf(conf) - val deserializedDriver = serializer.deserialize[DriverInfo](serializer.serialize(driverInfo)) + val deserializedDriver = deserialized.activeDrivers.head assert(!deserializedDriver.desc.command.environment.contains("topsecret")) assert(deserializedDriver.desc.command.environment("PASSWORD") == Utils.REDACTION_REPLACEMENT_TEXT) diff --git a/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala b/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala index fdef4e51800ed..6839afdeeff8e 100644 --- a/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/master/PersistenceEngineSuite.scala @@ -25,8 +25,6 @@ import java.util.concurrent.ThreadLocalRandom import org.apache.curator.test.TestingServer import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} -import org.apache.spark.deploy.{ApplicationDescription, Command, DriverDescription} -import org.apache.spark.deploy.DeployTestUtils.defaultResourceProfile import org.apache.spark.internal.config.Deploy.ZOOKEEPER_URL import org.apache.spark.io.CompressionCodec import org.apache.spark.rpc.{RpcEndpoint, RpcEnv} @@ -44,55 +42,6 @@ class PersistenceEngineSuite extends SparkFunSuite { } } - test("SPARK-58592: FileSystemPersistenceEngine redacts secrets in ApplicationInfo and " + - "DriverInfo") { - withTempDir { dir => - val conf = new SparkConf() - val serializer = new JavaSerializer(conf) - val engine = new FileSystemPersistenceEngine(dir.getAbsolutePath, serializer) - try { - 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 java.util.Date(0), null, Int.MaxValue).withConf(conf) - engine.addApplication(appInfo) - - val driverDesc = new DriverDescription("hdfs://some.jar", 100, 3, false, cmd) - val driverInfo = new DriverInfo(0, "driver-1", driverDesc, new java.util.Date(0)) - .withConf(conf) - engine.addDriver(driverInfo) - - // The plaintext secrets must not be present in the bytes actually written to disk. - Seq("app_app-1", "driver_driver-1").foreach { fileName => - val bytes = Files.readAllBytes(Paths.get(dir.getAbsolutePath, fileName)) - val contents = new String(bytes, java.nio.charset.StandardCharsets.ISO_8859_1) - assert(!contents.contains("topsecret")) - assert(!contents.contains("env-token")) - } - - val recoveredApp = engine.read[ApplicationInfo]("app_").head - assert(recoveredApp.desc.command.environment("PASSWORD") == - Utils.REDACTION_REPLACEMENT_TEXT) - assert(recoveredApp.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default") - assert(!recoveredApp.desc.command.javaOpts.contains("env-token")) - assert(recoveredApp.desc.command.javaOpts.contains("-Xmx2g")) - - val recoveredDriver = engine.read[DriverInfo]("driver_").head - assert(recoveredDriver.desc.command.environment("PASSWORD") == - Utils.REDACTION_REPLACEMENT_TEXT) - assert(recoveredDriver.desc.command.environment("JAVA_HOME") == "/usr/lib/jvm/default") - assert(!recoveredDriver.desc.command.javaOpts.contains("env-token")) - assert(recoveredDriver.desc.command.javaOpts.contains("-Xmx2g")) - } finally { - engine.close() - } - } - } - test("SPARK-46258: RocksDBPersistenceEngine") { withTempDir { dir => val conf = new SparkConf() From 2a420a27c16431abed65f086df3c2ca0e0950ad3 Mon Sep 17 00:00:00 2001 From: Yuming Wang Date: Thu, 6 Aug 2026 12:19:12 +0800 Subject: [PATCH 3/3] [SPARK-58592][CORE] Deduplicate redaction logic in Command.redactedCopy Move the redactedCopy(conf) method down to Command itself, eliminating the same 3-line redaction pattern duplicated in ApplicationInfo, DriverInfo, and JsonProtocol.writeApplicationDescription. Generated-by: GLM 5.2. --- core/src/main/scala/org/apache/spark/deploy/Command.scala | 7 +++++++ .../main/scala/org/apache/spark/deploy/JsonProtocol.scala | 7 +------ .../org/apache/spark/deploy/master/ApplicationInfo.scala | 5 +---- .../scala/org/apache/spark/deploy/master/DriverInfo.scala | 5 +---- 4 files changed, 10 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/Command.scala b/core/src/main/scala/org/apache/spark/deploy/Command.scala index a2b263544c6a2..d53762bfe6288 100644 --- a/core/src/main/scala/org/apache/spark/deploy/Command.scala +++ b/core/src/main/scala/org/apache/spark/deploy/Command.scala @@ -19,6 +19,9 @@ 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], @@ -26,4 +29,8 @@ private[spark] case class Command( 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)) } diff --git a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala index 2a3fd0d004e11..b299a9a9afd28 100644 --- a/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala +++ b/core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala @@ -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 { @@ -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) ~ diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala index cdd205318530f..0e433b8b1f726 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala @@ -207,10 +207,7 @@ private[spark] class ApplicationInfo( } private[deploy] def redactedCopy(conf: SparkConf): ApplicationInfo = { - val redactedCommand = desc.command.copy( - environment = Utils.redact(conf, desc.command.environment.toSeq).toMap, - javaOpts = Utils.redactCommandLineArgs(conf, desc.command.javaOpts)) - val redactedDesc = desc.copy(command = redactedCommand) + val redactedDesc = desc.copy(command = desc.command.redactedCopy(conf)) new ApplicationInfo(startTime, id, redactedDesc, submitDate, driver, defaultCores) } } diff --git a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala index bf8a3a8cc49ea..9ca18597fbbe9 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/DriverInfo.scala @@ -58,10 +58,7 @@ private[deploy] class DriverInfo( def resources: Map[String, ResourceInformation] = _resources private[deploy] def redactedCopy(conf: SparkConf): DriverInfo = { - val redactedCommand = desc.command.copy( - environment = Utils.redact(conf, desc.command.environment.toSeq).toMap, - javaOpts = Utils.redactCommandLineArgs(conf, desc.command.javaOpts)) - val redactedDesc = desc.copy(command = redactedCommand) + val redactedDesc = desc.copy(command = desc.command.redactedCopy(conf)) val copy = new DriverInfo(startTime, id, redactedDesc, submitDate) copy.withResources(_resources) copy