You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2017/01/10 13:05:15 UTC
[1/2] kafka git commit: KAFKA-3751; SASL/SCRAM implementation
Repository: kafka
Updated Branches:
refs/heads/trunk 1fea1c390 -> 275c5e1df
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java
index cc4befa..0bff0ee 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java
@@ -27,6 +27,9 @@ import java.util.Map;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.security.authenticator.CredentialCache;
+import org.apache.kafka.common.security.scram.ScramCredentialUtils;
+import org.apache.kafka.common.security.scram.ScramMechanism;
import org.apache.kafka.common.utils.MockTime;
/**
@@ -42,6 +45,7 @@ public class NioEchoServer extends Thread {
private final AcceptorThread acceptorThread;
private final Selector selector;
private volatile WritableByteChannel outputChannel;
+ private final CredentialCache credentialCache;
public NioEchoServer(SecurityProtocol securityProtocol, Map<String, ?> configs, String serverHost) throws Exception {
serverSocketChannel = ServerSocketChannel.open();
@@ -50,7 +54,10 @@ public class NioEchoServer extends Thread {
this.port = serverSocketChannel.socket().getLocalPort();
this.socketChannels = Collections.synchronizedList(new ArrayList<SocketChannel>());
this.newChannels = Collections.synchronizedList(new ArrayList<SocketChannel>());
- ChannelBuilder channelBuilder = ChannelBuilders.create(securityProtocol, Mode.SERVER, LoginType.SERVER, configs, null, true);
+ this.credentialCache = new CredentialCache();
+ if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL)
+ ScramCredentialUtils.createCache(credentialCache, ScramMechanism.mechanismNames());
+ ChannelBuilder channelBuilder = ChannelBuilders.serverChannelBuilder(securityProtocol, configs, credentialCache);
this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", channelBuilder);
setName("echoserver");
setDaemon(true);
@@ -61,6 +68,10 @@ public class NioEchoServer extends Thread {
return port;
}
+ public CredentialCache credentialCache() {
+ return credentialCache;
+ }
+
@Override
public void run() {
try {
@@ -162,4 +173,4 @@ public class NioEchoServer extends Thread {
}
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index f75d5b7..f8b57f4 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -19,7 +19,6 @@ import org.apache.kafka.common.network.CertStores;
import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.ChannelBuilders;
import org.apache.kafka.common.network.LoginType;
-import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.network.NetworkSend;
import org.apache.kafka.common.network.NetworkTestUtils;
import org.apache.kafka.common.network.NioEchoServer;
@@ -40,6 +39,10 @@ import org.apache.kafka.common.requests.SaslHandshakeRequest;
import org.apache.kafka.common.requests.SaslHandshakeResponse;
import org.apache.kafka.common.security.JaasUtils;
import org.apache.kafka.common.security.plain.PlainLoginModule;
+import org.apache.kafka.common.security.scram.ScramCredential;
+import org.apache.kafka.common.security.scram.ScramFormatter;
+import org.apache.kafka.common.security.scram.ScramLoginModule;
+import org.apache.kafka.common.security.scram.ScramMechanism;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -47,6 +50,8 @@ import org.junit.Test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -209,8 +214,9 @@ public class SaslAuthenticatorTest {
@Test
public void testMultipleServerMechanisms() throws Exception {
SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
- configureMechanisms("DIGEST-MD5", Arrays.asList("DIGEST-MD5", "PLAIN"));
+ configureMechanisms("DIGEST-MD5", Arrays.asList("DIGEST-MD5", "PLAIN", "SCRAM-SHA-256"));
server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
String node1 = "1";
saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
@@ -222,6 +228,125 @@ public class SaslAuthenticatorTest {
InetSocketAddress addr = new InetSocketAddress("127.0.0.1", server.port());
selector.connect(node2, addr, BUFFER_SIZE, BUFFER_SIZE);
NetworkTestUtils.checkClientConnection(selector, node2, 100, 10);
+ selector.close();
+
+ String node3 = "3";
+ saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
+ createSelector(securityProtocol, saslClientConfigs);
+ selector.connect(node3, new InetSocketAddress("127.0.0.1", server.port()), BUFFER_SIZE, BUFFER_SIZE);
+ NetworkTestUtils.checkClientConnection(selector, node3, 100, 10);
+ selector.close();
+ selector = null;
+ }
+
+ /**
+ * Tests good path SASL/SCRAM-SHA-256 client and server channels.
+ */
+ @Test
+ public void testValidSaslScramSha256() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ configureMechanisms("SCRAM-SHA-256", Arrays.asList("SCRAM-SHA-256"));
+
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
+ createAndCheckClientConnection(securityProtocol, "0");
+ }
+
+ /**
+ * Tests all supported SCRAM client and server channels. Also tests that all
+ * supported SCRAM mechanisms can be supported simultaneously on a server.
+ */
+ @Test
+ public void testValidSaslScramMechanisms() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ configureMechanisms("SCRAM-SHA-256", new ArrayList<>(ScramMechanism.mechanismNames()));
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
+
+ for (String mechanism : ScramMechanism.mechanismNames()) {
+ saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, mechanism);
+ createAndCheckClientConnection(securityProtocol, "node-" + mechanism);
+ }
+ }
+
+ /**
+ * Tests that SASL/SCRAM clients fail authentication if password is invalid.
+ */
+ @Test
+ public void testInvalidPasswordSaslScram() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Arrays.asList("SCRAM-SHA-256"));
+ Map<String, Object> options = new HashMap<>();
+ options.put("username", TestJaasConfig.USERNAME);
+ options.put("password", "invalidpassword");
+ jaasConfig.createOrUpdateEntry(JaasUtils.LOGIN_CONTEXT_CLIENT, ScramLoginModule.class.getName(), options);
+
+ String node = "0";
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
+ createClientConnection(securityProtocol, node);
+ NetworkTestUtils.waitForChannelClose(selector, node);
+ }
+
+ /**
+ * Tests that SASL/SCRAM clients without valid username fail authentication.
+ */
+ @Test
+ public void testUnknownUserSaslScram() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Arrays.asList("SCRAM-SHA-256"));
+ Map<String, Object> options = new HashMap<>();
+ options.put("username", "unknownUser");
+ options.put("password", TestJaasConfig.PASSWORD);
+ jaasConfig.createOrUpdateEntry(JaasUtils.LOGIN_CONTEXT_CLIENT, ScramLoginModule.class.getName(), options);
+
+ String node = "0";
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
+ createClientConnection(securityProtocol, node);
+ NetworkTestUtils.waitForChannelClose(selector, node);
+ }
+
+ /**
+ * Tests that SASL/SCRAM clients fail authentication if credentials are not available for
+ * the specific SCRAM mechanism.
+ */
+ @Test
+ public void testUserCredentialsUnavailableForScramMechanism() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ configureMechanisms("SCRAM-SHA-256", new ArrayList<>(ScramMechanism.mechanismNames()));
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(TestJaasConfig.USERNAME, TestJaasConfig.PASSWORD);
+
+ server.credentialCache().cache(ScramMechanism.SCRAM_SHA_256.mechanismName(), ScramCredential.class).remove(TestJaasConfig.USERNAME);
+ String node = "1";
+ saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
+ createClientConnection(securityProtocol, node);
+ NetworkTestUtils.waitForChannelClose(selector, node);
+ selector.close();
+
+ saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
+ createAndCheckClientConnection(securityProtocol, "2");
+ }
+
+ /**
+ * Tests SASL/SCRAM with username containing characters that need
+ * to be encoded.
+ */
+ @Test
+ public void testScramUsernameWithSpecialCharacters() throws Exception {
+ SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
+ String username = "special user= test,scram";
+ String password = username + "-password";
+ TestJaasConfig jaasConfig = configureMechanisms("SCRAM-SHA-256", Arrays.asList("SCRAM-SHA-256"));
+ Map<String, Object> options = new HashMap<>();
+ options.put("username", username);
+ options.put("password", password);
+ jaasConfig.createOrUpdateEntry(JaasUtils.LOGIN_CONTEXT_CLIENT, ScramLoginModule.class.getName(), options);
+
+ server = NetworkTestUtils.createEchoServer(securityProtocol, saslServerConfigs);
+ updateScramCredentialCache(username, password);
+ createAndCheckClientConnection(securityProtocol, "0");
}
/**
@@ -606,7 +731,7 @@ public class SaslAuthenticatorTest {
private void createSelector(SecurityProtocol securityProtocol, Map<String, Object> clientConfigs) {
String saslMechanism = (String) saslClientConfigs.get(SaslConfigs.SASL_MECHANISM);
- this.channelBuilder = ChannelBuilders.create(securityProtocol, Mode.CLIENT, LoginType.CLIENT, clientConfigs, saslMechanism, true);
+ this.channelBuilder = ChannelBuilders.clientChannelBuilder(securityProtocol, LoginType.CLIENT, clientConfigs, saslMechanism, true);
this.selector = NetworkTestUtils.createSelector(channelBuilder);
}
@@ -660,4 +785,16 @@ public class SaslAuthenticatorTest {
assertEquals(1, selector.completedReceives().size());
return selector.completedReceives().get(0).payload();
}
+
+ @SuppressWarnings("unchecked")
+ private void updateScramCredentialCache(String username, String password) throws NoSuchAlgorithmException {
+ for (String mechanism : (List<String>) saslServerConfigs.get(SaslConfigs.SASL_ENABLED_MECHANISMS)) {
+ ScramMechanism scramMechanism = ScramMechanism.forMechanismName(mechanism);
+ if (scramMechanism != null) {
+ ScramFormatter formatter = new ScramFormatter(scramMechanism);
+ ScramCredential credential = formatter.generateCredential(password, 4096);
+ server.credentialCache().cache(scramMechanism.mechanismName(), ScramCredential.class).put(username, credential);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
index 22a3267..a27b87a 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/TestJaasConfig.java
@@ -12,6 +12,7 @@
*/
package org.apache.kafka.common.security.authenticator;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -23,6 +24,8 @@ import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.security.JaasUtils;
import org.apache.kafka.common.security.plain.PlainLoginModule;
+import org.apache.kafka.common.security.scram.ScramLoginModule;
+import org.apache.kafka.common.security.scram.ScramMechanism;
public class TestJaasConfig extends Configuration {
@@ -35,7 +38,7 @@ public class TestJaasConfig extends Configuration {
TestJaasConfig config = new TestJaasConfig();
config.createOrUpdateEntry(JaasUtils.LOGIN_CONTEXT_CLIENT, loginModule(clientMechanism), defaultClientOptions());
for (String mechanism : serverMechanisms) {
- config.createOrUpdateEntry(JaasUtils.LOGIN_CONTEXT_SERVER, loginModule(mechanism), defaultServerOptions());
+ config.addEntry(JaasUtils.LOGIN_CONTEXT_SERVER, loginModule(mechanism), defaultServerOptions(mechanism));
}
Configuration.setConfiguration(config);
return config;
@@ -59,6 +62,14 @@ public class TestJaasConfig extends Configuration {
entryMap.put(name, new AppConfigurationEntry[] {entry});
}
+ public void addEntry(String name, String loginModule, Map<String, Object> options) {
+ AppConfigurationEntry entry = new AppConfigurationEntry(loginModule, LoginModuleControlFlag.REQUIRED, options);
+ AppConfigurationEntry[] existing = entryMap.get(name);
+ AppConfigurationEntry[] newEntries = existing == null ? new AppConfigurationEntry[1] : Arrays.copyOf(existing, existing.length + 1);
+ newEntries[newEntries.length - 1] = entry;
+ entryMap.put(name, newEntries);
+ }
+
@Override
public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
return entryMap.get(name);
@@ -74,7 +85,10 @@ public class TestJaasConfig extends Configuration {
loginModule = TestDigestLoginModule.class.getName();
break;
default:
- throw new IllegalArgumentException("Unsupported mechanism " + mechanism);
+ if (ScramMechanism.isScram(mechanism))
+ loginModule = ScramLoginModule.class.getName();
+ else
+ throw new IllegalArgumentException("Unsupported mechanism " + mechanism);
}
return loginModule;
}
@@ -86,9 +100,17 @@ public class TestJaasConfig extends Configuration {
return options;
}
- public static Map<String, Object> defaultServerOptions() {
+ public static Map<String, Object> defaultServerOptions(String mechanism) {
Map<String, Object> options = new HashMap<>();
- options.put("user_" + USERNAME, PASSWORD);
+ switch (mechanism) {
+ case "PLAIN":
+ case "DIGEST-MD5":
+ options.put("user_" + USERNAME, PASSWORD);
+ break;
+ default:
+ if (!ScramMechanism.isScram(mechanism))
+ throw new IllegalArgumentException("Unsupported mechanism " + mechanism);
+ }
return options;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/security/scram/ScramCredentialUtilsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/ScramCredentialUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramCredentialUtilsTest.java
new file mode 100644
index 0000000..9acc16e
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramCredentialUtilsTest.java
@@ -0,0 +1,97 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.kafka.common.security.authenticator.CredentialCache;
+import org.junit.Before;
+
+public class ScramCredentialUtilsTest {
+
+ private ScramFormatter formatter;
+
+ @Before
+ public void setUp() throws NoSuchAlgorithmException {
+ formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_256);
+ }
+
+ @Test
+ public void stringConversion() {
+ ScramCredential credential = formatter.generateCredential("password", 1024);
+ assertTrue("Salt must not be empty", credential.salt().length > 0);
+ assertTrue("Stored key must not be empty", credential.storedKey().length > 0);
+ assertTrue("Server key must not be empty", credential.serverKey().length > 0);
+ ScramCredential credential2 = ScramCredentialUtils.credentialFromString(ScramCredentialUtils.credentialToString(credential));
+ assertArrayEquals(credential.salt(), credential2.salt());
+ assertArrayEquals(credential.storedKey(), credential2.storedKey());
+ assertArrayEquals(credential.serverKey(), credential2.serverKey());
+ assertEquals(credential.iterations(), credential2.iterations());
+ }
+
+ @Test
+ public void generateCredential() {
+ ScramCredential credential1 = formatter.generateCredential("password", 4096);
+ ScramCredential credential2 = formatter.generateCredential("password", 4096);
+ // Random salt should ensure that the credentials persisted are different every time
+ assertNotEquals(ScramCredentialUtils.credentialToString(credential1), ScramCredentialUtils.credentialToString(credential2));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void invalidCredential() {
+ ScramCredentialUtils.credentialFromString("abc");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void missingFields() {
+ String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password", 2048));
+ ScramCredentialUtils.credentialFromString(cred.substring(cred.indexOf(',')));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void extraneousFields() {
+ String cred = ScramCredentialUtils.credentialToString(formatter.generateCredential("password", 2048));
+ ScramCredentialUtils.credentialFromString(cred + ",a=test");
+ }
+
+ @Test
+ public void scramCredentialCache() throws Exception {
+ CredentialCache cache = new CredentialCache();
+ ScramCredentialUtils.createCache(cache, Arrays.asList("SCRAM-SHA-512", "PLAIN"));
+ assertNotNull("Cache not created for enabled mechanism", cache.cache(ScramMechanism.SCRAM_SHA_512.mechanismName(), ScramCredential.class));
+ assertNull("Cache created for disabled mechanism", cache.cache(ScramMechanism.SCRAM_SHA_256.mechanismName(), ScramCredential.class));
+
+ 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);
+ sha512Cache.put("userA", credentialA);
+ assertEquals(credentialA, sha512Cache.get("userA"));
+ assertNull("Invalid user credential", sha512Cache.get("userB"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/security/scram/ScramFormatterTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/ScramFormatterTest.java b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramFormatterTest.java
new file mode 100644
index 0000000..bb88f8f
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramFormatterTest.java
@@ -0,0 +1,94 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import org.junit.Test;
+
+import javax.xml.bind.DatatypeConverter;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFirstMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFirstMessage;
+
+public class ScramFormatterTest {
+
+ /**
+ * Tests that the formatter implementation produces the same values for the
+ * example included in <a href="https://tools.ietf.org/html/rfc5802#section-5">RFC 7677</a>
+ */
+ @Test
+ public void rfc7677Example() throws Exception {
+ ScramFormatter formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_256);
+
+ String password = "pencil";
+ 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=";
+ String s2 = "v=6rriTRBi23WpRR/wtup+mMhUZUn/dB5nLTJRsjl95G4=";
+ ClientFirstMessage clientFirst = new ClientFirstMessage(formatter.toBytes(c1));
+ ServerFirstMessage serverFirst = new ServerFirstMessage(formatter.toBytes(s1));
+ ClientFinalMessage clientFinal = new ClientFinalMessage(formatter.toBytes(c2));
+ ServerFinalMessage serverFinal = new ServerFinalMessage(formatter.toBytes(s2));
+
+ String username = clientFirst.saslName();
+ assertEquals("user", username);
+ String clientNonce = clientFirst.nonce();
+ assertEquals("rOprNGfwEbeRWgbNEkqO", clientNonce);
+ String serverNonce = serverFirst.nonce().substring(clientNonce.length());
+ assertEquals("%hvYDpWUa2RaTCAfuxFIlj)hNlF$k0", serverNonce);
+ byte[] salt = serverFirst.salt();
+ assertArrayEquals(DatatypeConverter.parseBase64Binary("W22ZaJ0SNY7soEsUEjb6gQ=="), salt);
+ int iterations = serverFirst.iterations();
+ assertEquals(4096, iterations);
+ byte[] channelBinding = clientFinal.channelBinding();
+ assertArrayEquals(DatatypeConverter.parseBase64Binary("biws"), channelBinding);
+ byte[] serverSignature = serverFinal.serverSignature();
+ assertArrayEquals(DatatypeConverter.parseBase64Binary("6rriTRBi23WpRR/wtup+mMhUZUn/dB5nLTJRsjl95G4="), serverSignature);
+
+ byte[] saltedPassword = formatter.saltedPassword(password, salt, iterations);
+ byte[] serverKey = formatter.serverKey(saltedPassword);
+ byte[] computedProof = formatter.clientProof(saltedPassword, clientFirst, serverFirst, clientFinal);
+ assertArrayEquals(clientFinal.proof(), computedProof);
+ byte[] computedSignature = formatter.serverSignature(serverKey, clientFirst, serverFirst, clientFinal);
+ assertArrayEquals(serverFinal.serverSignature(), computedSignature);
+
+ // Minimum iterations defined in RFC-7677
+ assertEquals(4096, ScramMechanism.SCRAM_SHA_256.minIterations());
+ }
+
+ /**
+ * Tests encoding of username
+ */
+ @Test
+ public void saslName() throws Exception {
+ String[] usernames = {"user1", "123", "1,2", "user=A", "user==B", "user,1", "user 1", ",", "=", ",=", "=="};
+ ScramFormatter formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_256);
+ for (String username : usernames) {
+ String saslName = formatter.saslName(username);
+ // There should be no commas in saslName (comma is used as field separator in SASL messages)
+ assertEquals(-1, saslName.indexOf(','));
+ // There should be no "=" in the saslName apart from those used in encoding (comma is =2C and equals is =3D)
+ assertEquals(-1, saslName.replace("=2C", "").replace("=3D", "").indexOf('='));
+ assertEquals(username, formatter.username(saslName));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/test/java/org/apache/kafka/common/security/scram/ScramMessagesTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/ScramMessagesTest.java b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramMessagesTest.java
new file mode 100644
index 0000000..dddc4d3
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramMessagesTest.java
@@ -0,0 +1,348 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+
+import javax.security.sasl.SaslException;
+import javax.xml.bind.DatatypeConverter;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import org.apache.kafka.common.security.scram.ScramMessages.AbstractScramMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFirstMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFirstMessage;
+
+public class ScramMessagesTest {
+
+ private static final String[] VALID_EXTENSIONS = {
+ "ext=val1",
+ "anotherext=name1=value1 name2=another test value \"\'!$[]()",
+ "first=val1,second=name1 = value ,third=123"
+ };
+ private static final String[] INVALID_EXTENSIONS = {
+ "ext1=value",
+ "ext",
+ "ext=value1,value2",
+ "ext=,",
+ "ext =value"
+ };
+
+ private static final String[] VALID_RESERVED = {
+ "m=reserved-value",
+ "m=name1=value1 name2=another test value \"\'!$[]()"
+ };
+ private static final String[] INVALID_RESERVED = {
+ "m",
+ "m=name,value",
+ "m=,"
+ };
+
+ private ScramFormatter formatter;
+
+ @Before
+ public void setUp() throws Exception {
+ formatter = new ScramFormatter(ScramMechanism.SCRAM_SHA_256);
+ }
+
+ @Test
+ public void validClientFirstMessage() throws SaslException {
+ String nonce = formatter.secureRandomString();
+ ClientFirstMessage m = new ClientFirstMessage("someuser", nonce);
+ checkClientFirstMessage(m, "someuser", nonce, "");
+
+ // Default format used by Kafka client: only user and nonce are specified
+ String str = String.format("n,,n=testuser,r=%s", nonce);
+ m = createScramMessage(ClientFirstMessage.class, str);
+ checkClientFirstMessage(m, "testuser", nonce, "");
+ m = new ClientFirstMessage(m.toBytes());
+ checkClientFirstMessage(m, "testuser", nonce, "");
+
+ // Username containing comma, encoded as =2C
+ str = String.format("n,,n=test=2Cuser,r=%s", nonce);
+ m = createScramMessage(ClientFirstMessage.class, str);
+ checkClientFirstMessage(m, "test=2Cuser", nonce, "");
+ assertEquals("test,user", formatter.username(m.saslName()));
+
+ // Username containing equals, encoded as =3D
+ str = String.format("n,,n=test=3Duser,r=%s", nonce);
+ m = createScramMessage(ClientFirstMessage.class, str);
+ checkClientFirstMessage(m, "test=3Duser", nonce, "");
+ assertEquals("test=user", formatter.username(m.saslName()));
+
+ // Optional authorization id specified
+ str = String.format("n,a=testauthzid,n=testuser,r=%s", nonce);
+ checkClientFirstMessage(createScramMessage(ClientFirstMessage.class, str), "testuser", nonce, "testauthzid");
+
+ // Optional reserved value specified
+ for (String reserved : VALID_RESERVED) {
+ str = String.format("n,,%s,n=testuser,r=%s", reserved, nonce);
+ checkClientFirstMessage(createScramMessage(ClientFirstMessage.class, str), "testuser", nonce, "");
+ }
+
+ // Optional extension specified
+ for (String extension : VALID_EXTENSIONS) {
+ str = String.format("n,,n=testuser,r=%s,%s", nonce, extension);
+ checkClientFirstMessage(createScramMessage(ClientFirstMessage.class, str), "testuser", nonce, "");
+ }
+ }
+
+ @Test
+ public void invalidClientFirstMessage() throws SaslException {
+ String nonce = formatter.secureRandomString();
+ // Invalid entry in gs2-header
+ String invalid = String.format("n,x=something,n=testuser,r=%s", nonce);
+ checkInvalidScramMessage(ClientFirstMessage.class, invalid);
+
+ // Invalid reserved entry
+ for (String reserved : INVALID_RESERVED) {
+ invalid = String.format("n,,%s,n=testuser,r=%s", reserved, nonce);
+ checkInvalidScramMessage(ClientFirstMessage.class, invalid);
+ }
+
+ // Invalid extension
+ for (String extension : INVALID_EXTENSIONS) {
+ invalid = String.format("n,,n=testuser,r=%s,%s", nonce, extension);
+ checkInvalidScramMessage(ClientFirstMessage.class, invalid);
+ }
+ }
+
+ @Test
+ public void validServerFirstMessage() throws SaslException {
+ String clientNonce = formatter.secureRandomString();
+ String serverNonce = formatter.secureRandomString();
+ String nonce = clientNonce + serverNonce;
+ String salt = randomBytesAsString();
+
+ ServerFirstMessage m = new ServerFirstMessage(clientNonce, serverNonce, toBytes(salt), 8192);
+ checkServerFirstMessage(m, nonce, salt, 8192);
+
+ // Default format used by Kafka clients, only nonce, salt and iterations are specified
+ String str = String.format("r=%s,s=%s,i=4096", nonce, salt);
+ m = createScramMessage(ServerFirstMessage.class, str);
+ checkServerFirstMessage(m, nonce, salt, 4096);
+ m = new ServerFirstMessage(m.toBytes());
+ checkServerFirstMessage(m, nonce, salt, 4096);
+
+ // Optional reserved value
+ for (String reserved : VALID_RESERVED) {
+ str = String.format("%s,r=%s,s=%s,i=4096", reserved, nonce, salt);
+ checkServerFirstMessage(createScramMessage(ServerFirstMessage.class, str), nonce, salt, 4096);
+ }
+
+ // Optional extension
+ for (String extension : VALID_EXTENSIONS) {
+ str = String.format("r=%s,s=%s,i=4096,%s", nonce, salt, extension);
+ checkServerFirstMessage(createScramMessage(ServerFirstMessage.class, str), nonce, salt, 4096);
+ }
+ }
+
+ @Test
+ public void invalidServerFirstMessage() throws SaslException {
+ String nonce = formatter.secureRandomString();
+ String salt = randomBytesAsString();
+
+ // Invalid iterations
+ String invalid = String.format("r=%s,s=%s,i=0", nonce, salt);
+ checkInvalidScramMessage(ServerFirstMessage.class, invalid);
+
+ // Invalid salt
+ invalid = String.format("r=%s,s=%s,i=4096", nonce, "=123");
+ checkInvalidScramMessage(ServerFirstMessage.class, invalid);
+
+ // Invalid format
+ invalid = String.format("r=%s,invalid,s=%s,i=4096", nonce, salt);
+ checkInvalidScramMessage(ServerFirstMessage.class, invalid);
+
+ // Invalid reserved entry
+ for (String reserved : INVALID_RESERVED) {
+ invalid = String.format("%s,r=%s,s=%s,i=4096", reserved, nonce, salt);
+ checkInvalidScramMessage(ServerFirstMessage.class, invalid);
+ }
+
+ // Invalid extension
+ for (String extension : INVALID_EXTENSIONS) {
+ invalid = String.format("r=%s,s=%s,i=4096,%s", nonce, salt, extension);
+ checkInvalidScramMessage(ServerFirstMessage.class, invalid);
+ }
+ }
+
+ @Test
+ public void validClientFinalMessage() throws SaslException {
+ String nonce = formatter.secureRandomString();
+ String channelBinding = randomBytesAsString();
+ String proof = randomBytesAsString();
+
+ ClientFinalMessage m = new ClientFinalMessage(toBytes(channelBinding), nonce);
+ assertNull("Invalid proof", m.proof());
+ m.proof(toBytes(proof));
+ checkClientFinalMessage(m, channelBinding, nonce, proof);
+
+ // Default format used by Kafka client: channel-binding, nonce and proof are specified
+ String str = String.format("c=%s,r=%s,p=%s", channelBinding, nonce, proof);
+ m = createScramMessage(ClientFinalMessage.class, str);
+ checkClientFinalMessage(m, channelBinding, nonce, proof);
+ m = new ClientFinalMessage(m.toBytes());
+ checkClientFinalMessage(m, channelBinding, nonce, proof);
+
+ // Optional extension specified
+ for (String extension : VALID_EXTENSIONS) {
+ str = String.format("c=%s,r=%s,%s,p=%s", channelBinding, nonce, extension, proof);
+ checkClientFinalMessage(createScramMessage(ClientFinalMessage.class, str), channelBinding, nonce, proof);
+ }
+ }
+
+ @Test
+ public void invalidClientFinalMessage() throws SaslException {
+ String nonce = formatter.secureRandomString();
+ String channelBinding = randomBytesAsString();
+ String proof = randomBytesAsString();
+
+ // Invalid channel binding
+ String invalid = String.format("c=ab,r=%s,p=%s", nonce, proof);
+ checkInvalidScramMessage(ClientFirstMessage.class, invalid);
+
+ // Invalid proof
+ invalid = String.format("c=%s,r=%s,p=123", channelBinding, nonce);
+ checkInvalidScramMessage(ClientFirstMessage.class, invalid);
+
+ // Invalid extensions
+ for (String extension : INVALID_EXTENSIONS) {
+ invalid = String.format("c=%s,r=%s,%s,p=%s", channelBinding, nonce, extension, proof);
+ checkInvalidScramMessage(ClientFinalMessage.class, invalid);
+ }
+ }
+
+ @Test
+ public void validServerFinalMessage() throws SaslException {
+ String serverSignature = randomBytesAsString();
+
+ ServerFinalMessage m = new ServerFinalMessage("unknown-user", null);
+ checkServerFinalMessage(m, "unknown-user", null);
+ m = new ServerFinalMessage(null, toBytes(serverSignature));
+ checkServerFinalMessage(m, null, serverSignature);
+
+ // Default format used by Kafka clients for successful final message
+ String str = String.format("v=%s", serverSignature);
+ m = createScramMessage(ServerFinalMessage.class, str);
+ checkServerFinalMessage(m, null, serverSignature);
+ m = new ServerFinalMessage(m.toBytes());
+ checkServerFinalMessage(m, null, serverSignature);
+
+ // Default format used by Kafka clients for final message with error
+ str = String.format("e=other-error", serverSignature);
+ m = createScramMessage(ServerFinalMessage.class, str);
+ checkServerFinalMessage(m, "other-error", null);
+ m = new ServerFinalMessage(m.toBytes());
+ checkServerFinalMessage(m, "other-error", null);
+
+ // Optional extension
+ for (String extension : VALID_EXTENSIONS) {
+ str = String.format("v=%s,%s", serverSignature, extension);
+ checkServerFinalMessage(createScramMessage(ServerFinalMessage.class, str), null, serverSignature);
+ }
+ }
+
+ @Test
+ public void invalidServerFinalMessage() throws SaslException {
+ String serverSignature = randomBytesAsString();
+
+ // Invalid error
+ String invalid = "e=error1,error2";
+ checkInvalidScramMessage(ServerFinalMessage.class, invalid);
+
+ // Invalid server signature
+ invalid = String.format("v=1=23");
+ checkInvalidScramMessage(ServerFinalMessage.class, invalid);
+
+ // Invalid extensions
+ for (String extension : INVALID_EXTENSIONS) {
+ invalid = String.format("v=%s,%s", serverSignature, extension);
+ checkInvalidScramMessage(ServerFinalMessage.class, invalid);
+
+ invalid = String.format("e=unknown-user,%s", extension);
+ checkInvalidScramMessage(ServerFinalMessage.class, invalid);
+ }
+ }
+
+ private String randomBytesAsString() {
+ return DatatypeConverter.printBase64Binary(formatter.secureRandomBytes());
+ }
+
+ private byte[] toBytes(String base64Str) {
+ return DatatypeConverter.parseBase64Binary(base64Str);
+ };
+
+ private void checkClientFirstMessage(ClientFirstMessage message, String saslName, String nonce, String authzid) {
+ assertEquals(saslName, message.saslName());
+ assertEquals(nonce, message.nonce());
+ assertEquals(authzid, message.authorizationId());
+ }
+
+ private void checkServerFirstMessage(ServerFirstMessage message, String nonce, String salt, int iterations) {
+ assertEquals(nonce, message.nonce());
+ assertArrayEquals(DatatypeConverter.parseBase64Binary(salt), message.salt());
+ assertEquals(iterations, message.iterations());
+ }
+
+ private void checkClientFinalMessage(ClientFinalMessage message, String channelBinding, String nonce, String proof) {
+ assertArrayEquals(DatatypeConverter.parseBase64Binary(channelBinding), message.channelBinding());
+ assertEquals(nonce, message.nonce());
+ assertArrayEquals(DatatypeConverter.parseBase64Binary(proof), message.proof());
+ }
+
+ private void checkServerFinalMessage(ServerFinalMessage message, String error, String serverSignature) {
+ assertEquals(error, message.error());
+ if (serverSignature == null)
+ assertNull("Unexpected server signature", message.serverSignature());
+ else
+ assertArrayEquals(DatatypeConverter.parseBase64Binary(serverSignature), message.serverSignature());
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T extends AbstractScramMessage> T createScramMessage(Class<T> clazz, String message) throws SaslException {
+ byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
+ if (clazz == ClientFirstMessage.class)
+ return (T) new ClientFirstMessage(bytes);
+ else if (clazz == ServerFirstMessage.class)
+ return (T) new ServerFirstMessage(bytes);
+ else if (clazz == ClientFinalMessage.class)
+ return (T) new ClientFinalMessage(bytes);
+ else if (clazz == ServerFinalMessage.class)
+ return (T) new ServerFinalMessage(bytes);
+ else
+ throw new IllegalArgumentException("Unknown message type: " + clazz);
+ }
+
+ private <T extends AbstractScramMessage> void checkInvalidScramMessage(Class<T> clazz, String message) {
+ try {
+ createScramMessage(clazz, message);
+ fail("Exception not throws for invalid message of type " + clazz + " : " + message);
+ } catch (SaslException e) {
+ // Expected exception
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/admin/AdminUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminUtils.scala b/core/src/main/scala/kafka/admin/AdminUtils.scala
index e95d327..91cd426 100644
--- a/core/src/main/scala/kafka/admin/AdminUtils.scala
+++ b/core/src/main/scala/kafka/admin/AdminUtils.scala
@@ -20,7 +20,7 @@ package kafka.admin
import kafka.common._
import kafka.cluster.Broker
import kafka.log.LogConfig
-import kafka.server.{DynamicConfig, ConfigType}
+import kafka.server.{DynamicConfig, ConfigEntityName, ConfigType}
import kafka.utils._
import kafka.utils.ZkUtils._
import java.util.Random
@@ -504,7 +504,10 @@ object AdminUtils extends Logging with AdminUtilities {
*
*/
def changeUserOrUserClientIdConfig(zkUtils: ZkUtils, sanitizedEntityName: String, configs: Properties) {
- DynamicConfig.Client.validate(configs)
+ if (sanitizedEntityName == ConfigEntityName.Default || sanitizedEntityName.contains("/clients"))
+ DynamicConfig.Client.validate(configs)
+ else
+ DynamicConfig.User.validate(configs)
changeEntityConfig(zkUtils, ConfigType.User, sanitizedEntityName, configs)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/admin/ConfigCommand.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index a6176a2..3985490 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -25,6 +25,7 @@ import kafka.log.LogConfig
import kafka.server.{ConfigEntityName, ConfigType, DynamicConfig, QuotaId}
import kafka.utils.{CommandLineUtils, ZkUtils}
import org.apache.kafka.common.security.JaasUtils
+import org.apache.kafka.common.security.scram._
import org.apache.kafka.common.utils.Utils
import scala.collection._
import scala.collection.JavaConverters._
@@ -46,6 +47,8 @@ import scala.collection.JavaConverters._
*/
object ConfigCommand extends Config {
+ val DefaultScramIterations = 4096
+
def main(args: Array[String]): Unit = {
val opts = new ConfigCommandOptions(args)
@@ -81,6 +84,9 @@ object ConfigCommand extends Config {
val entityType = entity.root.entityType
val entityName = entity.fullSanitizedName
+ if (entityType == ConfigType.User)
+ preProcessScramCredentials(configsToBeAdded)
+
// compile the final set of configs
val configs = utils.fetchEntityConfig(zkUtils, entityType, entityName)
@@ -102,6 +108,27 @@ object ConfigCommand extends Config {
println(s"Completed Updating config for entity: $entity.")
}
+ private def preProcessScramCredentials(configsToBeAdded: Properties) {
+ def scramCredential(mechanism: ScramMechanism, credentialStr: String): String = {
+ val pattern = "(?:iterations=([0-9]*),)?password=(.*)".r
+ val (iterations, password) = credentialStr match {
+ case pattern(iterations, password) => (if (iterations != null) iterations.toInt else DefaultScramIterations, password)
+ case _ => throw new IllegalArgumentException(s"Invalid credential property $mechanism=$credentialStr")
+ }
+ if (iterations < mechanism.minIterations())
+ throw new IllegalArgumentException(s"Iterations $iterations is less than the minimum ${mechanism.minIterations()} required for $mechanism")
+ val credential = new ScramFormatter(mechanism).generateCredential(password, iterations)
+ ScramCredentialUtils.credentialToString(credential)
+ }
+ for (mechanism <- ScramMechanism.values) {
+ configsToBeAdded.getProperty(mechanism.mechanismName) match {
+ case null =>
+ case value =>
+ configsToBeAdded.setProperty(mechanism.mechanismName, scramCredential(mechanism, value))
+ }
+ }
+ }
+
private def parseBroker(broker: String): Int = {
try broker.toInt
catch {
@@ -128,9 +155,10 @@ object ConfigCommand extends Config {
val props = new Properties
if (opts.options.has(opts.addConfig)) {
//split by commas, but avoid those in [], then into KV pairs
+ val pattern = "(?=[^\\]]*(?:\\[|$))"
val configsToBeAdded = opts.options.valueOf(opts.addConfig)
- .split(",(?=[^\\]]*(?:\\[|$))")
- .map(_.split("""\s*=\s*"""))
+ .split("," + pattern)
+ .map(_.split("""\s*=\s*""" + pattern))
require(configsToBeAdded.forall(config => config.length == 2), "Invalid entity config: all configs to be added must be in the format \"key=val\".")
//Create properties, parsing square brackets from values if necessary
configsToBeAdded.foreach(pair => props.setProperty(pair(0).trim, pair(1).replaceAll("\\[?\\]?", "").trim))
@@ -283,7 +311,7 @@ object ConfigCommand extends Config {
val addConfig = parser.accepts("add-config", "Key Value pairs of configs to add. Square brackets can be used to group values which contain commas: 'k1=v1,k2=[v1,v2,v2],k3=v3'. The following is a list of valid configurations: " +
"For entity_type '" + ConfigType.Topic + "': " + LogConfig.configNames.map("\t" + _).mkString(nl, nl, nl) +
"For entity_type '" + ConfigType.Broker + "': " + DynamicConfig.Broker.names.asScala.map("\t" + _).mkString(nl, nl, nl) +
- "For entity_type '" + ConfigType.User + "': " + DynamicConfig.Client.names.asScala.map("\t" + _).mkString(nl, nl, nl) +
+ "For entity_type '" + ConfigType.User + "': " + DynamicConfig.User.names.asScala.map("\t" + _).mkString(nl, nl, nl) +
"For entity_type '" + ConfigType.Client + "': " + DynamicConfig.Client.names.asScala.map("\t" + _).mkString(nl, nl, nl) +
s"Entity types '${ConfigType.User}' and '${ConfigType.Client}' may be specified together to update config for clients of a specific user.")
.withRequiredArg
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index a820171..96b0e46 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -26,7 +26,7 @@ import kafka.server.KafkaConfig
import kafka.utils._
import org.apache.kafka.clients.{ClientRequest, ClientResponse, ManualMetadataUpdater, NetworkClient}
import org.apache.kafka.common.metrics.Metrics
-import org.apache.kafka.common.network.{ChannelBuilders, LoginType, Mode, NetworkReceive, Selectable, Selector}
+import org.apache.kafka.common.network.{ChannelBuilders, LoginType, NetworkReceive, Selectable, Selector}
import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol}
import org.apache.kafka.common.requests
import org.apache.kafka.common.requests.{UpdateMetadataRequest, _}
@@ -90,9 +90,8 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
val brokerEndPoint = broker.getBrokerEndPoint(config.interBrokerSecurityProtocol)
val brokerNode = new Node(broker.id, brokerEndPoint.host, brokerEndPoint.port)
val networkClient = {
- val channelBuilder = ChannelBuilders.create(
+ val channelBuilder = ChannelBuilders.clientChannelBuilder(
config.interBrokerSecurityProtocol,
- Mode.CLIENT,
LoginType.SERVER,
config.values,
config.saslMechanismInterBrokerProtocol,
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index 95c5fdf..d8d0144 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -29,11 +29,12 @@ import com.yammer.metrics.core.Gauge
import kafka.cluster.{BrokerEndPoint, EndPoint}
import kafka.common.KafkaException
import kafka.metrics.KafkaMetricsGroup
+import kafka.security.CredentialProvider
import kafka.server.KafkaConfig
import kafka.utils._
import org.apache.kafka.common.errors.InvalidRequestException
import org.apache.kafka.common.metrics._
-import org.apache.kafka.common.network.{ChannelBuilders, KafkaChannel, LoginType, Mode, Selectable, Selector => KSelector}
+import org.apache.kafka.common.network.{ChannelBuilders, KafkaChannel, Selectable, Selector => KSelector}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.protocol.SecurityProtocol
import org.apache.kafka.common.protocol.types.SchemaException
@@ -49,7 +50,7 @@ import scala.util.control.{ControlThrowable, NonFatal}
* Acceptor has N Processor threads that each have their own selector and read requests from sockets
* M Handler threads that handle requests and produce responses back to the processor threads for writing.
*/
-class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time) extends Logging with KafkaMetricsGroup {
+class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time, val credentialProvider: CredentialProvider) extends Logging with KafkaMetricsGroup {
private val endpoints = config.listeners
private val numProcessorThreads = config.numNetworkThreads
@@ -147,7 +148,8 @@ class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time
config.connectionsMaxIdleMs,
protocol,
config.values,
- metrics
+ metrics,
+ credentialProvider
)
}
@@ -373,7 +375,8 @@ private[kafka] class Processor(val id: Int,
connectionsMaxIdleMs: Long,
protocol: SecurityProtocol,
channelConfigs: java.util.Map[String, _],
- metrics: Metrics) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
+ metrics: Metrics,
+ credentialProvider: CredentialProvider) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
private object ConnectionId {
def fromString(s: String): Option[ConnectionId] = s.split("-") match {
@@ -411,7 +414,7 @@ private[kafka] class Processor(val id: Int,
"socket-server",
metricTags,
false,
- ChannelBuilders.create(protocol, Mode.SERVER, LoginType.SERVER, channelConfigs, null, true))
+ ChannelBuilders.serverChannelBuilder(protocol, channelConfigs, credentialProvider.credentialCache))
override def run() {
startupComplete()
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/security/CredentialProvider.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/security/CredentialProvider.scala b/core/src/main/scala/kafka/security/CredentialProvider.scala
new file mode 100644
index 0000000..5d9d7ba
--- /dev/null
+++ b/core/src/main/scala/kafka/security/CredentialProvider.scala
@@ -0,0 +1,52 @@
+/**
+ * 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 kafka.security
+
+import java.util.{List, Properties}
+
+import org.apache.kafka.common.security.authenticator.CredentialCache
+import org.apache.kafka.common.security.scram.{ScramCredential, ScramCredentialUtils, ScramMechanism}
+import org.apache.kafka.common.config.ConfigDef
+import org.apache.kafka.common.config.ConfigDef._
+
+class CredentialProvider(saslEnabledMechanisms: List[String]) {
+
+ val credentialCache = new CredentialCache
+ ScramCredentialUtils.createCache(credentialCache, saslEnabledMechanisms)
+
+ def updateCredentials(username: String, config: Properties) {
+ for (mechanism <- ScramMechanism.values()) {
+ val cache = credentialCache.cache(mechanism.mechanismName, classOf[ScramCredential])
+ if (cache != null) {
+ config.getProperty(mechanism.mechanismName) match {
+ case null => cache.remove(username)
+ case c => cache.put(username, ScramCredentialUtils.credentialFromString(c))
+ }
+ }
+ }
+ }
+}
+
+object CredentialProvider {
+ def userCredentialConfigs: ConfigDef = {
+ ScramMechanism.values.foldLeft(new ConfigDef) {
+ (c, m) => c.define(m.mechanismName, Type.STRING, null, Importance.MEDIUM, s"User credentials for SCRAM mechanism ${m.mechanismName}")
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/server/ConfigHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 7f204d5..56d53ff 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -22,6 +22,7 @@ import java.util.Properties
import DynamicConfig.Broker._
import kafka.api.ApiVersion
import kafka.log.{LogConfig, LogManager}
+import kafka.security.CredentialProvider
import kafka.server.Constants._
import kafka.server.QuotaFactory.QuotaManagers
import kafka.utils.Logging
@@ -30,6 +31,7 @@ import org.apache.kafka.common.config.ConfigException
import org.apache.kafka.common.metrics.Quota
import org.apache.kafka.common.metrics.Quota._
import scala.collection.JavaConverters._
+
/**
* The ConfigHandler is used to process config change notifications received by the DynamicConfigManager
*/
@@ -134,7 +136,7 @@ class ClientIdConfigHandler(private val quotaManagers: QuotaManagers) extends Qu
* The callback provides the node name containing sanitized user principal, client-id if this is
* a <user, client-id> update and the full properties set read from ZK.
*/
-class UserConfigHandler(private val quotaManagers: QuotaManagers) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
+class UserConfigHandler(private val quotaManagers: QuotaManagers, val credentialProvider: CredentialProvider) extends QuotaConfigHandler(quotaManagers) with ConfigHandler {
def processConfigChanges(quotaEntityPath: String, config: Properties) {
// Entity path is <user> or <user>/clients/<client>
@@ -144,6 +146,8 @@ class UserConfigHandler(private val quotaManagers: QuotaManagers) extends QuotaC
val sanitizedUser = entities(0)
val clientId = if (entities.length == 3) Some(entities(2)) else None
updateQuotaConfig(Some(sanitizedUser), clientId, config)
+ if (!clientId.isDefined && sanitizedUser != ConfigEntityName.Default)
+ credentialProvider.updateCredentials(QuotaId.desanitize(sanitizedUser), config)
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/server/DynamicConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index 4a9d0a9..e68f921 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -19,6 +19,7 @@ package kafka.server
import java.util.Properties
import kafka.log.LogConfig
+import kafka.security.CredentialProvider
import org.apache.kafka.common.config.ConfigDef
import org.apache.kafka.common.config.ConfigDef.Importance._
import org.apache.kafka.common.config.ConfigDef.Range._
@@ -81,6 +82,18 @@ object DynamicConfig {
def validate(props: Properties) = DynamicConfig.validate(clientConfigs, props)
}
+ object User {
+
+ //Definitions
+ private val userConfigs = CredentialProvider.userCredentialConfigs
+ .define(Client.ProducerByteRateOverrideProp, LONG, Client.DefaultProducerOverride, MEDIUM, Client.ProducerOverrideDoc)
+ .define(Client.ConsumerByteRateOverrideProp, LONG, Client.DefaultConsumerOverride, MEDIUM, Client.ConsumerOverrideDoc)
+
+ def names = userConfigs.names
+
+ def validate(props: Properties) = DynamicConfig.validate(userConfigs, props)
+ }
+
private def validate(configDef: ConfigDef, props: Properties) = {
//Validate Names
val names = configDef.names()
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index ef72847..df40c64 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -33,6 +33,7 @@ import kafka.coordinator.GroupCoordinator
import kafka.log.{CleanerConfig, LogConfig, LogManager}
import kafka.metrics.{KafkaMetricsGroup, KafkaMetricsReporter}
import kafka.network.{BlockingChannel, SocketServer}
+import kafka.security.CredentialProvider
import kafka.security.auth.Authorizer
import kafka.utils._
import org.I0Itec.zkclient.ZkClient
@@ -117,6 +118,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
var dynamicConfigHandlers: Map[String, ConfigHandler] = null
var dynamicConfigManager: DynamicConfigManager = null
+ var credentialProvider: CredentialProvider = null
var groupCoordinator: GroupCoordinator = null
@@ -202,8 +204,9 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
this.logIdent = "[Kafka Server " + config.brokerId + "], "
metadataCache = new MetadataCache(config.brokerId)
+ credentialProvider = new CredentialProvider(config.saslEnabledMechanisms)
- socketServer = new SocketServer(config, metrics, time)
+ socketServer = new SocketServer(config, metrics, time, credentialProvider)
socketServer.startup()
/* start replica manager */
@@ -242,7 +245,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
/* start dynamic config manager */
dynamicConfigHandlers = Map[String, ConfigHandler](ConfigType.Topic -> new TopicConfigHandler(logManager, config, quotaManagers),
ConfigType.Client -> new ClientIdConfigHandler(quotaManagers),
- ConfigType.User -> new UserConfigHandler(quotaManagers),
+ ConfigType.User -> new UserConfigHandler(quotaManagers, credentialProvider),
ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers))
// Create the config manager. start listening to notifications
@@ -351,9 +354,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
def networkClientControlledShutdown(retries: Int): Boolean = {
val metadataUpdater = new ManualMetadataUpdater()
val networkClient = {
- val channelBuilder = ChannelBuilders.create(
+ val channelBuilder = ChannelBuilders.clientChannelBuilder(
config.interBrokerSecurityProtocol,
- Mode.CLIENT,
LoginType.SERVER,
config.values,
config.saslMechanismInterBrokerProtocol,
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
index 3811be3..485a25e 100644
--- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -76,9 +76,8 @@ class ReplicaFetcherThread(name: String,
// as the metrics tag to avoid metric name conflicts with
// more than one fetcher thread to the same broker
private val networkClient = {
- val channelBuilder = ChannelBuilders.create(
+ val channelBuilder = ChannelBuilders.clientChannelBuilder(
brokerConfig.interBrokerSecurityProtocol,
- Mode.CLIENT,
LoginType.SERVER,
brokerConfig.values,
brokerConfig.saslMechanismInterBrokerProtocol,
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
index 479e749..4d86040 100644
--- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
@@ -60,7 +60,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
override val consumerCount = 2
override val serverCount = 3
- override def setAclsBeforeServersStart() {
+ override def configureSecurityBeforeServersStart() {
AclCommand.main(clusterAclArgs)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
new file mode 100644
index 0000000..b483884
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
@@ -0,0 +1,49 @@
+/**
+ * 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 kafka.api
+
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.apache.kafka.common.security.scram.ScramMechanism
+import kafka.utils.JaasTestUtils
+import kafka.admin.ConfigCommand
+import kafka.utils.ZkUtils
+import scala.collection.JavaConverters._
+
+class SaslScramSslEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
+ override protected def securityProtocol = SecurityProtocol.SASL_SSL
+ override protected def kafkaClientSaslMechanism = "SCRAM-SHA-256"
+ override protected def kafkaServerSaslMechanisms = ScramMechanism.mechanismNames.asScala.toList
+ override val clientPrincipal = JaasTestUtils.KafkaScramUser
+ override val kafkaPrincipal = JaasTestUtils.KafkaScramAdmin
+ private val clientPassword = JaasTestUtils.KafkaScramPassword
+ private val kafkaPassword = JaasTestUtils.KafkaScramAdminPassword
+
+ override def configureSecurityBeforeServersStart() {
+ super.configureSecurityBeforeServersStart()
+ zkUtils.makeSurePersistentPathExists(ZkUtils.EntityConfigChangesPath)
+
+ def configCommandArgs(username: String, password: String) : Array[String] = {
+ val credentials = kafkaServerSaslMechanisms.map(m => s"$m=[iterations=4096,password=$password]")
+ Array("--zookeeper", zkConnect,
+ "--alter", "--add-config", credentials.mkString(","),
+ "--entity-type", "users",
+ "--entity-name", username)
+ }
+ ConfigCommand.main(configCommandArgs(kafkaPrincipal, kafkaPassword))
+ ConfigCommand.main(configCommandArgs(clientPrincipal, clientPassword))
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 62015cb..445ee09 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -24,9 +24,12 @@ import kafka.server.{ConfigEntityName, QuotaId}
import kafka.utils.{Logging, ZkUtils}
import kafka.zk.ZooKeeperTestHarness
+import org.apache.kafka.common.security.scram.{ScramCredential, ScramCredentialUtils, ScramMechanism}
import org.easymock.EasyMock
import org.junit.Assert._
import org.junit.Test
+import scala.collection.mutable
+import scala.collection.JavaConverters._
class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
@Test
@@ -233,6 +236,51 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
}
@Test
+ def testScramCredentials(): Unit = {
+ def createOpts(user: String, config: String): ConfigCommandOptions = {
+ new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+ "--entity-name", user,
+ "--entity-type", "users",
+ "--alter",
+ "--add-config", config))
+ }
+
+ def deleteOpts(user: String, mechanism: String) = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+ "--entity-name", user,
+ "--entity-type", "users",
+ "--alter",
+ "--delete-config", mechanism))
+
+ val credentials = mutable.Map[String, Properties]()
+ case class CredentialChange(val user: String, val mechanisms: Set[String], val iterations: Int) extends TestAdminUtils {
+ override def fetchEntityConfig(zkUtils: ZkUtils, entityType: String, entityName: String): Properties = {
+ credentials.getOrElse(entityName, new Properties())
+ }
+ override def changeUserOrUserClientIdConfig(zkUtils: ZkUtils, sanitizedEntityName: String, configChange: Properties): Unit = {
+ assertEquals(user, sanitizedEntityName)
+ assertEquals(mechanisms, configChange.keySet().asScala)
+ for (mechanism <- mechanisms) {
+ val value = configChange.getProperty(mechanism)
+ assertEquals(-1, value.indexOf("password="))
+ val scramCredential = ScramCredentialUtils.credentialFromString(value)
+ assertEquals(iterations, scramCredential.iterations)
+ if (configChange != null)
+ credentials.put(user, configChange)
+ }
+ }
+ }
+ val optsA = createOpts("userA", "SCRAM-SHA-256=[iterations=8192,password=abc, def]")
+ ConfigCommand.alterConfig(null, optsA, CredentialChange("userA", Set("SCRAM-SHA-256"), 8192))
+ val optsB = createOpts("userB", "SCRAM-SHA-256=[iterations=4096,password=abc, def],SCRAM-SHA-512=[password=1234=abc]")
+ ConfigCommand.alterConfig(null, optsB, CredentialChange("userB", Set("SCRAM-SHA-256", "SCRAM-SHA-512"), 4096))
+
+ val del256 = deleteOpts("userB", "SCRAM-SHA-256")
+ ConfigCommand.alterConfig(null, del256, CredentialChange("userB", Set("SCRAM-SHA-512"), 4096))
+ val del512 = deleteOpts("userB", "SCRAM-SHA-512")
+ ConfigCommand.alterConfig(null, del512, CredentialChange("userB", Set(), 4096))
+ }
+
+ @Test
def testQuotaConfigEntity() {
def createOpts(entityType: String, entityName: Option[String], otherArgs: Array[String]) : ConfigCommandOptions = {
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
index f418b30..13b37e1 100755
--- a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
+++ b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
@@ -46,7 +46,7 @@ trait KafkaServerTestHarness extends ZooKeeperTestHarness {
def generateConfigs(): Seq[KafkaConfig]
/**
- * Override this in case ACLs must be set before `servers` are started.
+ * Override this in case ACLs or security credentials must be set before `servers` are started.
*
* This is required in some cases because of the topic creation in the setup of `IntegrationTestHarness`. If the ACLs
* are only set later, tests may fail. The failure could manifest itself as a cluster action
@@ -56,7 +56,7 @@ trait KafkaServerTestHarness extends ZooKeeperTestHarness {
*
* The default implementation of this method is a no-op.
*/
- def setAclsBeforeServersStart() {}
+ def configureSecurityBeforeServersStart() {}
def configs: Seq[KafkaConfig] = {
if (instanceConfigs == null)
@@ -78,7 +78,7 @@ trait KafkaServerTestHarness extends ZooKeeperTestHarness {
throw new KafkaException("Must supply at least one server config.")
// default implementation is a no-op, it is overridden by subclasses if required
- setAclsBeforeServersStart()
+ configureSecurityBeforeServersStart()
servers = configs.map(TestUtils.createServer(_)).toBuffer
brokerList = TestUtils.getBrokerListStrFromServers(servers, securityProtocol)
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index c6f90ff..8547818 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -25,6 +25,7 @@ import javax.net.ssl._
import com.yammer.metrics.core.Gauge
import com.yammer.metrics.{Metrics => YammerMetrics}
+import kafka.security.CredentialProvider
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.common.TopicPartition
@@ -54,7 +55,8 @@ class SocketServerTest extends JUnitSuite {
props.put("connections.max.idle.ms", "60000")
val config = KafkaConfig.fromProps(props)
val metrics = new Metrics
- val server = new SocketServer(config, metrics, Time.SYSTEM)
+ val credentialProvider = new CredentialProvider(config.saslEnabledMechanisms)
+ val server = new SocketServer(config, metrics, Time.SYSTEM, credentialProvider)
server.startup()
val sockets = new ArrayBuffer[Socket]
@@ -241,7 +243,7 @@ class SocketServerTest extends JUnitSuite {
val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
overrideProps.put(KafkaConfig.MaxConnectionsPerIpOverridesProp, s"localhost:$overrideNum")
val serverMetrics = new Metrics()
- val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM)
+ val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM, credentialProvider)
try {
overrideServer.startup()
// make the maximum allowable number of connections
@@ -271,7 +273,7 @@ class SocketServerTest extends JUnitSuite {
overrideProps.put(KafkaConfig.ListenersProp, "SSL://localhost:0")
val serverMetrics = new Metrics
- val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM)
+ val overrideServer = new SocketServer(KafkaConfig.fromProps(overrideProps), serverMetrics, Time.SYSTEM, credentialProvider)
try {
overrideServer.startup()
val sslContext = SSLContext.getInstance("TLSv1.2")
@@ -319,10 +321,10 @@ class SocketServerTest extends JUnitSuite {
val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
val serverMetrics = new Metrics
var conn: Socket = null
- val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM) {
+ val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM, credentialProvider) {
override def newProcessor(id: Int, connectionQuotas: ConnectionQuotas, protocol: SecurityProtocol): Processor = {
new Processor(id, time, config.socketRequestMaxBytes, requestChannel, connectionQuotas,
- config.connectionsMaxIdleMs, protocol, config.values, metrics) {
+ config.connectionsMaxIdleMs, protocol, config.values, metrics, credentialProvider) {
override protected[network] def sendResponse(response: RequestChannel.Response) {
conn.close()
super.sendResponse(response)
@@ -369,7 +371,7 @@ class SocketServerTest extends JUnitSuite {
props.setProperty(KafkaConfig.ConnectionsMaxIdleMsProp, "100")
val serverMetrics = new Metrics
var conn: Socket = null
- val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM)
+ val overrideServer = new SocketServer(KafkaConfig.fromProps(props), serverMetrics, Time.SYSTEM, credentialProvider)
try {
overrideServer.startup()
conn = connect(overrideServer)
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala b/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
index 70b0b2f..ed08e8a 100644
--- a/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
@@ -56,6 +56,22 @@ object JaasTestUtils {
}
}
+ case class ScramLoginModule(username: String,
+ password: String,
+ debug: Boolean = false,
+ validUsers: Map[String, String] = Map.empty) {
+ def toJaasModule: JaasModule = {
+ JaasModule(
+ "org.apache.kafka.common.security.scram.ScramLoginModule",
+ debug = debug,
+ entries = Map(
+ "username" -> username,
+ "password" -> password
+ ) ++ validUsers.map { case (user, pass) => s"user_$user" -> pass }
+ )
+ }
+ }
+
case class JaasModule(moduleName: String,
debug: Boolean,
entries: Map[String, String]) {
@@ -94,6 +110,11 @@ object JaasTestUtils {
private val KafkaPlainAdmin = "admin"
private val KafkaPlainAdminPassword = "admin-secret"
+ val KafkaScramUser = "scram-user"
+ val KafkaScramPassword = "scram-user-secret"
+ val KafkaScramAdmin = "scram-admin"
+ val KafkaScramAdminPassword = "scram-admin-secret"
+
def writeZkFile(): String = {
val jaasFile = TestUtils.tempFile()
writeToFile(jaasFile, zkSections)
@@ -138,6 +159,11 @@ object JaasTestUtils {
KafkaPlainAdminPassword,
debug = false,
Map(KafkaPlainAdmin -> KafkaPlainAdminPassword, KafkaPlainUser -> KafkaPlainPassword)).toJaasModule
+ case "SCRAM-SHA-256" | "SCRAM-SHA-512" =>
+ ScramLoginModule(
+ KafkaScramAdmin,
+ KafkaScramAdminPassword,
+ debug = false).toJaasModule
case mechanism => throw new IllegalArgumentException("Unsupported server mechanism " + mechanism)
}
new JaasSection(KafkaServerContextName, modules)
@@ -159,6 +185,11 @@ object JaasTestUtils {
KafkaPlainUser,
KafkaPlainPassword
).toJaasModule
+ case "SCRAM-SHA-256" | "SCRAM-SHA-512" =>
+ ScramLoginModule(
+ KafkaScramUser,
+ KafkaScramPassword
+ ).toJaasModule
case mechanism => throw new IllegalArgumentException("Unsupported client mechanism " + mechanism)
}
}
[2/2] kafka git commit: KAFKA-3751; SASL/SCRAM implementation
Posted by ju...@apache.org.
KAFKA-3751; SASL/SCRAM implementation
Implementation of KIP-84: https://cwiki.apache.org/confluence/display/KAFKA/KIP-84%3A+Support+SASL+SCRAM+mechanisms
Author: Rajini Sivaram <ra...@googlemail.com>
Reviewers: Jun Rao <ju...@gmail.com>, Ismael Juma <is...@juma.me.uk>
Closes #2086 from rajinisivaram/KAFKA-3751
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/275c5e1d
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/275c5e1d
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/275c5e1d
Branch: refs/heads/trunk
Commit: 275c5e1df237808fe72b8d9933f826949d4b5781
Parents: 1fea1c3
Author: Rajini Sivaram <ra...@googlemail.com>
Authored: Tue Jan 10 08:05:07 2017 -0500
Committer: Jun Rao <ju...@gmail.com>
Committed: Tue Jan 10 08:05:07 2017 -0500
----------------------------------------------------------------------
checkstyle/import-control.xml | 5 +
.../org/apache/kafka/clients/ClientUtils.java | 3 +-
.../kafka/common/network/ChannelBuilders.java | 43 ++-
.../common/network/SaslChannelBuilder.java | 9 +-
.../apache/kafka/common/security/JaasUtils.java | 12 +-
.../security/authenticator/CredentialCache.java | 71 ++++
.../authenticator/SaslServerAuthenticator.java | 12 +-
.../common/security/kerberos/KerberosLogin.java | 2 +-
.../common/security/plain/PlainSaslServer.java | 2 +-
.../common/security/scram/ScramCredential.java | 50 +++
.../security/scram/ScramCredentialCallback.java | 32 ++
.../security/scram/ScramCredentialUtils.java | 86 +++++
.../common/security/scram/ScramFormatter.java | 169 +++++++++
.../common/security/scram/ScramLoginModule.java | 67 ++++
.../common/security/scram/ScramMechanism.java | 79 +++++
.../common/security/scram/ScramMessages.java | 272 +++++++++++++++
.../common/security/scram/ScramSaslClient.java | 239 +++++++++++++
.../security/scram/ScramSaslClientProvider.java | 39 +++
.../common/security/scram/ScramSaslServer.java | 212 +++++++++++
.../security/scram/ScramSaslServerProvider.java | 39 +++
.../scram/ScramServerCallbackHandler.java | 60 ++++
.../kafka/common/network/NioEchoServer.java | 15 +-
.../authenticator/SaslAuthenticatorTest.java | 143 +++++++-
.../security/authenticator/TestJaasConfig.java | 32 +-
.../scram/ScramCredentialUtilsTest.java | 97 ++++++
.../security/scram/ScramFormatterTest.java | 94 +++++
.../security/scram/ScramMessagesTest.java | 348 +++++++++++++++++++
.../src/main/scala/kafka/admin/AdminUtils.scala | 7 +-
.../main/scala/kafka/admin/ConfigCommand.scala | 34 +-
.../controller/ControllerChannelManager.scala | 5 +-
.../main/scala/kafka/network/SocketServer.scala | 13 +-
.../kafka/security/CredentialProvider.scala | 52 +++
.../main/scala/kafka/server/ConfigHandler.scala | 6 +-
.../main/scala/kafka/server/DynamicConfig.scala | 13 +
.../main/scala/kafka/server/KafkaServer.scala | 10 +-
.../kafka/server/ReplicaFetcherThread.scala | 3 +-
.../kafka/api/EndToEndAuthorizationTest.scala | 2 +-
.../SaslScramSslEndToEndAuthorizationTest.scala | 49 +++
.../unit/kafka/admin/ConfigCommandTest.scala | 48 +++
.../integration/KafkaServerTestHarness.scala | 6 +-
.../unit/kafka/network/SocketServerTest.scala | 14 +-
.../scala/unit/kafka/utils/JaasTestUtils.scala | 31 ++
42 files changed, 2462 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 62cd77a..9fcd329 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -77,6 +77,11 @@
<allow pkg="org.apache.kafka.common.errors" />
<allow pkg="org.apache.kafka.clients" />
</subpackage>
+ <subpackage name="scram">
+ <allow pkg="javax.crypto" />
+ <allow pkg="javax.xml.bind" />
+ <allow pkg="org.apache.kafka.common.errors" />
+ </subpackage>
</subpackage>
<subpackage name="protocol">
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
index be663ca..b7ff715 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
@@ -22,7 +22,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.kafka.common.network.ChannelBuilders;
import org.apache.kafka.common.network.LoginType;
-import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Protocol;
import org.apache.kafka.common.protocol.SecurityProtocol;
@@ -86,7 +85,7 @@ public class ClientUtils {
if (!SecurityProtocol.nonTestingValues().contains(securityProtocol))
throw new ConfigException("Invalid SecurityProtocol " + securityProtocol);
String clientSaslMechanism = (String) configs.get(SaslConfigs.SASL_MECHANISM);
- return ChannelBuilders.create(securityProtocol, Mode.CLIENT, LoginType.CLIENT, configs, clientSaslMechanism, true);
+ return ChannelBuilders.clientChannelBuilder(securityProtocol, LoginType.CLIENT, configs, clientSaslMechanism, true);
}
public static Collection<ApiVersionsResponse.ApiVersion> buildExpectedApiVersions(Collection<ApiKeys> apiKeys) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
index 2d6ba8a..5a1486c 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
@@ -17,6 +17,7 @@ import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.security.auth.DefaultPrincipalBuilder;
import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.apache.kafka.common.security.authenticator.CredentialCache;
import org.apache.kafka.common.utils.Utils;
import java.util.Map;
@@ -27,22 +28,48 @@ public class ChannelBuilders {
/**
* @param securityProtocol the securityProtocol
- * @param mode the mode, it must be non-null if `securityProtocol` is not `PLAINTEXT`;
- * it is ignored otherwise
* @param loginType the loginType, it must be non-null if `securityProtocol` is SASL_*; it is ignored otherwise
- * @param configs client/server configs
+ * @param configs client configs
* @param clientSaslMechanism SASL mechanism if mode is CLIENT, ignored otherwise
* @param saslHandshakeRequestEnable flag to enable Sasl handshake requests; disabled only for SASL
* inter-broker connections with inter-broker protocol version < 0.10
* @return the configured `ChannelBuilder`
* @throws IllegalArgumentException if `mode` invariants described above is not maintained
*/
- public static ChannelBuilder create(SecurityProtocol securityProtocol,
+ public static ChannelBuilder clientChannelBuilder(SecurityProtocol securityProtocol,
+ LoginType loginType,
+ Map<String, ?> configs,
+ String clientSaslMechanism,
+ boolean saslHandshakeRequestEnable) {
+
+ if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL) {
+ if (loginType == null)
+ throw new IllegalArgumentException("`loginType` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
+ if (clientSaslMechanism == null)
+ throw new IllegalArgumentException("`clientSaslMechanism` must be non-null in client mode if `securityProtocol` is `" + securityProtocol + "`");
+ }
+ return create(securityProtocol, Mode.CLIENT, loginType, configs, clientSaslMechanism, saslHandshakeRequestEnable, null);
+ }
+
+ /**
+ * @param securityProtocol the securityProtocol
+ * @param configs server configs
+ * @param credentialCache Credential cache for SASL/SCRAM if SCRAM is enabled
+ * @return the configured `ChannelBuilder`
+ */
+ public static ChannelBuilder serverChannelBuilder(SecurityProtocol securityProtocol,
+ Map<String, ?> configs,
+ CredentialCache credentialCache) {
+ return create(securityProtocol, Mode.SERVER, LoginType.SERVER, configs, null, true, credentialCache);
+ }
+
+ private static ChannelBuilder create(SecurityProtocol securityProtocol,
Mode mode,
LoginType loginType,
Map<String, ?> configs,
String clientSaslMechanism,
- boolean saslHandshakeRequestEnable) {
+ boolean saslHandshakeRequestEnable,
+ CredentialCache credentialCache) {
ChannelBuilder channelBuilder;
switch (securityProtocol) {
@@ -53,11 +80,7 @@ public class ChannelBuilders {
case SASL_SSL:
case SASL_PLAINTEXT:
requireNonNullMode(mode, securityProtocol);
- if (loginType == null)
- throw new IllegalArgumentException("`loginType` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
- if (mode == Mode.CLIENT && clientSaslMechanism == null)
- throw new IllegalArgumentException("`clientSaslMechanism` must be non-null in client mode if `securityProtocol` is `" + securityProtocol + "`");
- channelBuilder = new SaslChannelBuilder(mode, loginType, securityProtocol, clientSaslMechanism, saslHandshakeRequestEnable);
+ channelBuilder = new SaslChannelBuilder(mode, loginType, securityProtocol, clientSaslMechanism, saslHandshakeRequestEnable, credentialCache);
break;
case PLAINTEXT:
case TRACE:
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
index ba2f7df..b556f38 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
@@ -23,6 +23,7 @@ import javax.security.auth.login.Configuration;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.JaasUtils;
import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
+import org.apache.kafka.common.security.authenticator.CredentialCache;
import org.apache.kafka.common.security.authenticator.LoginManager;
import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator;
import org.apache.kafka.common.security.authenticator.SaslServerAuthenticator;
@@ -40,6 +41,7 @@ public class SaslChannelBuilder implements ChannelBuilder {
private final Mode mode;
private final LoginType loginType;
private final boolean handshakeRequestEnable;
+ private final CredentialCache credentialCache;
private Configuration jaasConfig;
private LoginManager loginManager;
@@ -47,12 +49,14 @@ public class SaslChannelBuilder implements ChannelBuilder {
private Map<String, ?> configs;
private KerberosShortNamer kerberosShortNamer;
- public SaslChannelBuilder(Mode mode, LoginType loginType, SecurityProtocol securityProtocol, String clientSaslMechanism, boolean handshakeRequestEnable) {
+ public SaslChannelBuilder(Mode mode, LoginType loginType, SecurityProtocol securityProtocol,
+ String clientSaslMechanism, boolean handshakeRequestEnable, CredentialCache credentialCache) {
this.mode = mode;
this.loginType = loginType;
this.securityProtocol = securityProtocol;
this.handshakeRequestEnable = handshakeRequestEnable;
this.clientSaslMechanism = clientSaslMechanism;
+ this.credentialCache = credentialCache;
}
public void configure(Map<String, ?> configs) throws KafkaException {
@@ -98,7 +102,7 @@ public class SaslChannelBuilder implements ChannelBuilder {
Authenticator authenticator;
if (mode == Mode.SERVER)
authenticator = new SaslServerAuthenticator(id, jaasConfig, loginManager.subject(), kerberosShortNamer,
- socketChannel.socket().getLocalAddress().getHostName(), maxReceiveSize);
+ socketChannel.socket().getLocalAddress().getHostName(), maxReceiveSize, credentialCache);
else
authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(),
socketChannel.socket().getInetAddress().getHostName(), clientSaslMechanism, handshakeRequestEnable);
@@ -124,5 +128,4 @@ public class SaslChannelBuilder implements ChannelBuilder {
return new PlaintextTransportLayer(key);
}
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
index aa328d4..10e591d 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java
@@ -83,17 +83,19 @@ public class JaasUtils {
/**
* Returns the configuration option for <code>key</code> from the server login context
- * of the default JAAS configuration.
+ * of the default JAAS configuration. If login module name is specified, return option value
+ * only from that module.
*/
- public static String defaultServerJaasConfigOption(String key) throws IOException {
- return jaasConfigOption(Configuration.getConfiguration(), LoginType.SERVER.contextName(), key);
+ public static String defaultServerJaasConfigOption(String key, String loginModuleName) throws IOException {
+ return jaasConfigOption(Configuration.getConfiguration(), LoginType.SERVER.contextName(), key, loginModuleName);
}
/**
* Returns the configuration option for <code>key</code> from the login context
* <code>loginContextName</code> of the specified JAAS configuration.
+ * If login module name is specified, return option value only from that module.
*/
- public static String jaasConfigOption(Configuration jaasConfig, String loginContextName, String key) throws IOException {
+ public static String jaasConfigOption(Configuration jaasConfig, String loginContextName, String key, String loginModuleName) throws IOException {
AppConfigurationEntry[] configurationEntries = jaasConfig.getAppConfigurationEntry(loginContextName);
if (configurationEntries == null) {
String errorMessage = "Could not find a '" + loginContextName + "' entry in this JAAS configuration.";
@@ -101,6 +103,8 @@ public class JaasUtils {
}
for (AppConfigurationEntry entry: configurationEntries) {
+ if (loginModuleName != null && !loginModuleName.equals(entry.getLoginModuleName()))
+ continue;
Object val = entry.getOptions().get(key);
if (val != null)
return (String) val;
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
new file mode 100644
index 0000000..568dcb5
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/CredentialCache.java
@@ -0,0 +1,71 @@
+/**
+ * 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.kafka.common.security.authenticator;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class CredentialCache {
+
+ private final Map<String, Cache<? extends Object>> cacheMap = new HashMap<>();
+
+ public <C> Cache<C> createCache(String mechanism, Class<C> credentialClass) {
+ Cache<C> cache = new Cache<C>(credentialClass);
+ cacheMap.put(mechanism, cache);
+ return cache;
+ }
+
+ @SuppressWarnings("unchecked")
+ public <C> Cache<C> cache(String mechanism, Class<C> credentialClass) {
+ Cache<?> cache = cacheMap.get(mechanism);
+ if (cache != null) {
+ if (cache.credentialClass() != credentialClass)
+ throw new IllegalArgumentException("Invalid credential class " + credentialClass + ", expected " + cache.credentialClass());
+ return (Cache<C>) cache;
+ } else
+ return null;
+ }
+
+ public static class Cache<C> {
+ private final Class<C> credentialClass;
+ private final ConcurrentHashMap<String, C> credentials;
+
+ public Cache(Class<C> credentialClass) {
+ this.credentialClass = credentialClass;
+ this.credentials = new ConcurrentHashMap<>();
+ }
+
+ public C get(String username) {
+ return credentials.get(username);
+ }
+
+ public C put(String username, C credential) {
+ return credentials.put(username, credential);
+ }
+
+ public C remove(String username) {
+ return credentials.remove(username);
+ }
+
+ public Class<C> credentialClass() {
+ return credentialClass;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
index 27744ad..069e12f 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
@@ -44,6 +44,9 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.PrincipalBuilder;
import org.apache.kafka.common.security.kerberos.KerberosName;
import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
+import org.apache.kafka.common.security.scram.ScramCredential;
+import org.apache.kafka.common.security.scram.ScramMechanism;
+import org.apache.kafka.common.security.scram.ScramServerCallbackHandler;
import org.ietf.jgss.GSSContext;
import org.ietf.jgss.GSSCredential;
import org.ietf.jgss.GSSException;
@@ -83,6 +86,7 @@ public class SaslServerAuthenticator implements Authenticator {
private final KerberosShortNamer kerberosNamer;
private final int maxReceiveSize;
private final String host;
+ private final CredentialCache credentialCache;
// Current SASL state
private SaslState saslState = SaslState.GSSAPI_OR_HANDSHAKE_REQUEST;
@@ -101,7 +105,7 @@ public class SaslServerAuthenticator implements Authenticator {
private NetworkReceive netInBuffer;
private Send netOutBuffer;
- public SaslServerAuthenticator(String node, Configuration jaasConfig, final Subject subject, KerberosShortNamer kerberosNameParser, String host, int maxReceiveSize) throws IOException {
+ public SaslServerAuthenticator(String node, Configuration jaasConfig, final Subject subject, KerberosShortNamer kerberosNameParser, String host, int maxReceiveSize, CredentialCache credentialCache) throws IOException {
if (subject == null)
throw new IllegalArgumentException("subject cannot be null");
this.node = node;
@@ -110,6 +114,7 @@ public class SaslServerAuthenticator implements Authenticator {
this.kerberosNamer = kerberosNameParser;
this.maxReceiveSize = maxReceiveSize;
this.host = host;
+ this.credentialCache = credentialCache;
}
public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs) {
@@ -123,7 +128,10 @@ public class SaslServerAuthenticator implements Authenticator {
private void createSaslServer(String mechanism) throws IOException {
this.saslMechanism = mechanism;
- callbackHandler = new SaslServerCallbackHandler(jaasConfig, kerberosNamer);
+ if (!ScramMechanism.isScram(mechanism))
+ callbackHandler = new SaslServerCallbackHandler(jaasConfig, kerberosNamer);
+ else
+ callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism, ScramCredential.class));
callbackHandler.configure(configs, Mode.SERVER, subject, saslMechanism);
if (mechanism.equals(SaslConfigs.GSSAPI_MECHANISM)) {
if (subject.getPrincipals().isEmpty())
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java
index d8a040b..d2baf91 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java
@@ -294,7 +294,7 @@ public class KerberosLogin extends AbstractLogin {
private String getServiceName(Configuration jaasConfig, Map<String, ?> configs, String loginContext) {
String jaasServiceName;
try {
- jaasServiceName = JaasUtils.jaasConfigOption(jaasConfig, loginContext, JaasUtils.SERVICE_NAME);
+ jaasServiceName = JaasUtils.jaasConfigOption(jaasConfig, loginContext, JaasUtils.SERVICE_NAME, null);
} catch (IOException e) {
throw new KafkaException("JAAS configuration entry not found", e);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java b/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java
index ec7d696..0928057 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java
@@ -92,7 +92,7 @@ public class PlainSaslServer implements SaslServer {
authorizationID = username;
try {
- String expectedPassword = JaasUtils.defaultServerJaasConfigOption(JAAS_USER_PREFIX + username);
+ String expectedPassword = JaasUtils.defaultServerJaasConfigOption(JAAS_USER_PREFIX + username, PlainLoginModule.class.getName());
if (!password.equals(expectedPassword)) {
throw new SaslException("Authentication failed: Invalid username or password");
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredential.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredential.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredential.java
new file mode 100644
index 0000000..7de48f2
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredential.java
@@ -0,0 +1,50 @@
+/**
+ * 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.kafka.common.security.scram;
+
+public class ScramCredential {
+
+ private final byte[] salt;
+ private final byte[] serverKey;
+ private final byte[] storedKey;
+ private final int iterations;
+
+ public ScramCredential(byte[] salt, byte[] storedKey, byte[] serverKey, int iterations) {
+ this.salt = salt;
+ this.serverKey = serverKey;
+ this.storedKey = storedKey;
+ this.iterations = iterations;
+ }
+
+ public byte[] salt() {
+ return salt;
+ }
+
+ public byte[] serverKey() {
+ return serverKey;
+ }
+
+ public byte[] storedKey() {
+ return storedKey;
+ }
+
+ public int iterations() {
+ return iterations;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialCallback.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialCallback.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialCallback.java
new file mode 100644
index 0000000..07c1c93
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialCallback.java
@@ -0,0 +1,32 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import javax.security.auth.callback.Callback;
+
+public class ScramCredentialCallback implements Callback {
+ private ScramCredential scramCredential;
+
+ public ScramCredential scramCredential() {
+ return scramCredential;
+ }
+
+ public void scramCredential(ScramCredential scramCredential) {
+ this.scramCredential = scramCredential;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
new file mode 100644
index 0000000..359e46d
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramCredentialUtils.java
@@ -0,0 +1,86 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.util.Collection;
+import java.util.Properties;
+
+import javax.xml.bind.DatatypeConverter;
+
+import org.apache.kafka.common.security.authenticator.CredentialCache;
+
+/**
+ * SCRAM Credential persistence utility functions. Implements format conversion used
+ * for the credential store implemented in Kafka. Credentials are persisted as a comma-separated
+ * String of key-value pairs:
+ * <pre>
+ * salt=<i>salt</i>,stored_key=<i>stored_key</i>,server_key=<i>server_key</i>,iterations=<i>iterations</i>
+ * </pre>
+ *
+ */
+public class ScramCredentialUtils {
+ private static final String SALT = "salt";
+ private static final String STORED_KEY = "stored_key";
+ private static final String SERVER_KEY = "server_key";
+ private static final String ITERATIONS = "iterations";
+
+ public static String credentialToString(ScramCredential credential) {
+ return String.format("%s=%s,%s=%s,%s=%s,%s=%d",
+ SALT,
+ DatatypeConverter.printBase64Binary(credential.salt()),
+ STORED_KEY,
+ DatatypeConverter.printBase64Binary(credential.storedKey()),
+ SERVER_KEY,
+ DatatypeConverter.printBase64Binary(credential.serverKey()),
+ ITERATIONS,
+ credential.iterations());
+ }
+
+ public static ScramCredential credentialFromString(String str) {
+ Properties props = toProps(str);
+ if (props.size() != 4 || !props.containsKey(SALT) || !props.containsKey(STORED_KEY) ||
+ !props.containsKey(SERVER_KEY) || !props.containsKey(ITERATIONS)) {
+ throw new IllegalArgumentException("Credentials not valid: " + str);
+ }
+ byte[] salt = DatatypeConverter.parseBase64Binary(props.getProperty(SALT));
+ byte[] storedKey = DatatypeConverter.parseBase64Binary(props.getProperty(STORED_KEY));
+ byte[] serverKey = DatatypeConverter.parseBase64Binary(props.getProperty(SERVER_KEY));
+ int iterations = Integer.parseInt(props.getProperty(ITERATIONS));
+ return new ScramCredential(salt, storedKey, serverKey, iterations);
+ }
+
+ private static Properties toProps(String str) {
+ Properties props = new Properties();
+ String[] tokens = str.split(",");
+ for (String token : tokens) {
+ int index = token.indexOf('=');
+ if (index <= 0)
+ throw new IllegalArgumentException("Credentials not valid: " + str);
+ props.put(token.substring(0, index), token.substring(index + 1));
+ }
+ return props;
+ }
+
+ public static void createCache(CredentialCache cache, Collection<String> enabledMechanisms) {
+ for (String mechanism : ScramMechanism.mechanismNames()) {
+ if (enabledMechanisms.contains(mechanism))
+ cache.createCache(mechanism, ScramCredential.class);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramFormatter.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramFormatter.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramFormatter.java
new file mode 100644
index 0000000..e600573
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramFormatter.java
@@ -0,0 +1,169 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFirstMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFirstMessage;
+
+/**
+ * Scram message salt and hash functions defined in <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>.
+ */
+public class ScramFormatter {
+
+ private final MessageDigest messageDigest;
+ private final Mac mac;
+ private final SecureRandom random;
+
+ public ScramFormatter(ScramMechanism mechanism) throws NoSuchAlgorithmException {
+ this.messageDigest = MessageDigest.getInstance(mechanism.hashAlgorithm());
+ this.mac = Mac.getInstance(mechanism.macAlgorithm());
+ this.random = new SecureRandom();
+ }
+
+ public byte[] hmac(byte[] key, byte[] bytes) throws InvalidKeyException {
+ mac.init(new SecretKeySpec(key, mac.getAlgorithm()));
+ return mac.doFinal(bytes);
+ }
+
+ public byte[] hash(byte[] str) {
+ return messageDigest.digest(str);
+ }
+
+ public byte[] xor(byte[] first, byte[] second) {
+ if (first.length != second.length)
+ throw new IllegalArgumentException("Argument arrays must be of the same length");
+ byte[] result = new byte[first.length];
+ for (int i = 0; i < result.length; i++)
+ result[i] = (byte) (first[i] ^ second[i]);
+ return result;
+ }
+
+ public byte[] hi(byte[] str, byte[] salt, int iterations) throws InvalidKeyException {
+ mac.init(new SecretKeySpec(str, mac.getAlgorithm()));
+ mac.update(salt);
+ byte[] u1 = mac.doFinal(new byte[]{0, 0, 0, 1});
+ byte[] prev = u1;
+ byte[] result = u1;
+ for (int i = 2; i <= iterations; i++) {
+ byte[] ui = hmac(str, prev);
+ result = xor(result, ui);
+ prev = ui;
+ }
+ return result;
+ }
+
+ public byte[] normalize(String str) {
+ return toBytes(str);
+ }
+
+ public byte[] saltedPassword(String password, byte[] salt, int iterations) throws InvalidKeyException {
+ return hi(normalize(password), salt, iterations);
+ }
+
+ public byte[] clientKey(byte[] saltedPassword) throws InvalidKeyException {
+ return hmac(saltedPassword, toBytes("Client Key"));
+ }
+
+ public byte[] storedKey(byte[] clientKey) {
+ return hash(clientKey);
+ }
+
+ public String saslName(String username) {
+ return username.replace("=", "=3D").replace(",", "=2C");
+ }
+
+ public String username(String saslName) {
+ String username = saslName.replace("=2C", ",");
+ if (username.replace("=3D", "").indexOf('=') >= 0)
+ throw new IllegalArgumentException("Invalid username: " + saslName);
+ return username.replace("=3D", "=");
+ }
+
+ public String authMessage(String clientFirstMessageBare, String serverFirstMessage, String clientFinalMessageWithoutProof) {
+ return clientFirstMessageBare + "," + serverFirstMessage + "," + clientFinalMessageWithoutProof;
+ }
+
+ public byte[] clientSignature(byte[] storedKey, ClientFirstMessage clientFirstMessage, ServerFirstMessage serverFirstMessage, ClientFinalMessage clientFinalMessage) throws InvalidKeyException {
+ byte[] authMessage = authMessage(clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ return hmac(storedKey, authMessage);
+ }
+
+ public byte[] clientProof(byte[] saltedPassword, ClientFirstMessage clientFirstMessage, ServerFirstMessage serverFirstMessage, ClientFinalMessage clientFinalMessage) throws InvalidKeyException {
+ byte[] clientKey = clientKey(saltedPassword);
+ byte[] storedKey = hash(clientKey);
+ byte[] clientSignature = hmac(storedKey, authMessage(clientFirstMessage, serverFirstMessage, clientFinalMessage));
+ return xor(clientKey, clientSignature);
+ }
+
+ private byte[] authMessage(ClientFirstMessage clientFirstMessage, ServerFirstMessage serverFirstMessage, ClientFinalMessage clientFinalMessage) {
+ return toBytes(authMessage(clientFirstMessage.clientFirstMessageBare(),
+ serverFirstMessage.toMessage(),
+ clientFinalMessage.clientFinalMessageWithoutProof()));
+ }
+
+ public byte[] storedKey(byte[] clientSignature, byte[] clientProof) {
+ return hash(xor(clientSignature, clientProof));
+ }
+
+ public byte[] serverKey(byte[] saltedPassword) throws InvalidKeyException {
+ return hmac(saltedPassword, toBytes("Server Key"));
+ }
+
+ public byte[] serverSignature(byte[] serverKey, ClientFirstMessage clientFirstMessage, ServerFirstMessage serverFirstMessage, ClientFinalMessage clientFinalMessage) throws InvalidKeyException {
+ byte[] authMessage = authMessage(clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ return hmac(serverKey, authMessage);
+ }
+
+ public String secureRandomString() {
+ return new BigInteger(130, random).toString(Character.MAX_RADIX);
+ }
+
+ public byte[] secureRandomBytes() {
+ return toBytes(secureRandomString());
+ }
+
+ public byte[] toBytes(String str) {
+ return str.getBytes(StandardCharsets.UTF_8);
+ }
+
+ public ScramCredential generateCredential(String password, int iterations) {
+ try {
+ byte[] salt = secureRandomBytes();
+ byte[] saltedPassword = saltedPassword(password, salt, iterations);
+ byte[] clientKey = clientKey(saltedPassword);
+ byte[] storedKey = storedKey(clientKey);
+ byte[] serverKey = serverKey(saltedPassword);
+ return new ScramCredential(salt, storedKey, serverKey, iterations);
+ } catch (InvalidKeyException e) {
+ throw new KafkaException("Could not create credential", e);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramLoginModule.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramLoginModule.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramLoginModule.java
new file mode 100644
index 0000000..1e4b643
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramLoginModule.java
@@ -0,0 +1,67 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.util.Map;
+
+import javax.security.auth.Subject;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.login.LoginException;
+import javax.security.auth.spi.LoginModule;
+
+public class ScramLoginModule implements LoginModule {
+
+ private static final String USERNAME_CONFIG = "username";
+ private static final String PASSWORD_CONFIG = "password";
+
+ static {
+ ScramSaslClientProvider.initialize();
+ ScramSaslServerProvider.initialize();
+ }
+
+ @Override
+ public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
+ String username = (String) options.get(USERNAME_CONFIG);
+ if (username != null)
+ subject.getPublicCredentials().add(username);
+ String password = (String) options.get(PASSWORD_CONFIG);
+ if (password != null)
+ subject.getPrivateCredentials().add(password);
+ }
+
+ @Override
+ public boolean login() throws LoginException {
+ return true;
+ }
+
+ @Override
+ public boolean logout() throws LoginException {
+ return true;
+ }
+
+ @Override
+ public boolean commit() throws LoginException {
+ return true;
+ }
+
+ @Override
+ public boolean abort() throws LoginException {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMechanism.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMechanism.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMechanism.java
new file mode 100644
index 0000000..86b51bc
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMechanism.java
@@ -0,0 +1,79 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public enum ScramMechanism {
+
+ SCRAM_SHA_256("SHA-256", "HmacSHA256", 4096),
+ SCRAM_SHA_512("SHA-512", "HmacSHA512", 4096);
+
+ private final String mechanismName;
+ private final String hashAlgorithm;
+ private final String macAlgorithm;
+ private final int minIterations;
+
+ private static final Map<String, ScramMechanism> MECHANISMS_MAP;
+
+ static {
+ Map<String, ScramMechanism> map = new HashMap<>();
+ for (ScramMechanism mech : values())
+ map.put(mech.mechanismName, mech);
+ MECHANISMS_MAP = Collections.unmodifiableMap(map);
+ }
+
+ ScramMechanism(String hashAlgorithm, String macAlgorithm, int minIterations) {
+ this.mechanismName = "SCRAM-" + hashAlgorithm;
+ this.hashAlgorithm = hashAlgorithm;
+ this.macAlgorithm = macAlgorithm;
+ this.minIterations = minIterations;
+ }
+
+ public final String mechanismName() {
+ return mechanismName;
+ }
+
+ public String hashAlgorithm() {
+ return hashAlgorithm;
+ }
+
+ public String macAlgorithm() {
+ return macAlgorithm;
+ }
+
+ public int minIterations() {
+ return minIterations;
+ }
+
+ public static ScramMechanism forMechanismName(String mechanismName) {
+ return MECHANISMS_MAP.get(mechanismName);
+ }
+
+ public static Collection<String> mechanismNames() {
+ return MECHANISMS_MAP.keySet();
+ }
+
+ public static boolean isScram(String mechanismName) {
+ return MECHANISMS_MAP.containsKey(mechanismName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMessages.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMessages.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMessages.java
new file mode 100644
index 0000000..3f94e65
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramMessages.java
@@ -0,0 +1,272 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.nio.charset.StandardCharsets;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import javax.security.sasl.SaslException;
+import javax.xml.bind.DatatypeConverter;
+
+/**
+ * SCRAM request/response message creation and parsing based on
+ * <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>
+ *
+ */
+public class ScramMessages {
+
+ static abstract class AbstractScramMessage {
+
+ static final String ALPHA = "[A-Za-z]+";
+ static final String VALUE_SAFE = "[\\x01-\\x7F&&[^=,]]+";
+ static final String VALUE = "[\\x01-\\x7F&&[^,]]+";
+ static final String PRINTABLE = "[\\x21-\\x7E&&[^,]]+";
+ static final String SASLNAME = "(?:[\\x01-\\x7F&&[^=,]]|=2C|=3D)+";
+ static final String BASE64_CHAR = "[a-zA-Z0-9/+]";
+ static final String BASE64 = String.format("(?:%s{4})*(?:%s{3}=|%s{2}==)?", BASE64_CHAR, BASE64_CHAR, BASE64_CHAR);
+ static final String RESERVED = String.format("(m=%s,)?", VALUE);
+ static final String EXTENSIONS = String.format("(,%s=%s)*", ALPHA, VALUE);
+
+ abstract String toMessage();
+
+ public byte[] toBytes() {
+ return toMessage().getBytes(StandardCharsets.UTF_8);
+ }
+
+ protected String toMessage(byte[] messageBytes) {
+ return new String(messageBytes, StandardCharsets.UTF_8);
+ }
+ }
+
+ /**
+ * Format:
+ * gs2-header [reserved-mext ","] username "," nonce ["," extensions]
+ * Limitations:
+ * Only gs2-header "n" is supported.
+ * Extensions are ignored.
+ *
+ */
+ public static class ClientFirstMessage extends AbstractScramMessage {
+ private static final Pattern PATTERN = Pattern.compile(String.format(
+ "n,(a=(?<authzid>%s))?,%sn=(?<saslname>%s),r=(?<nonce>%s)%s",
+ SASLNAME,
+ RESERVED,
+ SASLNAME,
+ PRINTABLE,
+ EXTENSIONS));
+
+ private final String saslName;
+ private final String nonce;
+ private final String authorizationId;
+ public ClientFirstMessage(byte[] messageBytes) throws SaslException {
+ String message = toMessage(messageBytes);
+ Matcher matcher = PATTERN.matcher(message);
+ if (!matcher.matches())
+ throw new SaslException("Invalid SCRAM client first message format: " + message);
+ String authzid = matcher.group("authzid");
+ this.authorizationId = authzid != null ? authzid : "";
+ this.saslName = matcher.group("saslname");
+ this.nonce = matcher.group("nonce");
+ }
+ public ClientFirstMessage(String saslName, String nonce) {
+ this.saslName = saslName;
+ this.nonce = nonce;
+ this.authorizationId = ""; // Optional authzid not specified in gs2-header
+ }
+ public String saslName() {
+ return saslName;
+ }
+ public String nonce() {
+ return nonce;
+ }
+ public String authorizationId() {
+ return authorizationId;
+ }
+ public String gs2Header() {
+ return "n," + authorizationId + ",";
+ }
+ public String clientFirstMessageBare() {
+ return String.format("n=%s,r=%s", saslName, nonce);
+ }
+ String toMessage() {
+ return gs2Header() + clientFirstMessageBare();
+ }
+ }
+
+ /**
+ * Format:
+ * [reserved-mext ","] nonce "," salt "," iteration-count ["," extensions]
+ * Limitations:
+ * Extensions are ignored.
+ *
+ */
+ public static class ServerFirstMessage extends AbstractScramMessage {
+ private static final Pattern PATTERN = Pattern.compile(String.format(
+ "%sr=(?<nonce>%s),s=(?<salt>%s),i=(?<iterations>[0-9]+)%s",
+ RESERVED,
+ PRINTABLE,
+ BASE64,
+ EXTENSIONS));
+
+ private final String nonce;
+ private final byte[] salt;
+ private final int iterations;
+ public ServerFirstMessage(byte[] messageBytes) throws SaslException {
+ String message = toMessage(messageBytes);
+ Matcher matcher = PATTERN.matcher(message);
+ if (!matcher.matches())
+ throw new SaslException("Invalid SCRAM server first message format: " + message);
+ try {
+ this.iterations = Integer.parseInt(matcher.group("iterations"));
+ if (this.iterations <= 0)
+ throw new SaslException("Invalid SCRAM server first message format: invalid iterations " + iterations);
+ } catch (NumberFormatException e) {
+ throw new SaslException("Invalid SCRAM server first message format: invalid iterations");
+ }
+ this.nonce = matcher.group("nonce");
+ String salt = matcher.group("salt");
+ this.salt = DatatypeConverter.parseBase64Binary(salt);
+ }
+ public ServerFirstMessage(String clientNonce, String serverNonce, byte[] salt, int iterations) {
+ this.nonce = clientNonce + serverNonce;
+ this.salt = salt;
+ this.iterations = iterations;
+ }
+ public String nonce() {
+ return nonce;
+ }
+ public byte[] salt() {
+ return salt;
+ }
+ public int iterations() {
+ return iterations;
+ }
+ String toMessage() {
+ return String.format("r=%s,s=%s,i=%d", nonce, DatatypeConverter.printBase64Binary(salt), iterations);
+ }
+ }
+ /**
+ * Format:
+ * channel-binding "," nonce ["," extensions]"," proof
+ * Limitations:
+ * Extensions are ignored.
+ *
+ */
+ public static class ClientFinalMessage extends AbstractScramMessage {
+ private static final Pattern PATTERN = Pattern.compile(String.format(
+ "c=(?<channel>%s),r=(?<nonce>%s)%s,p=(?<proof>%s)",
+ BASE64,
+ PRINTABLE,
+ EXTENSIONS,
+ BASE64));
+
+ private final byte[] channelBinding;
+ private final String nonce;
+ private byte[] proof;
+ public ClientFinalMessage(byte[] messageBytes) throws SaslException {
+ String message = toMessage(messageBytes);
+ Matcher matcher = PATTERN.matcher(message);
+ if (!matcher.matches())
+ throw new SaslException("Invalid SCRAM client final message format: " + message);
+
+ this.channelBinding = DatatypeConverter.parseBase64Binary(matcher.group("channel"));
+ this.nonce = matcher.group("nonce");
+ this.proof = DatatypeConverter.parseBase64Binary(matcher.group("proof"));
+ }
+ public ClientFinalMessage(byte[] channelBinding, String nonce) {
+ this.channelBinding = channelBinding;
+ this.nonce = nonce;
+ }
+ public byte[] channelBinding() {
+ return channelBinding;
+ }
+ public String nonce() {
+ return nonce;
+ }
+ public byte[] proof() {
+ return proof;
+ }
+ public void proof(byte[] proof) {
+ this.proof = proof;
+ }
+ public String clientFinalMessageWithoutProof() {
+ return String.format("c=%s,r=%s",
+ DatatypeConverter.printBase64Binary(channelBinding),
+ nonce);
+ }
+ String toMessage() {
+ return String.format("%s,p=%s",
+ clientFinalMessageWithoutProof(),
+ DatatypeConverter.printBase64Binary(proof));
+ }
+ }
+ /**
+ * Format:
+ * ("e=" server-error-value | "v=" base64_server_signature) ["," extensions]
+ * Limitations:
+ * Extensions are ignored.
+ *
+ */
+ public static class ServerFinalMessage extends AbstractScramMessage {
+ private static final Pattern PATTERN = Pattern.compile(String.format(
+ "(?:e=(?<error>%s))|(?:v=(?<signature>%s))%s",
+ VALUE_SAFE,
+ BASE64,
+ EXTENSIONS));
+
+ private final String error;
+ private final byte[] serverSignature;
+ public ServerFinalMessage(byte[] messageBytes) throws SaslException {
+ String message = toMessage(messageBytes);
+ Matcher matcher = PATTERN.matcher(message);
+ if (!matcher.matches())
+ throw new SaslException("Invalid SCRAM server final message format: " + message);
+ String error = null;
+ try {
+ error = matcher.group("error");
+ } catch (IllegalArgumentException e) {
+ // ignore
+ }
+ if (error == null) {
+ this.serverSignature = DatatypeConverter.parseBase64Binary(matcher.group("signature"));
+ this.error = null;
+ } else {
+ this.serverSignature = null;
+ this.error = error;
+ }
+ }
+ public ServerFinalMessage(String error, byte[] serverSignature) {
+ this.error = error;
+ this.serverSignature = serverSignature;
+ }
+ public String error() {
+ return error;
+ }
+ public byte[] serverSignature() {
+ return serverSignature;
+ }
+ String toMessage() {
+ if (error != null)
+ return "e=" + error;
+ else
+ return "v=" + DatatypeConverter.printBase64Binary(serverSignature);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClient.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClient.java
new file mode 100644
index 0000000..4220a69
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClient.java
@@ -0,0 +1,239 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslClientFactory;
+import javax.security.sasl.SaslException;
+
+import org.apache.kafka.common.errors.IllegalSaslStateException;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFirstMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SaslClient implementation for SASL/SCRAM.
+ * <p>
+ * This implementation expects a login module that populates username as
+ * the Subject's public credential and password as the private credential.
+ *
+ * @see <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>
+ *
+ */
+public class ScramSaslClient implements SaslClient {
+
+ private static final Logger log = LoggerFactory.getLogger(ScramSaslClient.class);
+
+ enum State {
+ SEND_CLIENT_FIRST_MESSAGE,
+ RECEIVE_SERVER_FIRST_MESSAGE,
+ RECEIVE_SERVER_FINAL_MESSAGE,
+ COMPLETE,
+ FAILED
+ };
+
+ private final ScramMechanism mechanism;
+ private final CallbackHandler callbackHandler;
+ private final ScramFormatter formatter;
+ private String clientNonce;
+ private State state;
+ private byte[] saltedPassword;
+ private ScramMessages.ClientFirstMessage clientFirstMessage;
+ private ScramMessages.ServerFirstMessage serverFirstMessage;
+ private ScramMessages.ClientFinalMessage clientFinalMessage;
+
+ public ScramSaslClient(ScramMechanism mechanism, CallbackHandler cbh) throws NoSuchAlgorithmException {
+ this.mechanism = mechanism;
+ this.callbackHandler = cbh;
+ this.formatter = new ScramFormatter(mechanism);
+ setState(State.SEND_CLIENT_FIRST_MESSAGE);
+ }
+
+ @Override
+ public String getMechanismName() {
+ return mechanism.mechanismName();
+ }
+
+ @Override
+ public boolean hasInitialResponse() {
+ return true;
+ }
+
+ @Override
+ public byte[] evaluateChallenge(byte[] challenge) throws SaslException {
+ try {
+ switch (state) {
+ case SEND_CLIENT_FIRST_MESSAGE:
+ if (challenge != null && challenge.length != 0)
+ throw new SaslException("Expected empty challenge");
+ clientNonce = formatter.secureRandomString();
+ NameCallback nameCallback = new NameCallback("Name:");
+ try {
+ callbackHandler.handle(new Callback[]{nameCallback});
+ } catch (IOException | UnsupportedCallbackException e) {
+ throw new SaslException("User name could not be obtained", e);
+ }
+ String username = nameCallback.getName();
+ String saslName = formatter.saslName(username);
+ this.clientFirstMessage = new ScramMessages.ClientFirstMessage(saslName, clientNonce);
+ setState(State.RECEIVE_SERVER_FIRST_MESSAGE);
+ return clientFirstMessage.toBytes();
+
+ case RECEIVE_SERVER_FIRST_MESSAGE:
+ this.serverFirstMessage = new ServerFirstMessage(challenge);
+ if (!serverFirstMessage.nonce().startsWith(clientNonce))
+ throw new SaslException("Invalid server nonce: does not start with client nonce");
+ if (serverFirstMessage.iterations() < mechanism.minIterations())
+ throw new SaslException("Requested iterations " + serverFirstMessage.iterations() + " is less than the minimum " + mechanism.minIterations() + " for " + mechanism);
+ PasswordCallback passwordCallback = new PasswordCallback("Password:", false);
+ try {
+ callbackHandler.handle(new Callback[]{passwordCallback});
+ } catch (IOException | UnsupportedCallbackException e) {
+ throw new SaslException("User name could not be obtained", e);
+ }
+ this.clientFinalMessage = handleServerFirstMessage(passwordCallback.getPassword());
+ setState(State.RECEIVE_SERVER_FINAL_MESSAGE);
+ return clientFinalMessage.toBytes();
+
+ case RECEIVE_SERVER_FINAL_MESSAGE:
+ ServerFinalMessage serverFinalMessage = new ServerFinalMessage(challenge);
+ if (serverFinalMessage.error() != null)
+ throw new SaslException("Sasl authentication using " + mechanism + " failed with error: " + serverFinalMessage.error());
+ handleServerFinalMessage(serverFinalMessage.serverSignature());
+ setState(State.COMPLETE);
+ return null;
+
+ default:
+ throw new IllegalSaslStateException("Unexpected challenge in Sasl client state " + state);
+ }
+ } catch (SaslException e) {
+ setState(State.FAILED);
+ throw e;
+ }
+ }
+
+ @Override
+ public boolean isComplete() {
+ return state == State.COMPLETE;
+ }
+
+ @Override
+ public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return Arrays.copyOfRange(incoming, offset, offset + len);
+ }
+
+ @Override
+ public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return Arrays.copyOfRange(outgoing, offset, offset + len);
+ }
+
+ @Override
+ public Object getNegotiatedProperty(String propName) {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return null;
+ }
+
+ @Override
+ public void dispose() throws SaslException {
+ }
+
+ private void setState(State state) {
+ log.debug("Setting SASL/{} client state to {}", mechanism, state);
+ this.state = state;
+ }
+
+ private ClientFinalMessage handleServerFirstMessage(char[] password) throws SaslException {
+ try {
+ byte[] passwordBytes = formatter.normalize(new String(password));
+ this.saltedPassword = formatter.hi(passwordBytes, serverFirstMessage.salt(), serverFirstMessage.iterations());
+
+ ClientFinalMessage clientFinalMessage = new ClientFinalMessage("n,,".getBytes(StandardCharsets.UTF_8), serverFirstMessage.nonce());
+ byte[] clientProof = formatter.clientProof(saltedPassword, clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ clientFinalMessage.proof(clientProof);
+ return clientFinalMessage;
+ } catch (InvalidKeyException e) {
+ throw new SaslException("Client final message could not be created", e);
+ }
+ }
+
+ private void handleServerFinalMessage(byte[] signature) throws SaslException {
+ try {
+ byte[] serverKey = formatter.serverKey(saltedPassword);
+ byte[] serverSignature = formatter.serverSignature(serverKey, clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ if (!Arrays.equals(signature, serverSignature))
+ throw new SaslException("Invalid server signature in server final message");
+ } catch (InvalidKeyException e) {
+ throw new SaslException("Sasl server signature verification failed", e);
+ }
+ }
+
+ public static class ScramSaslClientFactory implements SaslClientFactory {
+
+ @Override
+ public SaslClient createSaslClient(String[] mechanisms,
+ String authorizationId,
+ String protocol,
+ String serverName,
+ Map<String, ?> props,
+ CallbackHandler cbh) throws SaslException {
+
+ ScramMechanism mechanism = null;
+ for (String mech : mechanisms) {
+ mechanism = ScramMechanism.forMechanismName(mech);
+ if (mechanism != null)
+ break;
+ }
+ if (mechanism == null)
+ throw new SaslException(String.format("Requested mechanisms '%s' not supported. Supported mechanisms are '%s'.",
+ Arrays.asList(mechanisms), ScramMechanism.mechanismNames()));
+
+ try {
+ return new ScramSaslClient(mechanism, cbh);
+ } catch (NoSuchAlgorithmException e) {
+ throw new SaslException("Hash algorithm not supported for mechanism " + mechanism, e);
+ }
+ }
+
+ @Override
+ public String[] getMechanismNames(Map<String, ?> props) {
+ Collection<String> mechanisms = ScramMechanism.mechanismNames();
+ return mechanisms.toArray(new String[mechanisms.size()]);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClientProvider.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClientProvider.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClientProvider.java
new file mode 100644
index 0000000..aa5fc56
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslClientProvider.java
@@ -0,0 +1,39 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.security.Provider;
+import java.security.Security;
+
+import org.apache.kafka.common.security.scram.ScramSaslClient.ScramSaslClientFactory;
+
+public class ScramSaslClientProvider extends Provider {
+
+ private static final long serialVersionUID = 1L;
+
+ protected ScramSaslClientProvider() {
+ super("SASL/SCRAM Client Provider", 1.0, "SASL/SCRAM Client Provider for Kafka");
+ for (ScramMechanism mechanism : ScramMechanism.values())
+ super.put("SaslClientFactory." + mechanism.mechanismName(), ScramSaslClientFactory.class.getName());
+ }
+
+ public static void initialize() {
+ Security.addProvider(new ScramSaslClientProvider());
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServer.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServer.java
new file mode 100644
index 0000000..4298f98
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServer.java
@@ -0,0 +1,212 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.io.IOException;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import javax.security.sasl.SaslServerFactory;
+
+import org.apache.kafka.common.errors.IllegalSaslStateException;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ClientFirstMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFinalMessage;
+import org.apache.kafka.common.security.scram.ScramMessages.ServerFirstMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SaslServer implementation for SASL/SCRAM. This server is configured with a callback
+ * handler for integration with a credential manager. Kafka brokers provide callbacks
+ * based on a Zookeeper-based password store.
+ *
+ * @see <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>
+ */
+public class ScramSaslServer implements SaslServer {
+
+ private static final Logger log = LoggerFactory.getLogger(ScramSaslServer.class);
+
+ enum State {
+ RECEIVE_CLIENT_FIRST_MESSAGE,
+ RECEIVE_CLIENT_FINAL_MESSAGE,
+ COMPLETE,
+ FAILED
+ };
+
+ private final ScramMechanism mechanism;
+ private final ScramFormatter formatter;
+ private final CallbackHandler callbackHandler;
+ private State state;
+ private String username;
+ private ClientFirstMessage clientFirstMessage;
+ private ServerFirstMessage serverFirstMessage;
+ private String serverNonce;
+ private ScramCredential scramCredential;
+
+ public ScramSaslServer(ScramMechanism mechanism, Map<String, ?> props, CallbackHandler callbackHandler) throws NoSuchAlgorithmException {
+ this.mechanism = mechanism;
+ this.formatter = new ScramFormatter(mechanism);
+ this.callbackHandler = callbackHandler;
+ setState(State.RECEIVE_CLIENT_FIRST_MESSAGE);
+ }
+
+ @Override
+ public byte[] evaluateResponse(byte[] response) throws SaslException {
+ try {
+ switch (state) {
+ case RECEIVE_CLIENT_FIRST_MESSAGE:
+ this.clientFirstMessage = new ClientFirstMessage(response);
+ serverNonce = formatter.secureRandomString();
+ try {
+ String saslName = clientFirstMessage.saslName();
+ this.username = formatter.username(saslName);
+ NameCallback nameCallback = new NameCallback("username", username);
+ ScramCredentialCallback credentialCallback = new ScramCredentialCallback();
+ callbackHandler.handle(new Callback[]{nameCallback, credentialCallback});
+ this.scramCredential = credentialCallback.scramCredential();
+ if (scramCredential == null)
+ throw new SaslException("Authentication failed: Invalid user credentials");
+ if (scramCredential.iterations() < mechanism.minIterations())
+ throw new SaslException("Iterations " + scramCredential.iterations() + " is less than the minimum " + mechanism.minIterations() + " for " + mechanism);
+ this.serverFirstMessage = new ServerFirstMessage(clientFirstMessage.nonce(),
+ serverNonce,
+ scramCredential.salt(),
+ scramCredential.iterations());
+ setState(State.RECEIVE_CLIENT_FINAL_MESSAGE);
+ return serverFirstMessage.toBytes();
+ } catch (IOException | NumberFormatException | UnsupportedCallbackException e) {
+ throw new SaslException("Authentication failed: Credentials could not be obtained", e);
+ }
+
+ case RECEIVE_CLIENT_FINAL_MESSAGE:
+ try {
+ ClientFinalMessage clientFinalMessage = new ClientFinalMessage(response);
+ verifyClientProof(clientFinalMessage);
+ byte[] serverKey = scramCredential.serverKey();
+ byte[] serverSignature = formatter.serverSignature(serverKey, clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ ServerFinalMessage serverFinalMessage = new ServerFinalMessage(null, serverSignature);
+ setState(State.COMPLETE);
+ return serverFinalMessage.toBytes();
+ } catch (InvalidKeyException e) {
+ throw new SaslException("Authentication failed: Invalid client final message", e);
+ }
+
+ default:
+ throw new IllegalSaslStateException("Unexpected challenge in Sasl server state " + state);
+ }
+ } catch (SaslException e) {
+ setState(State.FAILED);
+ throw e;
+ }
+ }
+
+ @Override
+ public String getAuthorizationID() {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ String authzId = clientFirstMessage.authorizationId();
+ return authzId == null || authzId.length() == 0 ? username : authzId;
+ }
+
+ @Override
+ public String getMechanismName() {
+ return mechanism.mechanismName();
+ }
+
+ @Override
+ public Object getNegotiatedProperty(String propName) {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return null;
+ }
+
+ @Override
+ public boolean isComplete() {
+ return state == State.COMPLETE;
+ }
+
+ @Override
+ public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return Arrays.copyOfRange(incoming, offset, offset + len);
+ }
+
+ @Override
+ public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException {
+ if (!isComplete())
+ throw new IllegalStateException("Authentication exchange has not completed");
+ return Arrays.copyOfRange(outgoing, offset, offset + len);
+ }
+
+ @Override
+ public void dispose() throws SaslException {
+ }
+
+ private void setState(State state) {
+ log.debug("Setting SASL/{} server state to {}", mechanism, state);
+ this.state = state;
+ }
+
+ private void verifyClientProof(ClientFinalMessage clientFinalMessage) throws SaslException {
+ try {
+ byte[] expectedStoredKey = scramCredential.storedKey();
+ byte[] clientSignature = formatter.clientSignature(expectedStoredKey, clientFirstMessage, serverFirstMessage, clientFinalMessage);
+ byte[] computedStoredKey = formatter.storedKey(clientSignature, clientFinalMessage.proof());
+ if (!Arrays.equals(computedStoredKey, expectedStoredKey))
+ throw new SaslException("Invalid client credentials");
+ } catch (InvalidKeyException e) {
+ throw new SaslException("Sasl client verification failed", e);
+ }
+ }
+
+ public static class ScramSaslServerFactory implements SaslServerFactory {
+
+ @Override
+ public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props, CallbackHandler cbh)
+ throws SaslException {
+
+ if (!ScramMechanism.isScram(mechanism)) {
+ throw new SaslException(String.format("Requested mechanism '%s' is not supported. Supported mechanisms are '%s'.",
+ mechanism, ScramMechanism.mechanismNames()));
+ }
+ try {
+ return new ScramSaslServer(ScramMechanism.forMechanismName(mechanism), props, cbh);
+ } catch (NoSuchAlgorithmException e) {
+ throw new SaslException("Hash algorithm not supported for mechanism " + mechanism, e);
+ }
+ }
+
+ @Override
+ public String[] getMechanismNames(Map<String, ?> props) {
+ Collection<String> mechanisms = ScramMechanism.mechanismNames();
+ return mechanisms.toArray(new String[mechanisms.size()]);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServerProvider.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServerProvider.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServerProvider.java
new file mode 100644
index 0000000..496cb5f
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramSaslServerProvider.java
@@ -0,0 +1,39 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.security.Provider;
+import java.security.Security;
+
+import org.apache.kafka.common.security.scram.ScramSaslServer.ScramSaslServerFactory;
+
+public class ScramSaslServerProvider extends Provider {
+
+ private static final long serialVersionUID = 1L;
+
+ protected ScramSaslServerProvider() {
+ super("SASL/SCRAM Server Provider", 1.0, "SASL/SCRAM Server Provider for Kafka");
+ for (ScramMechanism mechanism : ScramMechanism.values())
+ super.put("SaslServerFactory." + mechanism.mechanismName(), ScramSaslServerFactory.class.getName());
+ }
+
+ public static void initialize() {
+ Security.addProvider(new ScramSaslServerProvider());
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/275c5e1d/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java
new file mode 100644
index 0000000..46bfe57
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java
@@ -0,0 +1,60 @@
+/**
+ * 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.kafka.common.security.scram;
+
+import java.io.IOException;
+import java.util.Map;
+
+import javax.security.auth.Subject;
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+
+import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.security.auth.AuthCallbackHandler;
+import org.apache.kafka.common.security.authenticator.CredentialCache;
+
+public class ScramServerCallbackHandler implements AuthCallbackHandler {
+
+ private final CredentialCache.Cache<ScramCredential> credentialCache;
+
+ public ScramServerCallbackHandler(CredentialCache.Cache<ScramCredential> credentialCache) {
+ this.credentialCache = credentialCache;
+ }
+
+ @Override
+ public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+ String username = null;
+ for (Callback callback : callbacks) {
+ if (callback instanceof NameCallback)
+ username = ((NameCallback) callback).getDefaultName();
+ else if (callback instanceof ScramCredentialCallback)
+ ((ScramCredentialCallback) callback).scramCredential(credentialCache.get(username));
+ else
+ throw new UnsupportedCallbackException(callback);
+ }
+ }
+
+ @Override
+ public void configure(Map<String, ?> configs, Mode mode, Subject subject, String saslMechanism) {
+ }
+
+ @Override
+ public void close() {
+ }
+}
\ No newline at end of file