This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 61fbccb2a61b [SPARK-58658][CONNECT] Do not return redacted
configurations from the Config RPC
61fbccb2a61b is described below
commit 61fbccb2a61b0a008037e81c6b743a1f77a20bae
Author: Aleksandr Efimov <[email protected]>
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 <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit cd687e6259bbbeb4ab3ea35f31d96e3883c9b39f)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
