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/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/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 e66933b84af55..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 @@ -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,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) + } } 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..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 @@ -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,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 + } } 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..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) 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..fa570c1d9c553 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,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())