This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 8a01334b7dc4 [SPARK-58658][3.5][CONNECT] Do not return redacted
configurations from the Config RPC
8a01334b7dc4 is described below
commit 8a01334b7dc4fff475978676ff8a9126e7c2a16a
Author: Aleksandr Efimov <[email protected]>
AuthorDate: Fri Aug 28 00:56:36 2026 -0700
[SPARK-58658][3.5][CONNECT] Do not return redacted configurations from the
Config RPC
### What changes were proposed in this pull request?
Backport of #57892 ([SPARK-58658]) to branch-3.5, requested in
https://github.com/apache/spark/pull/57892#issuecomment-5336856708.
The cherry-pick does not apply: on 3.5 the handler lives under
`connector/connect/server/...` and the file has diverged - `handle` resolves
the session inline through `SparkConnectService.getOrCreateIsolatedSession`
rather than `SessionHolder.withSession`. The change itself carries over
unchanged:
- the four read paths - `Get`, `GetWithDefault`, `GetOption`, `GetAll` -
withhold entries matching `spark.redaction.regex`, each reporting a withheld
entry the way that same operation already reports an unset key;
- the pattern is read from the `SparkConf`, not from the session config,
which the client can write;
- matching covers the key or the value, following `Utils.redact`;
- `GetAll` filters before the requested prefix is stripped;
- `IsModifiable` is unchanged.
One hunk of the original has no counterpart here. `SparkConnectAuthSuite`
asserts that `spark.connect.authenticate.token` is not readable, and 3.5 has no
Connect authentication at all, so this backport carries the handler and the new
`SparkConnectConfigHandlerSuite` only. The suite is adapted to 3.5 APIs:
`SessionHolder.forTesting` instead of
`SparkConnectTestUtils.createDummySessionHolder`,
`SparkConnectService.invalidateAllSessions()`, `getErrorClass` instead of
`getCondition`, and [...]
### Why are the changes needed?
The Config RPC discloses anything sensitive that reached the `SparkConf`,
whatever put it there, because `SQLConf.mergeSparkConf` copies every entry into
the session config and no read path applies a denylist. The full rationale,
including the Kyuubi deployment where a client reads back a ZooKeeper digest
and an internal pre-shared secret it never held, is in #57892.
### Does this PR introduce _any_ user-facing change?
Yes, the same change as #57892: 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.
### How was this patch tested?
```
build/sbt "connect/testOnly
org.apache.spark.sql.connect.service.SparkConnectConfigHandlerSuite"
```
8 tests, all passing. With the handler reverted to the branch-3.5 version,
7 of the 8 fail; the one that still passes is the `IsModifiable` case, whose
behaviour is deliberately unchanged.
### 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 #58104 from alexandrefimov/SPARK-58658-3.5.
Authored-by: Aleksandr Efimov <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../service/SparkConnectConfigHandler.scala | 76 +++++++--
.../service/SparkConnectConfigHandlerSuite.scala | 171 +++++++++++++++++++++
2 files changed, 234 insertions(+), 13 deletions(-)
diff --git
a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
b/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
index 38fd88297f35..4bf2362476ed 100644
---
a/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandler.scala
+++
b/connector/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.collection.JavaConverters._
+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])
@@ -35,17 +38,24 @@ class SparkConnectConfigHandler(responseObserver:
StreamObserver[proto.ConfigRes
.getOrCreateIsolatedSession(request.getUserContext.getUserId,
request.getSessionId)
.session
+ // 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 =
session.sparkContext.conf.get(SECRET_REDACTION_PATTERN)
val builder = request.getOperation.getOpTypeCase match {
case proto.ConfigRequest.Operation.OpTypeCase.SET =>
handleSet(request.getOperation.getSet, session.conf)
case proto.ConfigRequest.Operation.OpTypeCase.GET =>
- handleGet(request.getOperation.getGet, session.conf)
+ handleGet(request.getOperation.getGet, session.conf, redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_WITH_DEFAULT =>
- handleGetWithDefault(request.getOperation.getGetWithDefault,
session.conf)
+ handleGetWithDefault(
+ request.getOperation.getGetWithDefault,
+ session.conf,
+ redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_OPTION =>
- handleGetOption(request.getOperation.getGetOption, session.conf)
+ handleGetOption(request.getOperation.getGetOption, session.conf,
redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.GET_ALL =>
- handleGetAll(request.getOperation.getGetAll, session.conf)
+ handleGetAll(request.getOperation.getGetAll, session.conf,
redactionPattern)
case proto.ConfigRequest.Operation.OpTypeCase.UNSET =>
handleUnset(request.getOperation.getUnset, session.conf)
case proto.ConfigRequest.Operation.OpTypeCase.IS_MODIFIABLE =>
@@ -72,10 +82,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)
}
@@ -84,12 +99,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
@@ -97,10 +119,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)
}
@@ -109,15 +137,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)))
@@ -162,6 +197,21 @@ object SparkConnectConfigHandler {
private[connect] val unsupportedConfigurations =
Set("spark.sql.execution.arrow.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/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala
b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala
new file mode 100644
index 000000000000..18774b089a65
--- /dev/null
+++
b/connector/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectConfigHandlerSuite.scala
@@ -0,0 +1,171 @@
+/*
+ * 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.collection.JavaConverters._
+import scala.concurrent.Promise
+import scala.concurrent.duration._
+
+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.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.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 = SessionHolder.forTesting(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 = SessionHolder.forTesting(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 = SessionHolder.forTesting(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 = SessionHolder.forTesting(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.getErrorClass === unset.getErrorClass)
+ }
+ }
+
+ test("GetOption returns no value for a redacted key") {
+ val sessionHolder = SessionHolder.forTesting(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 = SessionHolder.forTesting(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 = SessionHolder.forTesting(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 = SessionHolder.forTesting(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]
