[SPARK-59306][SS] Redact all JAAS credential styles in Kafka parameter logging

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
     }

connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala

@@ -122,4 +122,59 @@ 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")
+  }
 }