KAFKA-17014: ScramFormatter should not use String for password

clients/src/main/java/org/apache/kafka/common/security/scram/internals/ScramFormatter.java

@@ -23,11 +23,15 @@
 import org.apache.kafka.common.security.scram.internals.ScramMessages.ServerFirstMessage;
 
 import java.math.BigInteger;
+import java.nio.CharBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CharsetEncoder;
 import java.nio.charset.StandardCharsets;
 import java.security.InvalidKeyException;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
+import java.util.Arrays;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -86,11 +90,11 @@ public byte[] hi(byte[] str, byte[] salt, int iterations) throws InvalidKeyExcep
         return result;
     }
 
-    public static byte[] normalize(String str) {
-        return toBytes(str);
+    public static byte[] normalize(char[] chars) {
+        return toBytes(chars);
     }
 
-    public byte[] saltedPassword(String password, byte[] salt, int iterations) throws InvalidKeyException {
+    public byte[] saltedPassword(char[] password, byte[] salt, int iterations) throws InvalidKeyException {
         return hi(normalize(password), salt, iterations);
     }
 
@@ -166,11 +170,20 @@ public static byte[] secureRandomBytes(SecureRandom random) {
         return toBytes(secureRandomString(random));
     }
 
+    public static byte[] toBytes(char[] chars) {
+        final CharsetEncoder encoder = StandardCharsets.UTF_8.newEncoder();
+        try {
+            return encoder.encode(CharBuffer.wrap(chars)).array();
+        } catch (CharacterCodingException e) {
+            throw new IllegalStateException("Failed to encode " + Arrays.toString(chars), e);
+        }
+    }
+
     public static byte[] toBytes(String str) {
         return str.getBytes(StandardCharsets.UTF_8);
     }
 
-    public ScramCredential generateCredential(String password, int iterations) {
+    public ScramCredential generateCredential(char[] password, int iterations) {
         try {
             byte[] salt = secureRandomBytes();
             byte[] saltedPassword = saltedPassword(password, salt, iterations);

clients/src/main/java/org/apache/kafka/common/security/scram/internals/ScramSaslClient.java

@@ -190,7 +190,7 @@ private void setState(State state) {
 
     private ClientFinalMessage handleServerFirstMessage(char[] password) throws SaslException {
         try {
-            byte[] passwordBytes = ScramFormatter.normalize(new String(password));
+            byte[] passwordBytes = ScramFormatter.normalize(new String(password).toCharArray());
             this.saltedPassword = formatter.hi(passwordBytes, serverFirstMessage.salt(), serverFirstMessage.iterations());
 
             ClientFinalMessage clientFinalMessage = new ClientFinalMessage("n,,".getBytes(StandardCharsets.UTF_8), serverFirstMessage.nonce());

clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java

@@ -2338,7 +2338,7 @@ private void updateScramCredentialCache(String username, String password) throws
             ScramMechanism scramMechanism = ScramMechanism.forMechanismName(mechanism);
             if (scramMechanism != null) {
                 ScramFormatter formatter = new ScramFormatter(scramMechanism);
-                ScramCredential credential = formatter.generateCredential(password, 4096);
+                ScramCredential credential = formatter.generateCredential(password.toCharArray(), 4096);
                 credentialCache.cache(scramMechanism.mechanismName(), ScramCredential.class).put(username, credential);
             }
         }
@@ -2356,7 +2356,7 @@ private void updateTokenCredentialCache(String username, String password) throws
             ScramMechanism scramMechanism = ScramMechanism.forMechanismName(mechanism);
             if (scramMechanism != null) {
                 ScramFormatter formatter = new ScramFormatter(scramMechanism);
-                ScramCredential credential = formatter.generateCredential(password, 4096);
+                ScramCredential credential = formatter.generateCredential(password.toCharArray(), 4096);
                 server.tokenCache().credentialCache(scramMechanism.mechanismName()).put(username, credential);
             }
         }

clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramCredentialUtilsTest.java

@@ -45,7 +45,7 @@ public void setUp() throws NoSuchAlgorithmException {
 
     @Test
     public void stringConversion() {
-        ScramCredential credential = formatter.generateCredential("password", 1024);
+        ScramCredential credential = formatter.generateCredential("password".toCharArray(), 1024);
         assertTrue(credential.salt().length > 0, "Salt must not be empty");
         assertTrue(credential.storedKey().length > 0, "Stored key must not be empty");
         assertTrue(credential.serverKey().length > 0, "Server key must not be empty");
@@ -58,8 +58,8 @@ public void stringConversion() {
 
     @Test
     public void generateCredential() {
-        ScramCredential credential1 = formatter.generateCredential("password", 4096);
-        ScramCredential credential2 = formatter.generateCredential("password", 4096);
+        ScramCredential credential1 = formatter.generateCredential("password".toCharArray(), 4096);
+        ScramCredential credential2 = formatter.generateCredential("password".toCharArray(), 4096);
         // Random salt should ensure that the credentials persisted are different every time
         assertNotEquals(ScramCredentialUtils.credentialToString(credential1), ScramCredentialUtils.credentialToString(credential2));
     }
@@ -71,13 +71,13 @@ public void invalidCredential() {
 
     @Test
     public void missingFields() {
-        String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password", 2048));
+        String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password".toCharArray(), 2048));
         assertThrows(IllegalArgumentException.class, () -> ScramCredentialUtils.credentialFromString(cred.substring(cred.indexOf(','))));
     }
 
     @Test
     public void extraneousFields() {
-        String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password", 2048));
+        String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password".toCharArray(), 2048));
         assertThrows(IllegalArgumentException.class, () -> ScramCredentialUtils.credentialFromString(cred + ",a=test"));
     }
 
@@ -90,7 +90,7 @@ public void scramCredentialCache() throws Exception {
 
         CredentialCache.Cache<ScramCredential> sha512Cache = cache.cache(ScramMechanism.SCRAM_SHA_512.mechanismName(), ScramCredential.class);
         ScramFormatter formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_512);
-        ScramCredential credentialA = formatter.generateCredential("password", 4096);
+        ScramCredential credentialA = formatter.generateCredential("password".toCharArray(), 4096);
         sha512Cache.put("userA", credentialA);
         assertEquals(credentialA, sha512Cache.get("userA"));
         assertNull(sha512Cache.get("userB"), "Invalid user credential");

clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramFormatterTest.java

@@ -38,7 +38,7 @@ public class ScramFormatterTest {
     public void rfc7677Example() throws Exception {
         ScramFormatter formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_256);
 
-        String password = "pencil";
+        char[] password = "pencil".toCharArray();
         String c1 = "n,,n=user,r=rOprNGfwEbeRWgbNEkqO";
         String s1 = "r=rOprNGfwEbeRWgbNEkqO%hvYDpWUa2RaTCAfuxFIlj)hNlF$k0,s=W22ZaJ0SNY7soEsUEjb6gQ==,i=4096";
         String c2 = "c=biws,r=rOprNGfwEbeRWgbNEkqO%hvYDpWUa2RaTCAfuxFIlj)hNlF$k0,p=dHzbZapWIk4jUhN+Ute9ytag9zjfMHgsqmmiz7AndVQ=";

clients/src/test/java/org/apache/kafka/common/security/scram/internals/ScramSaslServerTest.java

@@ -53,8 +53,8 @@ public void setUp() throws Exception {
         ScramMechanism mechanism = ScramMechanism.SCRAM_SHA_256;
         formatter  = new ScramFormatter(mechanism);
         CredentialCache.Cache<ScramCredential> credentialCache = new CredentialCache().createCache(mechanism.mechanismName(), ScramCredential.class);
-        credentialCache.put(USER_A, formatter.generateCredential("passwordA", 4096));
-        credentialCache.put(USER_B, formatter.generateCredential("passwordB", 4096));
+        credentialCache.put(USER_A, formatter.generateCredential("passwordA".toCharArray(), 4096));
+        credentialCache.put(USER_B, formatter.generateCredential("passwordB".toCharArray(), 4096));
         ScramServerCallbackHandler callbackHandler = new ScramServerCallbackHandler(credentialCache, new DelegationTokenCache(ScramMechanism.mechanismNames()));
         saslServer = new ScramSaslServer(mechanism, new HashMap<>(), callbackHandler);
     }

core/src/main/scala/kafka/server/DelegationTokenManager.scala

@@ -106,7 +106,7 @@ class DelegationTokenManager(val config: KafkaConfig,
     val scramCredentialMap = mutable.Map[String, ScramCredential]()
 
     def scramCredential(mechanism: ScramMechanism): ScramCredential = {
-      new ScramFormatter(mechanism).generateCredential(hmacString, mechanism.minIterations)
+      new ScramFormatter(mechanism).generateCredential(hmacString.toCharArray, mechanism.minIterations)
     }
 
     for (mechanism <- ScramMechanism.values)

metadata/src/main/java/org/apache/kafka/metadata/storage/ScramParser.java

@@ -173,7 +173,9 @@ byte[] saltedPassword(byte[] salt, int iterations) throws Exception {
                 return configuredSaltedPassword.get();
             }
             return new ScramFormatter(mechanism).saltedPassword(
-                    configuredPasswordString.get(),
+                    configuredPasswordString
+                            .orElseThrow(() -> new IllegalStateException("configuredPasswordString is missing"))
+                            .toCharArray(),
                     salt,
                     iterations);
         }