You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/08/19 02:40:38 UTC
(spark) branch branch-4.1 updated: [SPARK-58658][CONNECT] Do not return redacted configurations from the Config RPC
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new c0574b0273f6 [SPARK-58658][CONNECT] Do not return redacted configurations from the Config RPC
c0574b0273f6 is described below
commit c0574b0273f6476123e3c1a88ce3e55ffcfd829b
Author: Aleksandr Efimov <dz...@gmail.com>
AuthorDate: Tue Aug 18 19:39:50 2026 -0700
[SPARK-58658][CONNECT] Do not return redacted configurations from the Config RPC
### What changes were proposed in this pull request?
The four read paths of the Config RPC — `Get`, `GetWithDefault`, `GetOption` and `GetAll` — stop returning configuration entries that match `spark.redaction.regex`. A withheld entry is reported the way that same operation already reports an unset key: `Get` fails with `SQL_CONF_NOT_FOUND`, `GetWithDefault` returns the caller's default, `GetOption` returns no value and `GetAll` omits the entry.
Three details are worth pointing at.
The pattern is read from the `SparkConf`, not from the session config. The session config is client-writable. `requireNonStaticConf` does guard `spark.redaction.regex`, but that guard is itself controlled by `spark.sql.legacy.setCommandRejectsSparkCoreConfs`, an ordinary non-static SQL conf, so two `Set` calls would otherwise be enough to clear the filter before reading.
Matching covers the key or the value, following `Utils.redact`. `SetCommand` already redacts through `SQLConf.redactOptions`, so a key-only match would leave the Config RPC laxer than `SET` over the same configuration — a password inside a JDBC URL is the obvious case.
Only `spark.redaction.regex` is applied, so the alignment with `SET` is partial, and partial by intent. `redactOptions` folds in `spark.sql.redaction.options.regex` as well, default `(?i)url`: `SET spark.sql.catalog.mycat.url` answers `*********(redacted)` where the Config RPC returns the URL. That pattern exists for SQL command output and its default would withhold every key with `url` in its name, so I left it out — on those keys the RPC stays the laxer of the two.
`GetAll` filters before the requested prefix is stripped. Filtering afterwards would return `spark.my.secret.value` as `value` for prefix `spark.my.secret.`, walking straight past the pattern.
`IsModifiable` is left as it is: whether a key is modifiable is a property of the name and discloses no value.
### Why are the changes needed?
The Config RPC discloses anything sensitive that reached `SparkConf`, whatever put it there. `spark.connect.authenticate.token` is the one key Spark itself puts there, and it looks benign only because it happens to be the client's own credential.
Apache Kyuubi makes the general shape visible. Kyuubi passes its own configuration to the Spark engine through `--conf` and prefixes every non-`spark.` key with `spark.`, so the whole engine configuration ends up in the driver's `SparkConf`, and `SQLConf.mergeSparkConf` copies all of it into the session config. Some of those entries are secrets shared across the deployment rather than owned by the connecting user — a ZooKeeper digest, an internal pre-shared secret. That is a plain cli [...]
### Does this PR introduce _any_ user-facing change?
Yes. A configuration entry whose key or value matches `spark.redaction.regex` is no longer returned by the Config RPC, and each read operation reports it as if it were unset.
Three consequences I would like reviewers to look at explicitly.
A client can no longer read back an entry it set itself in its own session: after `spark.conf.set("spark.sql.catalog.mycat.password", …)`, reading that key back now fails. Classic Spark still returns it. The filter could be narrowed to keys present in `sparkContext.conf`, which would keep session-local entries readable; I did not do that, because it would let a secret written into the session config server-side — by a plugin or an extension rather than through `SparkConf` — stay reada [...]
A batched `Get` fails as a whole if any key in the batch is withheld, since that is how a single unset key in a batch already behaves. With the default pattern none of the keys the clients batch internally match, but an operator who widens `spark.redaction.regex` far enough could turn `createDataFrame` into an error rather than a hidden value.
An empty `spark.redaction.regex` matches every key, so setting it to an empty string — a plausible attempt to turn redaction off — now withholds the entire configuration. `Utils.redact` degrades the same way, so I did not special-case it here, but it is worth a decision.
### How was this patch tested?
New `SparkConnectConfigHandlerSuite` covers the five read operations, the prefix-stripping order, the value match, and that the pattern is not taken from the session config. `SparkConnectAuthSuite` gains an end-to-end assertion that the authentication token is not readable while a non-sensitive server-side configuration still is.
```
build/sbt "connect/testOnly org.apache.spark.sql.connect.service.SparkConnectConfigHandlerSuite org.apache.spark.sql.connect.service.SparkConnectAuthSuite"
```
10 tests, all passing. With the handler reverted to master, 8 of the 10 fail; the two that still pass are the `IsModifiable` case, whose behaviour is deliberately unchanged, and the pre-existing authentication test.
### Was this patch authored or co-authored using generative AI tooling?
Yes — Claude Code (Opus 5) was used for code reading, drafting and review. Every line was reviewed by the author, who takes responsibility for the patch.
Closes #57892 from alexandrefimov/SPARK-58658-config-rpc-redaction.
Authored-by: Aleksandr Efimov <dz...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit cd687e6259bbbeb4ab3ea35f31d96e3883c9b39f)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../service/SparkConnectConfigHandler.scala | 73 +++++++--
.../connect/service/SparkConnectAuthSuite.scala | 22 ++-
.../service/SparkConnectConfigHandlerSuite.scala | 172 +++++++++++++++++++++
3 files changed, 253 insertions(+), 14 deletions(-)
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
index 06bc24b6ccae..73eba3a6f476 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
@@ -18,12 +18,15 @@
package org.apache.spark.sql.connect.service
import scala.jdk.CollectionConverters._
+import scala.util.matching.Regex
import io.grpc.stub.StreamObserver
import org.apache.spark.connect.proto
import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.SECRET_REDACTION_PATTERN
import org.apache.spark.sql.RuntimeConfig
+import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.internal.SQLConf
class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigResponse])
@@ -44,18 +47,22 @@ class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigRes
}
private def doHandle(r: proto.ConfigRequest, h: SessionHolder): Unit = h.withSession { s =>
+ // Read the pattern from the SparkConf rather than from the session config. The session config
+ // is writable by the client, so taking the pattern from there would let a client widen its own
+ // view of the configuration before reading it back.
+ val redactionPattern = s.sparkContext.conf.get(SECRET_REDACTION_PATTERN)
// Make sure we're using the current running session.
val builder = r.getOperation.getOpTypeCase match {
case proto.ConfigRequest.Operation.OpTypeCase.SET =>
handleSet(r.getOperation.getSet, s.conf)
case proto.ConfigRequest.Operation.OpTypeCase.GET =>
- handleGet(r.getOperation.getGet, s.conf)
+ handleGet(r.getOperation.getGet, s.conf, redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_WITH_DEFAULT =>
- handleGetWithDefault(r.getOperation.getGetWithDefault, s.conf)
+ handleGetWithDefault(r.getOperation.getGetWithDefault, s.conf, redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_OPTION =>
- handleGetOption(r.getOperation.getGetOption, s.conf)
+ handleGetOption(r.getOperation.getGetOption, s.conf, redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_ALL =>
- handleGetAll(r.getOperation.getGetAll, s.conf)
+ handleGetAll(r.getOperation.getGetAll, s.conf, redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.UNSET =>
handleUnset(r.getOperation.getUnset, s.conf)
case proto.ConfigRequest.Operation.OpTypeCase.IS_MODIFIABLE =>
@@ -94,10 +101,15 @@ class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigRes
private def handleGet(
operation: proto.ConfigRequest.Get,
- conf: RuntimeConfig): proto.ConfigResponse.Builder = {
+ conf: RuntimeConfig,
+ redactionPattern: Regex): proto.ConfigResponse.Builder = {
val builder = proto.ConfigResponse.newBuilder()
operation.getKeysList.asScala.iterator.foreach { key =>
val value = conf.get(key)
+ if (SparkConnectConfigHandler.isRedacted(key, Option(value), redactionPattern)) {
+ // This operation reports an unset key by failing, so a redacted key fails the same way.
+ throw QueryExecutionErrors.sqlConfigNotFoundError(key)
+ }
builder.addPairs(SparkConnectConfigHandler.toProtoKeyValue(key, Option(value)))
getWarning(key).foreach(builder.addWarnings)
}
@@ -106,12 +118,19 @@ class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigRes
private def handleGetWithDefault(
operation: proto.ConfigRequest.GetWithDefault,
- conf: RuntimeConfig): proto.ConfigResponse.Builder = {
+ conf: RuntimeConfig,
+ redactionPattern: Regex): proto.ConfigResponse.Builder = {
val builder = proto.ConfigResponse.newBuilder()
operation.getPairsList.asScala.iterator.foreach { pair =>
val (key, default) = SparkConnectConfigHandler.toKeyValue(pair)
- val value = conf.get(key, default.orNull)
- builder.addPairs(SparkConnectConfigHandler.toProtoKeyValue(key, Option(value)))
+ val stored = Option(conf.get(key, default.orNull))
+ // A redacted entry falls back to the caller's default, as an unset key does.
+ val value = if (SparkConnectConfigHandler.isRedacted(key, stored, redactionPattern)) {
+ default
+ } else {
+ stored
+ }
+ builder.addPairs(SparkConnectConfigHandler.toProtoKeyValue(key, value))
getWarning(key).foreach(builder.addWarnings)
}
builder
@@ -119,10 +138,16 @@ class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigRes
private def handleGetOption(
operation: proto.ConfigRequest.GetOption,
- conf: RuntimeConfig): proto.ConfigResponse.Builder = {
+ conf: RuntimeConfig,
+ redactionPattern: Regex): proto.ConfigResponse.Builder = {
val builder = proto.ConfigResponse.newBuilder()
operation.getKeysList.asScala.iterator.foreach { key =>
- val value = conf.getOption(key)
+ val stored = conf.getOption(key)
+ val value = if (SparkConnectConfigHandler.isRedacted(key, stored, redactionPattern)) {
+ None
+ } else {
+ stored
+ }
builder.addPairs(SparkConnectConfigHandler.toProtoKeyValue(key, value))
getWarning(key).foreach(builder.addWarnings)
}
@@ -131,15 +156,22 @@ class SparkConnectConfigHandler(responseObserver: StreamObserver[proto.ConfigRes
private def handleGetAll(
operation: proto.ConfigRequest.GetAll,
- conf: RuntimeConfig): proto.ConfigResponse.Builder = {
+ conf: RuntimeConfig,
+ redactionPattern: Regex): proto.ConfigResponse.Builder = {
val builder = proto.ConfigResponse.newBuilder()
+ // Drop redacted entries before the prefix is stripped below. Matching afterwards would let a
+ // GetAll with prefix `spark.my.secret.` return `spark.my.secret.value` as `value`, which no
+ // longer matches the pattern.
+ val visible = conf.getAll.iterator.filterNot { case (key, value) =>
+ SparkConnectConfigHandler.isRedacted(key, Option(value), redactionPattern)
+ }
val results = if (operation.hasPrefix) {
val prefix = operation.getPrefix
- conf.getAll.iterator
+ visible
.filter { case (key, _) => key.startsWith(prefix) }
.map { case (key, value) => (key.substring(prefix.length), value) }
} else {
- conf.getAll.iterator
+ visible
}
results.foreach { case (key, value) =>
builder.addPairs(SparkConnectConfigHandler.toProtoKeyValue(key, Option(value)))
@@ -185,6 +217,21 @@ object SparkConnectConfigHandler {
private[connect] val unsupportedConfigurations =
Set("spark.sql.execution.arrow.enabled", "spark.sql.execution.arrow.pyspark.fallback.enabled")
+ /**
+ * Whether a configuration entry is considered sensitive and must not be disclosed by the Config
+ * RPC. Follows `spark.redaction.regex` and matches the key or the value, the way `Utils.redact`
+ * does for the environment UI and event logs and `SetCommand` does for `SET`. Matching the
+ * value is what covers a secret carried by an innocuous key, such as a password inside a JDBC
+ * URL.
+ */
+ private[connect] def isRedacted(
+ key: String,
+ value: Option[String],
+ redactionPattern: Regex): Boolean = {
+ redactionPattern.findFirstIn(key).isDefined ||
+ value.exists(redactionPattern.findFirstIn(_).isDefined)
+ }
+
def toKeyValue(pair: proto.KeyValue): (String, Option[String]) = {
val key = pair.getKey
val value = if (pair.hasValue) {
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthSuite.scala
index 30f186ab7c2b..93c910c19aca 100644
--- a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthSuite.scala
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectAuthSuite.scala
@@ -23,8 +23,13 @@ import org.apache.spark.sql.connect.{SparkConnectServerTest, SparkSession}
import org.apache.spark.sql.connect.service.SparkConnectService
class SparkConnectAuthSuite extends SparkConnectServerTest {
+ private val tokenKey = "spark.connect.authenticate.token"
+ private val markerKey = "spark.connect.test.marker"
+
override protected def sparkConf = {
- super.sparkConf.set("spark.connect.authenticate.token", "deadbeef")
+ super.sparkConf
+ .set(tokenKey, "deadbeef")
+ .set(markerKey, "visible")
}
test("Test local authentication") {
@@ -43,4 +48,19 @@ class SparkConnectAuthSuite extends SparkConnectServerTest {
}
assert(exception.getMessage.contains("Invalid authentication token"))
}
+
+ test("Test the authentication token is not readable through the Config RPC") {
+ val session = SparkSession
+ .builder()
+ .remote(s"sc://localhost:${SparkConnectService.localPort}/;token=deadbeef")
+ .create()
+
+ assert(session.conf.getOption(tokenKey).isEmpty)
+ assert(session.conf.get(tokenKey, "absent") === "absent")
+ intercept[NoSuchElementException](session.conf.get(tokenKey))
+ assert(!session.conf.getAll.contains(tokenKey))
+
+ // Server-side configurations that are not sensitive stay readable.
+ assert(session.conf.get(markerKey) === "visible")
+ }
}
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala
new file mode 100644
index 000000000000..269f7836506b
--- /dev/null
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.service
+
+import scala.concurrent.Promise
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+import io.grpc.stub.StreamObserver
+
+import org.apache.spark.SparkNoSuchElementException
+import org.apache.spark.connect.proto
+import org.apache.spark.internal.config.SECRET_REDACTION_PATTERN
+import org.apache.spark.sql.connect.SparkConnectTestUtils
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.ThreadUtils
+
+class SparkConnectConfigHandlerSuite extends SharedSparkSession {
+
+ // Matches the default spark.redaction.regex on "password".
+ private val secretKey = "spark.test.connect.password"
+ private val secretValue = "hunter2"
+ private val plainKey = "spark.test.connect.endpoint"
+ private val plainValue = "localhost:15002"
+
+ protected override def afterEach(): Unit = {
+ super.afterEach()
+ SparkConnectService.sessionManager.invalidateAllSessions()
+ }
+
+ private def sendConfigRequest(
+ sessionHolder: SessionHolder,
+ customize: proto.ConfigRequest.Operation.Builder => Unit): proto.ConfigResponse = {
+ val operation = proto.ConfigRequest.Operation.newBuilder()
+ customize(operation)
+ val request = proto.ConfigRequest
+ .newBuilder()
+ .setUserContext(proto.UserContext.newBuilder().setUserId(sessionHolder.userId).build())
+ .setSessionId(sessionHolder.sessionId)
+ .setOperation(operation)
+ .build()
+ val responseObserver = new ConfigResponseObserver()
+ new SparkConnectConfigHandler(responseObserver).handle(request)
+ ThreadUtils.awaitResult(responseObserver.promise.future, 10.seconds)
+ }
+
+ /** The returned pairs, with an absent value mapped to None. */
+ private def pairs(response: proto.ConfigResponse): Map[String, Option[String]] = {
+ response.getPairsList.asScala
+ .map(pair => pair.getKey -> (if (pair.hasValue) Some(pair.getValue) else None))
+ .toMap
+ }
+
+ test("GetAll does not return keys matching the redaction pattern") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ withSQLConf(secretKey -> secretValue, plainKey -> plainValue) {
+ val returned = pairs(sendConfigRequest(sessionHolder, _.getGetAllBuilder))
+ assert(!returned.contains(secretKey))
+ assert(returned(plainKey) === Some(plainValue))
+ }
+ }
+
+ test("GetAll matches the redaction pattern on the full key, not the prefixed one") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ // Stripping the prefix leaves "value", which no longer matches the pattern. The filter has to
+ // run before the prefix comes off.
+ val prefix = "spark.test.connect.secret."
+ withSQLConf(prefix + "value" -> secretValue) {
+ val returned = pairs(sendConfigRequest(sessionHolder, _.getGetAllBuilder.setPrefix(prefix)))
+ assert(returned.isEmpty)
+ }
+ }
+
+ test("a secret carried by an innocuous key is withheld too") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ // `SET spark.test.connect.jdbc.url` already masks this value, because SetCommand redacts
+ // through Utils.redact, which matches the value as well. The Config RPC has to agree.
+ val urlKey = "spark.test.connect.jdbc.url"
+ val urlValue = "jdbc:postgresql://db:5432/app?user=app&password=hunter2"
+ withSQLConf(urlKey -> urlValue) {
+ assert(!pairs(sendConfigRequest(sessionHolder, _.getGetAllBuilder)).contains(urlKey))
+ val returned =
+ pairs(sendConfigRequest(sessionHolder, _.getGetOptionBuilder.addKeys(urlKey)))
+ assert(returned(urlKey) === None)
+ }
+ }
+
+ test("Get reports a redacted key the way it reports an unset one") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ withSQLConf(secretKey -> secretValue) {
+ val redacted = intercept[SparkNoSuchElementException] {
+ sendConfigRequest(sessionHolder, _.getGetBuilder.addKeys(secretKey))
+ }
+ val unset = intercept[SparkNoSuchElementException] {
+ sendConfigRequest(sessionHolder, _.getGetBuilder.addKeys("spark.test.connect.absent"))
+ }
+ assert(redacted.getCondition === unset.getCondition)
+ }
+ }
+
+ test("GetOption returns no value for a redacted key") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ withSQLConf(secretKey -> secretValue, plainKey -> plainValue) {
+ val returned = pairs(
+ sendConfigRequest(
+ sessionHolder,
+ _.getGetOptionBuilder.addKeys(secretKey).addKeys(plainKey)))
+ assert(returned(secretKey) === None)
+ assert(returned(plainKey) === Some(plainValue))
+ }
+ }
+
+ test("GetWithDefault returns the caller's default for a redacted key") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ withSQLConf(secretKey -> secretValue) {
+ val returned = pairs(
+ sendConfigRequest(
+ sessionHolder,
+ _.getGetWithDefaultBuilder.addPairsBuilder().setKey(secretKey).setValue("fallback")))
+ assert(returned(secretKey) === Some("fallback"))
+ }
+ }
+
+ test("the redaction pattern is not read from the session config") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ // Both of these are reachable through the Set operation: the legacy flag is what otherwise
+ // stops a client from writing spark.redaction.regex into its own session config.
+ withSQLConf(
+ SQLConf.SET_COMMAND_REJECTS_SPARK_CORE_CONFS.key -> "false",
+ secretKey -> secretValue) {
+ spark.conf.set(SECRET_REDACTION_PATTERN.key, "matches-no-key")
+ try {
+ val returned = pairs(sendConfigRequest(sessionHolder, _.getGetAllBuilder))
+ assert(!returned.contains(secretKey))
+ } finally {
+ spark.conf.unset(SECRET_REDACTION_PATTERN.key)
+ }
+ }
+ }
+
+ test("IsModifiable still answers for a redacted key") {
+ val sessionHolder = SparkConnectTestUtils.createDummySessionHolder(spark)
+ withSQLConf(secretKey -> secretValue) {
+ // Whether a key is modifiable is a property of the key, so it discloses no value.
+ val returned =
+ pairs(sendConfigRequest(sessionHolder, _.getIsModifiableBuilder.addKeys(secretKey)))
+ assert(returned(secretKey) === Some("false"))
+ }
+ }
+}
+
+private class ConfigResponseObserver extends StreamObserver[proto.ConfigResponse] {
+ val promise: Promise[proto.ConfigResponse] = Promise()
+ override def onNext(value: proto.ConfigResponse): Unit = promise.success(value)
+ override def onError(t: Throwable): Unit = promise.failure(t)
+ override def onCompleted(): Unit = {}
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org