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..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 @@ -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") + } }