From 76e3c3d8128b8d65d3dfecd4d6c9de9d7a36c7e2 Mon Sep 17 00:00:00 2001
From: Hyukjin Kwon <gurwls223@apache.org>
Date: Tue, 8 Sep 2026 12:46:51 +0900
Subject: [PATCH 1/2] [SPARK-59306][SS] Redact all JAAS credential styles in
 Kafka parameter logging

### What changes were proposed in this pull request?

`KafkaRedactionUtil.redactJaasParam` redacts credential values in a `sasl.jaas.config` string
before it is logged (at DEBUG). The previous regex `password="..."` only masked a double-quoted
value of an option named `password`, so a `password` value in single-quoted or unquoted form, and
the OAUTHBEARER `clientSecret` value, were not masked.

This parses each `key=value` option in the JAAS entry (escape-aware for quoted values, so an
escaped quote does not cut the value short) and redacts the value of an option in any quoting style
when its name matches either:

- the always-redacted credential options `password` and `clientSecret` (matched case-insensitively
  as a substring), which are masked unconditionally so redaction never depends on configuration; or
- the configured `spark.redaction.regex`, which widens coverage on top of the built-in options.

Because `spark.redaction.regex` replaces the default rather than extending it, consulting it alone
would be fail-open (a user pattern that omits `password` would silently stop masking it); the union
with the built-in options keeps the unconditional guarantee the old code had. Non-secret options
(the login-module class, control flag, `username`, `serviceName`, `clientId`, ...) are left intact
so the entry stays readable in logs.

### Why are the changes needed?

So Kafka JAAS credentials are not exposed in DEBUG logs regardless of how they are quoted, and so
the OAUTHBEARER `clientSecret` is masked as well as `password`.

### Does this PR introduce _any_ user-facing change?

No functional change. This only affects DEBUG log output: `password` values in single-quoted and
unquoted forms (in addition to the double-quoted form already masked), `clientSecret` values, and
values of any option whose name matches `spark.redaction.regex` are now shown as
`<option>="[redacted]"`. Since the default `spark.redaction.regex` includes `token`, a non-secret
`tokenauth` option is also masked -- a minor debuggability cost. Nothing used to connect is
affected.

### How was this patch tested?

`KafkaRedactionUtilSuite`: `password` (double-quoted, single-quoted, unquoted) and `clientSecret`
values are redacted while non-secret context (`username`, `clientId`) is preserved; an escaped
quote inside a credential value does not leak the tail; `password` is still redacted when
`spark.redaction.regex` is set to a pattern that omits it; and an option matching a configured
`spark.redaction.regex` is additionally redacted.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 4.8
---
 .../spark/kafka010/KafkaRedactionUtil.scala   | 39 +++++++++++++-
 .../kafka010/KafkaRedactionUtilSuite.scala    | 54 +++++++++++++++++++
 2 files changed, 92 insertions(+), 1 deletion(-)

diff --git a/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaRedactionUtil.scala b/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaRedactionUtil.scala
index 44c0b1cb32c99..4fd54c933de35 100644
--- a/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaRedactionUtil.scala
+++ b/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaRedactionUtil.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.kafka010
 
+import scala.util.matching.Regex
+
 import org.apache.kafka.common.config.SaslConfigs
 
 import org.apache.spark.{SparkConf, SparkEnv}
@@ -42,9 +44,44 @@ object KafkaRedactionUtil extends Logging {
     }
   }
 
+  // Matches a single `key=value` option inside a JAAS configuration entry. The value may be
+  // double-quoted, single-quoted, or unquoted (up to the next whitespace or ';'). The quoted
+  // alternatives are escape-aware -- an escaped quote (`\"`) does not terminate the value -- so the
+  // whole value is captured rather than being cut short at an embedded quote. Group 1 is the option
+  // name, group 2 the value.
+  private val jaasOptionPattern =
+    """([\w.$-]+)\s*=\s*("(?:[^"\\]|\\.)*"|'(?:[^'\\]|\\.)*'|[^\s;]+)""".r
+
+  // The JAAS credential option names that are always redacted, independent of any user
+  // configuration: the SASL `password` and the OAUTHBEARER `clientSecret`. Matched
+  // case-insensitively as a substring of the option name. Keeping this built-in guarantees the
+  // credentials are masked even when `spark.redaction.regex` is set to a pattern that does not
+  // cover them (that config replaces the default rather than extending it, so relying on it alone
+  // would be fail-open) -- mirroring the unconditional password redaction this code used to do.
+  private val alwaysRedactedOptions = "(?i)password|clientSecret".r
+
+  // Redacts the value of a JAAS credential option, regardless of quoting style (double-quoted,
+  // single-quoted, or unquoted), while leaving non-secret options (the login-module class, control
+  // flag, `username`, `serviceName`, `clientId`, ...) intact so the entry stays readable in logs.
+  // An option is redacted if its name matches the always-redacted set above OR the configured
+  // `spark.redaction.regex`, so a user-set pattern can widen coverage but never disable the
+  // built-in credential masking. Note: the default `spark.redaction.regex` includes `token`, so a
+  // non-secret `tokenauth` option is redacted too -- a minor debuggability cost accepted as safe.
   def redactJaasParam(param: String): String = {
     if (param != null && !param.isEmpty) {
-      param.replaceAll("password=\".*\"", s"""password="$REDACTION_REPLACEMENT_TEXT"""")
+      val redactionPattern = Option(SparkEnv.get).map(_.conf)
+        .getOrElse(new SparkConf()).get(SECRET_REDACTION_PATTERN)
+      jaasOptionPattern.replaceAllIn(param, m => {
+        val name = m.group(1)
+        val isSecret = alwaysRedactedOptions.findFirstMatchIn(name).isDefined ||
+          redactionPattern.findFirstMatchIn(name).isDefined
+        val replacement = if (isSecret) {
+          s"""$name="$REDACTION_REPLACEMENT_TEXT""""
+        } else {
+          m.matched
+        }
+        Regex.quoteReplacement(replacement)
+      })
     } else {
       param
     }
diff --git a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
index 225afbe5f3649..b8685184bec56 100644
--- a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
+++ b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
@@ -122,4 +122,58 @@ class KafkaRedactionUtilSuite extends SparkFunSuite with KafkaDelegationTokenTes
     assert(redactedJaasParams.contains(tokenId1))
     assert(!redactedJaasParams.contains(tokenPassword1))
   }
+
+  test("redactJaasParam should redact secret options in all quoting styles, keeping context") {
+    val prefix = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"u\" "
+    val oauth = "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required " +
+      "clientId=\"id\" "
+    // (input, secret value that must be gone, non-secret context that must survive)
+    val cases = Seq(
+      (prefix + "password=\"double-quoted-secret\";", "double-quoted-secret", "username=\"u\""),
+      (prefix + "password='single-quoted-secret';", "single-quoted-secret", "username=\"u\""),
+      (prefix + "password=unquoted-secret;", "unquoted-secret", "username=\"u\""),
+      (oauth + "clientSecret=\"oauth-secret\";", "oauth-secret", "clientId=\"id\"")
+    )
+    cases.foreach { case (param, secret, context) =>
+      val redacted = KafkaRedactionUtil.redactJaasParam(param)
+      assert(!redacted.contains(secret), s"credential not redacted in output: $redacted")
+      assert(redacted.contains(REDACTION_REPLACEMENT_TEXT), s"no redaction marker in: $redacted")
+      assert(redacted.contains(context), s"non-secret context dropped from output: $redacted")
+    }
+  }
+
+  test("redactJaasParam should not leak the tail of an escaped-quote credential value") {
+    // The value contains an escaped double quote, so the whole value must be treated as one token
+    // rather than ending at the embedded quote (which would leak the trailing characters).
+    val param = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
+      "username=\"u\" password=\"ab\\\"TAIL_SECRET\";"
+    val redacted = KafkaRedactionUtil.redactJaasParam(param)
+    assert(!redacted.contains("TAIL_SECRET"), s"escaped-quote credential tail leaked: $redacted")
+    assert(redacted.contains(REDACTION_REPLACEMENT_TEXT), s"no redaction marker in: $redacted")
+    assert(redacted.contains("username=\"u\""), s"non-secret context dropped: $redacted")
+  }
+
+  test("redactJaasParam always redacts credentials even when spark.redaction.regex omits them") {
+    // spark.redaction.regex replaces the default rather than extending it. A user pattern that does
+    // not mention `password` must not disable the built-in credential masking (fail-open guard).
+    setSparkEnv(Map(SECRET_REDACTION_PATTERN.key -> "(?i)my_custom_key"))
+    val param = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
+      "username=\"u\" password=\"PLAINTEXT_SECRET\";"
+    val redacted = KafkaRedactionUtil.redactJaasParam(param)
+    assert(!redacted.contains("PLAINTEXT_SECRET"), s"password leaked despite custom regex: $redacted")
+    assert(redacted.contains(REDACTION_REPLACEMENT_TEXT), s"no redaction marker in: $redacted")
+    assert(redacted.contains("username=\"u\""), s"non-secret context dropped: $redacted")
+  }
+
+  test("redactJaasParam additionally honors a configured spark.redaction.regex") {
+    // A user-configured pattern widens coverage on top of the always-redacted options: an option
+    // whose name matches the configured regex is redacted too, while non-matching options survive.
+    setSparkEnv(Map(SECRET_REDACTION_PATTERN.key -> "(?i)my_custom_key"))
+    val param = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
+      "username=\"u\" my_custom_key=\"CUSTOM_SECRET\";"
+    val redacted = KafkaRedactionUtil.redactJaasParam(param)
+    assert(!redacted.contains("CUSTOM_SECRET"), s"configured secret not redacted: $redacted")
+    assert(redacted.contains(REDACTION_REPLACEMENT_TEXT), s"no redaction marker in: $redacted")
+    assert(redacted.contains("username=\"u\""), s"non-secret context dropped: $redacted")
+  }
 }

From 398cc5e89e2bfe33e51566f1c925e6c07b779845 Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Wed, 9 Sep 2026 12:43:34 -0700
Subject: [PATCH 2/2] Update
 connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala

---
 .../org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala    | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
index b8685184bec56..967d2d9e08341 100644
--- a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
+++ b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala
@@ -160,7 +160,8 @@ class KafkaRedactionUtilSuite extends SparkFunSuite with KafkaDelegationTokenTes
     val param = "org.apache.kafka.common.security.plain.PlainLoginModule required " +
       "username=\"u\" password=\"PLAINTEXT_SECRET\";"
     val redacted = KafkaRedactionUtil.redactJaasParam(param)
-    assert(!redacted.contains("PLAINTEXT_SECRET"), s"password leaked despite custom regex: $redacted")
+    assert(!redacted.contains("PLAINTEXT_SECRET"),
+      s"password leaked despite custom regex: $redacted")
     assert(redacted.contains(REDACTION_REPLACEMENT_TEXT), s"no redaction marker in: $redacted")
     assert(redacted.contains("username=\"u\""), s"non-secret context dropped: $redacted")
   }