You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rs...@apache.org on 2017/09/14 09:16:19 UTC
[1/2] kafka git commit: KAFKA-5783;
Add KafkaPrincipalBuilder with support for SASL (KIP-189)
Repository: kafka
Updated Branches:
refs/heads/trunk f7b1add68 -> 3b5d88feb
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java
index 52a34fc..72e06a2 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java
@@ -21,7 +21,6 @@ import java.util.Map;
import org.apache.kafka.common.security.JaasContext;
import org.apache.kafka.common.security.auth.AuthCallbackHandler;
-import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,7 +30,6 @@ import javax.security.auth.callback.UnsupportedCallbackException;
import javax.security.sasl.AuthorizeCallback;
import javax.security.sasl.RealmCallback;
-import org.apache.kafka.common.security.kerberos.KerberosName;
import org.apache.kafka.common.network.Mode;
/**
@@ -42,12 +40,10 @@ import org.apache.kafka.common.network.Mode;
*/
public class SaslServerCallbackHandler implements AuthCallbackHandler {
private static final Logger LOG = LoggerFactory.getLogger(SaslServerCallbackHandler.class);
- private final KerberosShortNamer kerberosShortNamer;
private final JaasContext jaasContext;
- public SaslServerCallbackHandler(JaasContext jaasContext, KerberosShortNamer kerberosNameParser) throws IOException {
+ public SaslServerCallbackHandler(JaasContext jaasContext) throws IOException {
this.jaasContext = jaasContext;
- this.kerberosShortNamer = kerberosNameParser;
}
@Override
@@ -77,19 +73,10 @@ public class SaslServerCallbackHandler implements AuthCallbackHandler {
private void handleAuthorizeCallback(AuthorizeCallback ac) {
String authenticationID = ac.getAuthenticationID();
String authorizationID = ac.getAuthorizationID();
-
- LOG.info("Successfully authenticated client: authenticationID={}; authorizationID={}.", authenticationID,
- authorizationID);
+ LOG.info("Successfully authenticated client: authenticationID={}; authorizationID={}.",
+ authenticationID, authorizationID);
ac.setAuthorized(true);
-
- KerberosName kerberosName = KerberosName.parse(authenticationID);
- try {
- String userName = kerberosShortNamer.shortName(kerberosName);
- LOG.info("Setting authorizedID: {}", userName);
- ac.setAuthorizedID(userName);
- } catch (IOException e) {
- LOG.error("Failed to set name for '{}' based on Kerberos authentication rules.", kerberosName, e);
- }
+ ac.setAuthorizedID(authenticationID);
}
@Override
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
index c7905bb..6b2fe74 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.security.ssl;
import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.config.types.Password;
@@ -97,7 +98,7 @@ public class SslFactory implements Configurable {
String clientAuthConfig = clientAuthConfigOverride;
if (clientAuthConfig == null)
- clientAuthConfig = (String) configs.get(SslConfigs.SSL_CLIENT_AUTH_CONFIG);
+ clientAuthConfig = (String) configs.get(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG);
if (clientAuthConfig != null) {
if (clientAuthConfig.equals("required"))
this.needClientAuth = true;
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java b/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java
new file mode 100644
index 0000000..9c9bd44
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java
@@ -0,0 +1,37 @@
+/*
+ * 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.utils;
+
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+
+public class SecurityUtils {
+
+ public static KafkaPrincipal parseKafkaPrincipal(String str) {
+ if (str == null || str.isEmpty()) {
+ throw new IllegalArgumentException("expected a string in format principalType:principalName but got " + str);
+ }
+
+ String[] split = str.split(":", 2);
+
+ if (split.length != 2) {
+ throw new IllegalArgumentException("expected a string in format principalType:principalName but got " + str);
+ }
+
+ return new KafkaPrincipal(split[0], split[1]);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 2025107..1137045 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.common.utils;
-import java.text.DecimalFormat;
import org.apache.kafka.common.KafkaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,6 +44,7 @@ import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.StandardCopyOption;
import java.nio.file.attribute.BasicFileAttributes;
+import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -719,7 +719,7 @@ public class Utils {
try {
closeable.close();
} catch (Throwable t) {
- log.warn("Failed to close {}", name, t);
+ log.warn("Failed to close {} with type {}", name, closeable.getClass().getName(), t);
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/CertStores.java b/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
index b6cc1d4..916e619 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
@@ -16,14 +16,14 @@
*/
package org.apache.kafka.common.network;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.test.TestSslUtils;
+
import java.io.File;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;
-import org.apache.kafka.common.config.SslConfigs;
-import org.apache.kafka.test.TestSslUtils;
-
public class CertStores {
private final Map<String, Object> sslConfig;
@@ -45,8 +45,6 @@ public class CertStores {
Mode mode = server ? Mode.SERVER : Mode.CLIENT;
File truststoreFile = File.createTempFile(name + "TS", ".jks");
sslConfig = TestSslUtils.createSslConfig(!server, true, mode, truststoreFile, name, commonName, certBuilder);
- if (server)
- sslConfig.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
}
public Map<String, Object> getTrustingConfig(CertStores truststoreConfig) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java b/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java
new file mode 100644
index 0000000..de210e7
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.network;
+
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.security.auth.AuthenticationContext;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
+import org.apache.kafka.common.security.auth.PlaintextAuthenticationContext;
+import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.security.Principal;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ChannelBuildersTest {
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testCreateOldPrincipalBuilder() throws Exception {
+ TransportLayer transportLayer = EasyMock.mock(TransportLayer.class);
+ Authenticator authenticator = EasyMock.mock(Authenticator.class);
+
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, OldPrincipalBuilder.class);
+ KafkaPrincipalBuilder builder = ChannelBuilders.createPrincipalBuilder(configs, transportLayer, authenticator, null);
+
+ // test old principal builder is properly configured and delegated to
+ assertTrue(OldPrincipalBuilder.configured);
+
+ // test delegation
+ KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost()));
+ assertEquals(OldPrincipalBuilder.PRINCIPAL_NAME, principal.getName());
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ }
+
+ @Test
+ public void testCreateConfigurableKafkaPrincipalBuilder() {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, ConfigurableKafkaPrincipalBuilder.class);
+ KafkaPrincipalBuilder builder = ChannelBuilders.createPrincipalBuilder(configs, null, null, null);
+ assertTrue(builder instanceof ConfigurableKafkaPrincipalBuilder);
+ assertTrue(((ConfigurableKafkaPrincipalBuilder) builder).configured);
+ }
+
+ @SuppressWarnings("deprecation")
+ public static class OldPrincipalBuilder implements PrincipalBuilder {
+ private static boolean configured = false;
+ private static final String PRINCIPAL_NAME = "bob";
+
+ @Override
+ public void configure(Map<String, ?> configs) {
+ configured = true;
+ }
+
+ @Override
+ public Principal buildPrincipal(TransportLayer transportLayer, Authenticator authenticator) throws KafkaException {
+ return new Principal() {
+ @Override
+ public String getName() {
+ return PRINCIPAL_NAME;
+ }
+ };
+ }
+
+ @Override
+ public void close() throws KafkaException {
+
+ }
+ }
+
+ public static class ConfigurableKafkaPrincipalBuilder implements KafkaPrincipalBuilder, Configurable {
+ private boolean configured = false;
+
+ @Override
+ public void configure(Map<String, ?> configs) {
+ configured = true;
+ }
+
+ @Override
+ public KafkaPrincipal build(AuthenticationContext context) {
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
index 5683068..35f1377 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -16,27 +16,9 @@
*/
package org.apache.kafka.common.network;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.nio.channels.SelectionKey;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.io.File;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-
import org.apache.kafka.common.memory.MemoryPool;
import org.apache.kafka.common.memory.SimpleMemoryPool;
import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.security.ssl.SslFactory;
import org.apache.kafka.common.utils.LogContext;
@@ -46,6 +28,23 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
/**
* A set of tests for the selector. These use a test harness that runs a simple socket server that echos back responses.
*/
@@ -58,7 +57,6 @@ public class SslSelectorTest extends SelectorTest {
File trustStoreFile = File.createTempFile("truststore", ".jks");
Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(false, true, Mode.SERVER, trustStoreFile, "server");
- sslServerConfigs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
this.server = new EchoServer(SecurityProtocol.SSL, sslServerConfigs);
this.server.start();
this.time = new MockTime();
@@ -180,7 +178,6 @@ public class SslSelectorTest extends SelectorTest {
//the initial channel builder is for clients, we need a server one
File trustStoreFile = File.createTempFile("truststore", ".jks");
Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(false, true, Mode.SERVER, trustStoreFile, "server");
- sslServerConfigs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
channelBuilder = new SslChannelBuilder(Mode.SERVER);
channelBuilder.configure(sslServerConfigs);
selector = new Selector(NetworkReceive.UNLIMITED, 5000, metrics, time, "MetricGroup",
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
index 2ddd349..459a4af 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -16,34 +16,15 @@
*/
package org.apache.kafka.common.network;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.Channels;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLEngine;
-import javax.net.ssl.SSLParameters;
-
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.memory.MemoryPool;
-import org.apache.kafka.common.security.TestSecurityConfig;
-import org.apache.kafka.common.security.ssl.SslFactory;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.security.TestSecurityConfig;
+import org.apache.kafka.common.security.ssl.SslFactory;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
@@ -53,6 +34,25 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
/**
* Tests for the SSL transport layer. These use a test harness that runs a simple socket server that echos back responses.
*/
@@ -260,7 +260,7 @@ public class SslTransportLayerTest {
@Test
public void testClientAuthenticationRequiredValidProvided() throws Exception {
String node = "0";
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
server = createEchoServer(SecurityProtocol.SSL);
createSelector(sslClientConfigs);
InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
@@ -276,8 +276,8 @@ public class SslTransportLayerTest {
public void testListenerConfigOverride() throws Exception {
String node = "0";
ListenerName clientListenerName = new ListenerName("client");
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
- sslServerConfigs.put(clientListenerName.configPrefix() + SslConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
+ sslServerConfigs.put(clientListenerName.configPrefix() + BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
// `client` listener is not configured at this point, so client auth should be required
server = createEchoServer(SecurityProtocol.SSL);
@@ -317,7 +317,7 @@ public class SslTransportLayerTest {
public void testClientAuthenticationRequiredUntrustedProvided() throws Exception {
String node = "0";
sslServerConfigs = serverCertStores.getUntrustingConfig();
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
server = createEchoServer(SecurityProtocol.SSL);
createSelector(sslClientConfigs);
InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
@@ -333,7 +333,7 @@ public class SslTransportLayerTest {
@Test
public void testClientAuthenticationRequiredNotProvided() throws Exception {
String node = "0";
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required");
server = createEchoServer(SecurityProtocol.SSL);
sslClientConfigs.remove(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG);
@@ -354,7 +354,7 @@ public class SslTransportLayerTest {
public void testClientAuthenticationDisabledUntrustedProvided() throws Exception {
String node = "0";
sslServerConfigs = serverCertStores.getUntrustingConfig();
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
server = createEchoServer(SecurityProtocol.SSL);
createSelector(sslClientConfigs);
InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
@@ -370,7 +370,7 @@ public class SslTransportLayerTest {
@Test
public void testClientAuthenticationDisabledNotProvided() throws Exception {
String node = "0";
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
server = createEchoServer(SecurityProtocol.SSL);
sslClientConfigs.remove(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG);
@@ -390,7 +390,7 @@ public class SslTransportLayerTest {
@Test
public void testClientAuthenticationRequestedValidProvided() throws Exception {
String node = "0";
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "requested");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "requested");
server = createEchoServer(SecurityProtocol.SSL);
createSelector(sslClientConfigs);
InetSocketAddress addr = new InetSocketAddress("localhost", server.port());
@@ -406,7 +406,7 @@ public class SslTransportLayerTest {
@Test
public void testClientAuthenticationRequestedNotProvided() throws Exception {
String node = "0";
- sslServerConfigs.put(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "requested");
+ sslServerConfigs.put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "requested");
server = createEchoServer(SecurityProtocol.SSL);
sslClientConfigs.remove(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG);
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java b/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java
index 4cd5f5d..05294cf 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java
@@ -20,19 +20,19 @@ import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
-import org.apache.kafka.common.config.SaslConfigs;
-import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import java.util.Map;
public class TestSecurityConfig extends AbstractConfig {
private static final ConfigDef CONFIG = new ConfigDef()
- .define(SslConfigs.SSL_CLIENT_AUTH_CONFIG, Type.STRING, null, Importance.MEDIUM,
- SslConfigs.SSL_CLIENT_AUTH_DOC)
- .define(SaslConfigs.SASL_ENABLED_MECHANISMS, Type.LIST, SaslConfigs.DEFAULT_SASL_ENABLED_MECHANISMS,
- Importance.MEDIUM, SaslConfigs.SASL_ENABLED_MECHANISMS_DOC)
- .define(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Type.CLASS, SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS,
- Importance.MEDIUM, SslConfigs.PRINCIPAL_BUILDER_CLASS_DOC)
+ .define(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, Type.STRING, null, Importance.MEDIUM,
+ BrokerSecurityConfigs.SSL_CLIENT_AUTH_DOC)
+ .define(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Type.LIST,
+ BrokerSecurityConfigs.DEFAULT_SASL_ENABLED_MECHANISMS,
+ Importance.MEDIUM, BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_DOC)
+ .define(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Type.CLASS,
+ null, Importance.MEDIUM, BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_DOC)
.withClientSslSupport()
.withClientSaslSupport();
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/security/auth/DefaultKafkaPrincipalBuilderTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/auth/DefaultKafkaPrincipalBuilderTest.java b/clients/src/test/java/org/apache/kafka/common/security/auth/DefaultKafkaPrincipalBuilderTest.java
new file mode 100644
index 0000000..fdf9e3c
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/auth/DefaultKafkaPrincipalBuilderTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.auth;
+
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.network.Authenticator;
+import org.apache.kafka.common.network.TransportLayer;
+import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
+import org.apache.kafka.common.security.kerberos.KerberosName;
+import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
+import org.apache.kafka.common.security.scram.ScramMechanism;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockSupport;
+import org.junit.Test;
+
+import javax.net.ssl.SSLSession;
+import javax.security.sasl.SaslServer;
+import java.net.InetAddress;
+import java.security.Principal;
+
+import static org.junit.Assert.assertEquals;
+
+public class DefaultKafkaPrincipalBuilderTest extends EasyMockSupport {
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception {
+ TransportLayer transportLayer = mock(TransportLayer.class);
+ Authenticator authenticator = mock(Authenticator.class);
+ PrincipalBuilder oldPrincipalBuilder = mock(PrincipalBuilder.class);
+
+ EasyMock.expect(oldPrincipalBuilder.buildPrincipal(transportLayer, authenticator))
+ .andReturn(new DummyPrincipal("foo"));
+ oldPrincipalBuilder.close();
+ EasyMock.expectLastCall();
+
+ replayAll();
+
+ DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator,
+ transportLayer, oldPrincipalBuilder, null);
+
+ KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost()));
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals("foo", principal.getName());
+
+ builder.close();
+
+ verifyAll();
+ }
+
+ @Test
+ public void testReturnAnonymousPrincipalForPlaintext() throws Exception {
+ DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null);
+ assertEquals(KafkaPrincipal.ANONYMOUS, builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost())));
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ public void testUseOldPrincipalBuilderForSslIfProvided() throws Exception {
+ TransportLayer transportLayer = mock(TransportLayer.class);
+ Authenticator authenticator = mock(Authenticator.class);
+ PrincipalBuilder oldPrincipalBuilder = mock(PrincipalBuilder.class);
+ SSLSession session = mock(SSLSession.class);
+
+ EasyMock.expect(oldPrincipalBuilder.buildPrincipal(transportLayer, authenticator))
+ .andReturn(new DummyPrincipal("foo"));
+ oldPrincipalBuilder.close();
+ EasyMock.expectLastCall();
+
+ replayAll();
+
+ DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator,
+ transportLayer, oldPrincipalBuilder, null);
+
+ KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost()));
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals("foo", principal.getName());
+
+ builder.close();
+
+ verifyAll();
+ }
+
+ @Test
+ public void testUseSessionPeerPrincipalForSsl() throws Exception {
+ SSLSession session = mock(SSLSession.class);
+
+ EasyMock.expect(session.getPeerPrincipal()).andReturn(new DummyPrincipal("foo"));
+
+ replayAll();
+
+ DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null);
+
+ KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost()));
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals("foo", principal.getName());
+
+ verifyAll();
+ }
+
+ @Test
+ public void testPrincipalBuilderScram() throws Exception {
+ SaslServer server = mock(SaslServer.class);
+
+ EasyMock.expect(server.getMechanismName()).andReturn(ScramMechanism.SCRAM_SHA_256.mechanismName());
+ EasyMock.expect(server.getAuthorizationID()).andReturn("foo");
+
+ replayAll();
+
+ DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null);
+
+ KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server,
+ SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost()));
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals("foo", principal.getName());
+
+ verifyAll();
+ }
+
+ @Test
+ public void testPrincipalBuilderGssapi() throws Exception {
+ SaslServer server = mock(SaslServer.class);
+ KerberosShortNamer kerberosShortNamer = mock(KerberosShortNamer.class);
+
+ EasyMock.expect(server.getMechanismName()).andReturn(SaslConfigs.GSSAPI_MECHANISM);
+ EasyMock.expect(server.getAuthorizationID()).andReturn("foo/host@REALM.COM");
+ EasyMock.expect(kerberosShortNamer.shortName(EasyMock.anyObject(KerberosName.class)))
+ .andReturn("foo");
+
+ replayAll();
+
+ DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(kerberosShortNamer);
+
+ KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server,
+ SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost()));
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals("foo", principal.getName());
+
+ verifyAll();
+ }
+
+ private static class DummyPrincipal implements Principal {
+ private final String name;
+
+ private DummyPrincipal(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/security/auth/KafkaPrincipalTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/auth/KafkaPrincipalTest.java b/clients/src/test/java/org/apache/kafka/common/security/auth/KafkaPrincipalTest.java
index db905dd..f204a88 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/auth/KafkaPrincipalTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/auth/KafkaPrincipalTest.java
@@ -22,15 +22,6 @@ import org.junit.Test;
public class KafkaPrincipalTest {
@Test
- public void testPrincipalNameCanContainSeparator() {
- String name = "name" + KafkaPrincipal.SEPARATOR + "with" + KafkaPrincipal.SEPARATOR + "in" + KafkaPrincipal.SEPARATOR + "it";
-
- KafkaPrincipal principal = KafkaPrincipal.fromString(KafkaPrincipal.USER_TYPE + KafkaPrincipal.SEPARATOR + name);
- Assert.assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
- Assert.assertEquals(name, principal.getName());
- }
-
- @Test
public void testEqualsAndHashCode() {
String name = "KafkaUser";
KafkaPrincipal principal1 = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, name);
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 6b0eca3..2dd7db9 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,6 +19,7 @@ package org.apache.kafka.common.security.authenticator;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.network.CertStores;
import org.apache.kafka.common.network.ChannelBuilder;
@@ -634,7 +635,7 @@ public class SaslAuthenticatorTest {
public void testDynamicJaasConfiguration() throws Exception {
SecurityProtocol securityProtocol = SecurityProtocol.SASL_SSL;
saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
- saslServerConfigs.put(SaslConfigs.SASL_ENABLED_MECHANISMS, Arrays.asList("PLAIN"));
+ saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("PLAIN"));
Map<String, Object> serverOptions = new HashMap<>();
serverOptions.put("user_user1", "user1-secret");
serverOptions.put("user_user2", "user2-secret");
@@ -676,7 +677,7 @@ public class SaslAuthenticatorTest {
public void testJaasConfigurationForListener() throws Exception {
SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT;
saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
- saslServerConfigs.put(SaslConfigs.SASL_ENABLED_MECHANISMS, Arrays.asList("PLAIN"));
+ saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("PLAIN"));
TestJaasConfig staticJaasConfig = new TestJaasConfig();
@@ -782,7 +783,7 @@ public class SaslAuthenticatorTest {
private TestJaasConfig configureMechanisms(String clientMechanism, List<String> serverMechanisms) {
saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, clientMechanism);
- saslServerConfigs.put(SaslConfigs.SASL_ENABLED_MECHANISMS, serverMechanisms);
+ saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, serverMechanisms);
return TestJaasConfig.createConfiguration(clientMechanism, serverMechanisms);
}
@@ -860,7 +861,7 @@ public class SaslAuthenticatorTest {
@SuppressWarnings("unchecked")
private void updateScramCredentialCache(String username, String password) throws NoSuchAlgorithmException {
- for (String mechanism : (List<String>) saslServerConfigs.get(SaslConfigs.SASL_ENABLED_MECHANISMS)) {
+ for (String mechanism : (List<String>) saslServerConfigs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG)) {
ScramMechanism scramMechanism = ScramMechanism.forMechanismName(mechanism);
if (scramMechanism != null) {
ScramFormatter formatter = new ScramFormatter(scramMechanism);
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
index d37c206..9d8be8d 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.common.security.authenticator;
-import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.errors.IllegalSaslStateException;
import org.apache.kafka.common.network.InvalidReceiveException;
import org.apache.kafka.common.network.ListenerName;
@@ -26,7 +26,6 @@ import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.security.JaasContext;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
import org.apache.kafka.common.security.plain.PlainLoginModule;
import org.easymock.Capture;
import org.easymock.EasyMock;
@@ -35,7 +34,6 @@ import org.junit.Test;
import javax.security.auth.Subject;
import java.io.IOException;
-import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
@@ -49,11 +47,10 @@ public class SaslServerAuthenticatorTest {
@Test(expected = InvalidReceiveException.class)
public void testOversizeRequest() throws IOException {
- SaslServerAuthenticator authenticator = setupAuthenticator();
TransportLayer transportLayer = EasyMock.mock(TransportLayer.class);
- PrincipalBuilder principalBuilder = null; // SASL authenticator does not currently use principal builder
- Map<String, ?> configs = Collections.singletonMap(SaslConfigs.SASL_ENABLED_MECHANISMS,
+ Map<String, ?> configs = Collections.singletonMap(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG,
Collections.singletonList(SCRAM_SHA_256.mechanismName()));
+ SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer);
final Capture<ByteBuffer> size = EasyMock.newCapture();
EasyMock.expect(transportLayer.read(EasyMock.capture(size))).andAnswer(new IAnswer<Integer>() {
@@ -66,17 +63,15 @@ public class SaslServerAuthenticatorTest {
EasyMock.replay(transportLayer);
- authenticator.configure(transportLayer, principalBuilder, configs);
authenticator.authenticate();
}
@Test
public void testUnexpectedRequestType() throws IOException {
- SaslServerAuthenticator authenticator = setupAuthenticator();
TransportLayer transportLayer = EasyMock.mock(TransportLayer.class);
- PrincipalBuilder principalBuilder = null; // SASL authenticator does not currently use principal builder
- Map<String, ?> configs = Collections.singletonMap(SaslConfigs.SASL_ENABLED_MECHANISMS,
+ Map<String, ?> configs = Collections.singletonMap(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG,
Collections.singletonList(SCRAM_SHA_256.mechanismName()));
+ SaslServerAuthenticator authenticator = setupAuthenticator(configs, transportLayer);
final RequestHeader header = new RequestHeader(ApiKeys.METADATA, (short) 0, "clientId", 13243);
final Struct headerStruct = header.toStruct();
@@ -102,7 +97,6 @@ public class SaslServerAuthenticatorTest {
EasyMock.replay(transportLayer);
- authenticator.configure(transportLayer, principalBuilder, configs);
try {
authenticator.authenticate();
fail("Expected authenticate() to raise an exception");
@@ -111,13 +105,13 @@ public class SaslServerAuthenticatorTest {
}
}
- private SaslServerAuthenticator setupAuthenticator() throws IOException {
+ private SaslServerAuthenticator setupAuthenticator(Map<String, ?> configs, TransportLayer transportLayer) throws IOException {
TestJaasConfig jaasConfig = new TestJaasConfig();
jaasConfig.addEntry("jaasContext", PlainLoginModule.class.getName(), new HashMap<String, Object>());
JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig);
Subject subject = new Subject();
- return new SaslServerAuthenticator("node", jaasContext, subject, null, InetAddress.getLocalHost(),
- new CredentialCache(), new ListenerName("ssl"), SecurityProtocol.SASL_SSL);
+ return new SaslServerAuthenticator(configs, "node", jaasContext, subject, null, new CredentialCache(),
+ new ListenerName("ssl"), SecurityProtocol.SASL_SSL, transportLayer);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/test/java/org/apache/kafka/common/utils/SecurityUtilsTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/SecurityUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/SecurityUtilsTest.java
new file mode 100644
index 0000000..273c13a
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/utils/SecurityUtilsTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.utils;
+
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class SecurityUtilsTest {
+
+ @Test
+ public void testPrincipalNameCanContainSeparator() {
+ String name = "name:with:separator:in:it";
+ KafkaPrincipal principal = SecurityUtils.parseKafkaPrincipal(KafkaPrincipal.USER_TYPE + ":" + name);
+ assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType());
+ assertEquals(name, principal.getName());
+ }
+
+ @Test
+ public void testParseKafkaPrincipalWithNonUserPrincipalType() {
+ String name = "foo";
+ String principalType = "Group";
+ KafkaPrincipal principal = SecurityUtils.parseKafkaPrincipal(principalType + ":" + name);
+ assertEquals(principalType, principal.getPrincipalType());
+ assertEquals(name, principal.getName());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 ffe75a8..de2bcde 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -31,14 +31,11 @@ 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.memory.{MemoryPool, SimpleMemoryPool}
import org.apache.kafka.common.metrics._
import org.apache.kafka.common.metrics.stats.Rate
import org.apache.kafka.common.network.{ChannelBuilder, ChannelBuilders, KafkaChannel, ListenerName, Selectable, Send, Selector => KSelector}
import org.apache.kafka.common.protocol.SecurityProtocol
-import org.apache.kafka.common.security.auth.KafkaPrincipal
-import org.apache.kafka.common.protocol.types.SchemaException
import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
import org.apache.kafka.common.utils.{KafkaThread, LogContext, Time}
@@ -547,10 +544,9 @@ private[kafka] class Processor(val id: Int,
try {
openOrClosingChannel(receive.source) match {
case Some(channel) =>
- val principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName)
val header = RequestHeader.parse(receive.payload)
val context = new RequestContext(header, receive.source, channel.socketAddress,
- principal, listenerName, securityProtocol)
+ channel.principal, listenerName, securityProtocol)
val req = new RequestChannel.Request(processor = id, context = context,
startTimeNanos = time.nanoseconds, memoryPool, receive.payload)
requestChannel.sendRequest(req)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index de6559c..33eaf48 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -29,6 +29,7 @@ import kafka.utils.CoreUtils
import kafka.utils.Implicits._
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.common.config.ConfigDef.ValidList
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, SaslConfigs, SslConfigs, TopicConfig}
import org.apache.kafka.common.metrics.Sensor
import org.apache.kafka.common.network.ListenerName
@@ -193,7 +194,6 @@ object Defaults {
val MetricRecordingLevel = Sensor.RecordingLevel.INFO.toString()
/** ********* SSL configuration ***********/
- val PrincipalBuilderClass = SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS
val SslProtocol = SslConfigs.DEFAULT_SSL_PROTOCOL
val SslEnabledProtocols = SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS
val SslKeystoreType = SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE
@@ -212,7 +212,7 @@ object Defaults {
val SaslKerberosTicketRenewWindowFactor = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR
val SaslKerberosTicketRenewJitter = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER
val SaslKerberosMinTimeBeforeRelogin = SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN
- val SaslKerberosPrincipalToLocalRules = SaslConfigs.DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES
+ val SaslKerberosPrincipalToLocalRules = BrokerSecurityConfigs.DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES
}
object KafkaConfig {
@@ -375,8 +375,10 @@ object KafkaConfig {
val MetricReporterClassesProp: String = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG
val MetricRecordingLevelProp: String = CommonClientConfigs.METRICS_RECORDING_LEVEL_CONFIG
+ /** ******** Common Security Configuration *************/
+ val PrincipalBuilderClassProp = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG
+
/** ********* SSL Configuration ****************/
- val PrincipalBuilderClassProp = SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG
val SslProtocolProp = SslConfigs.SSL_PROTOCOL_CONFIG
val SslProviderProp = SslConfigs.SSL_PROVIDER_CONFIG
val SslCipherSuitesProp = SslConfigs.SSL_CIPHER_SUITES_CONFIG
@@ -392,17 +394,17 @@ object KafkaConfig {
val SslTrustManagerAlgorithmProp = SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG
val SslEndpointIdentificationAlgorithmProp = SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG
val SslSecureRandomImplementationProp = SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG
- val SslClientAuthProp = SslConfigs.SSL_CLIENT_AUTH_CONFIG
+ val SslClientAuthProp = BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG
/** ********* SASL Configuration ****************/
val SaslMechanismInterBrokerProtocolProp = "sasl.mechanism.inter.broker.protocol"
- val SaslEnabledMechanismsProp = SaslConfigs.SASL_ENABLED_MECHANISMS
+ val SaslEnabledMechanismsProp = BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG
val SaslKerberosServiceNameProp = SaslConfigs.SASL_KERBEROS_SERVICE_NAME
val SaslKerberosKinitCmdProp = SaslConfigs.SASL_KERBEROS_KINIT_CMD
val SaslKerberosTicketRenewWindowFactorProp = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR
val SaslKerberosTicketRenewJitterProp = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER
val SaslKerberosMinTimeBeforeReloginProp = SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN
- val SaslKerberosPrincipalToLocalRulesProp = SaslConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES
+ val SaslKerberosPrincipalToLocalRulesProp = BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG
/* Documentation */
/** ********* Zookeeper Configuration ***********/
@@ -636,8 +638,10 @@ object KafkaConfig {
val MetricReporterClassesDoc = CommonClientConfigs.METRIC_REPORTER_CLASSES_DOC
val MetricRecordingLevelDoc = CommonClientConfigs.METRICS_RECORDING_LEVEL_DOC
+ /** ******** Common Security Configuration *************/
+ val PrincipalBuilderClassDoc = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_DOC
+
/** ********* SSL Configuration ****************/
- val PrincipalBuilderClassDoc = SslConfigs.PRINCIPAL_BUILDER_CLASS_DOC
val SslProtocolDoc = SslConfigs.SSL_PROTOCOL_DOC
val SslProviderDoc = SslConfigs.SSL_PROVIDER_DOC
val SslCipherSuitesDoc = SslConfigs.SSL_CIPHER_SUITES_DOC
@@ -653,7 +657,7 @@ object KafkaConfig {
val SslTrustManagerAlgorithmDoc = SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_DOC
val SslEndpointIdentificationAlgorithmDoc = SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC
val SslSecureRandomImplementationDoc = SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_DOC
- val SslClientAuthDoc = SslConfigs.SSL_CLIENT_AUTH_DOC
+ val SslClientAuthDoc = BrokerSecurityConfigs.SSL_CLIENT_AUTH_DOC
/** ********* Sasl Configuration ****************/
val SaslMechanismInterBrokerProtocolDoc = "SASL mechanism used for inter-broker communication. Default is GSSAPI."
@@ -663,7 +667,7 @@ object KafkaConfig {
val SaslKerberosTicketRenewWindowFactorDoc = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC
val SaslKerberosTicketRenewJitterDoc = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC
val SaslKerberosMinTimeBeforeReloginDoc = SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC
- val SaslKerberosPrincipalToLocalRulesDoc = SaslConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC
+ val SaslKerberosPrincipalToLocalRulesDoc = BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC
private val configDef = {
import ConfigDef.Importance._
@@ -834,7 +838,7 @@ object KafkaConfig {
.define(ReplicationQuotaWindowSizeSecondsProp, INT, Defaults.ReplicationQuotaWindowSizeSeconds, atLeast(1), LOW, ReplicationQuotaWindowSizeSecondsDoc)
/** ********* SSL Configuration ****************/
- .define(PrincipalBuilderClassProp, CLASS, Defaults.PrincipalBuilderClass, MEDIUM, PrincipalBuilderClassDoc)
+ .define(PrincipalBuilderClassProp, CLASS, null, MEDIUM, PrincipalBuilderClassDoc)
.define(SslProtocolProp, STRING, Defaults.SslProtocol, MEDIUM, SslProtocolDoc)
.define(SslProviderProp, STRING, null, MEDIUM, SslProviderDoc)
.define(SslEnabledProtocolsProp, LIST, Defaults.SslEnabledProtocols, MEDIUM, SslEnabledProtocolsDoc)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 3376d23..bbb3249 100644
--- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala
@@ -69,7 +69,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
val topicWildcard = "*"
val part = 0
val tp = new TopicPartition(topic, part)
- val topicAndPartition = new TopicAndPartition(topic, part)
+ val topicAndPartition = TopicAndPartition(topic, part)
val clientPrincipal: String
val kafkaPrincipal: String
@@ -154,8 +154,8 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
* Starts MiniKDC and only then sets up the parent trait.
*/
@Before
- override def setUp {
- super.setUp
+ override def setUp() {
+ super.setUp()
AclCommand.main(topicBrokerReadAclArgs)
servers.foreach { s =>
TestUtils.waitAndVerifyAcls(TopicBrokerReadAcl, s.apis.authorizer.get, new Resource(Topic, "*"))
@@ -177,9 +177,9 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
* Closes MiniKDC last when tearing down.
*/
@After
- override def tearDown {
+ override def tearDown() {
consumers.foreach(_.wakeup())
- super.tearDown
+ super.tearDown()
closeSasl()
}
@@ -240,7 +240,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
*/
@Test(expected = classOf[KafkaException])
def testNoConsumeWithoutDescribeAclViaAssign(): Unit = {
- noConsumeWithoutDescribeAclSetup
+ noConsumeWithoutDescribeAclSetup()
consumers.head.assign(List(tp).asJava)
// the exception is expected when the consumer attempts to lookup offsets
consumeRecords(this.consumers.head)
@@ -248,7 +248,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
@Test(expected = classOf[TimeoutException])
def testNoConsumeWithoutDescribeAclViaSubscribe(): Unit = {
- noConsumeWithoutDescribeAclSetup
+ noConsumeWithoutDescribeAclSetup()
consumers.head.subscribe(List(topic).asJava)
// this should timeout since the consumer will not be able to fetch any metadata for the topic
consumeRecords(this.consumers.head, timeout = 3000)
@@ -273,7 +273,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
@Test
def testNoConsumeWithDescribeAclViaAssign(): Unit = {
- noConsumeWithDescribeAclSetup
+ noConsumeWithDescribeAclSetup()
consumers.head.assign(List(tp).asJava)
try {
@@ -287,7 +287,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas
@Test
def testNoConsumeWithDescribeAclViaSubscribe(): Unit = {
- noConsumeWithDescribeAclSetup
+ noConsumeWithDescribeAclSetup()
consumers.head.subscribe(List(topic).asJava)
try {
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
index b8dc57b..4b27239 100644
--- a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
+++ b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
@@ -26,6 +26,7 @@ import java.util.Properties
import org.apache.kafka.clients.producer.KafkaProducer
import kafka.server.KafkaConfig
import kafka.integration.KafkaServerTestHarness
+import org.apache.kafka.common.network.ListenerName
import org.junit.{After, Before}
import scala.collection.mutable.Buffer
@@ -46,14 +47,21 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness {
val consumers = Buffer[KafkaConsumer[Array[Byte], Array[Byte]]]()
val producers = Buffer[KafkaProducer[Array[Byte], Array[Byte]]]()
+ protected def interBrokerListenerName: ListenerName = listenerName
+
override def generateConfigs = {
val cfgs = TestUtils.createBrokerConfigs(serverCount, zkConnect, interBrokerSecurityProtocol = Some(securityProtocol),
trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount)
cfgs.foreach { config =>
- config.setProperty(KafkaConfig.ListenersProp, s"${listenerName.value}://localhost:${TestUtils.RandomPort}")
config.remove(KafkaConfig.InterBrokerSecurityProtocolProp)
- config.setProperty(KafkaConfig.InterBrokerListenerNameProp, listenerName.value)
- config.setProperty(KafkaConfig.ListenerSecurityProtocolMapProp, s"${listenerName.value}:${securityProtocol.name}")
+ config.setProperty(KafkaConfig.InterBrokerListenerNameProp, interBrokerListenerName.value)
+
+ val listenerNames = Set(listenerName, interBrokerListenerName)
+ val listeners = listenerNames.map(listenerName => s"${listenerName.value}://localhost:${TestUtils.RandomPort}").mkString(",")
+ val listenerSecurityMap = listenerNames.map(listenerName => s"${listenerName.value}:${securityProtocol.name}").mkString(",")
+
+ config.setProperty(KafkaConfig.ListenersProp, listeners)
+ config.setProperty(KafkaConfig.ListenerSecurityProtocolMapProp, listenerSecurityMap)
}
cfgs.foreach(_ ++= serverConfig)
cfgs.map(KafkaConfig.fromProps)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala
new file mode 100644
index 0000000..2aeccb4
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala
@@ -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 kafka.api
+
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder, PlaintextAuthenticationContext}
+import org.junit.Before
+
+// This test case uses a separate listener for client and inter-broker communication, from
+// which we derive corresponding principals
+object PlaintextEndToEndAuthorizationTest {
+ class TestClientPrincipalBuilder extends KafkaPrincipalBuilder {
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ context match {
+ case ctx: PlaintextAuthenticationContext if ctx.clientAddress != null =>
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "client")
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
+
+ class TestServerPrincipalBuilder extends KafkaPrincipalBuilder {
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ context match {
+ case ctx: PlaintextAuthenticationContext =>
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "server")
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
+}
+
+class PlaintextEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
+ import PlaintextEndToEndAuthorizationTest.{TestClientPrincipalBuilder, TestServerPrincipalBuilder}
+
+ override protected def securityProtocol = SecurityProtocol.PLAINTEXT
+ override protected def listenerName: ListenerName = new ListenerName("CLIENT")
+ override protected def interBrokerListenerName: ListenerName = new ListenerName("SERVER")
+
+ this.serverConfig.setProperty("listener.name.client." + BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG,
+ classOf[TestClientPrincipalBuilder].getName)
+ this.serverConfig.setProperty("listener.name.server." + BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG,
+ classOf[TestServerPrincipalBuilder].getName)
+ override val clientPrincipal = "client"
+ override val kafkaPrincipal = "server"
+
+ @Before
+ override def setUp() {
+ startSasl(jaasSections(List.empty, None, ZkSasl))
+ super.setUp()
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
index 975ca4c..c28de3e 100644
--- a/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
@@ -17,13 +17,38 @@
package kafka.api
import kafka.utils.{JaasTestUtils, TestUtils}
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
+import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder, SaslAuthenticationContext}
import org.junit.Test
+object SaslPlainSslEndToEndAuthorizationTest {
+ class TestPrincipalBuilder extends KafkaPrincipalBuilder {
+
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ context match {
+ case ctx: SaslAuthenticationContext =>
+ ctx.server.getAuthorizationID match {
+ case JaasTestUtils.KafkaPlainAdmin =>
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "admin")
+ case JaasTestUtils.KafkaPlainUser =>
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "user")
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
+ }
+}
+
class SaslPlainSslEndToEndAuthorizationTest extends SaslEndToEndAuthorizationTest {
+ import SaslPlainSslEndToEndAuthorizationTest.TestPrincipalBuilder
+
+ this.serverConfig.setProperty(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, classOf[TestPrincipalBuilder].getName)
+
override protected def kafkaClientSaslMechanism = "PLAIN"
override protected def kafkaServerSaslMechanisms = List("PLAIN")
- override val clientPrincipal = JaasTestUtils.KafkaPlainUser
- override val kafkaPrincipal = JaasTestUtils.KafkaPlainAdmin
+ override val clientPrincipal = "user"
+ override val kafkaPrincipal = "admin"
/**
* Checks that secure paths created by broker and acl paths created by AclCommand
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/integration/kafka/api/SaslSetup.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslSetup.scala b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
index 1128ad0..f874f4e 100644
--- a/core/src/test/scala/integration/kafka/api/SaslSetup.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
@@ -28,6 +28,7 @@ import kafka.utils.{JaasTestUtils, TestUtils}
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.security.authenticator.LoginManager
import org.apache.kafka.common.config.SaslConfigs
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
/*
* Implements an enumeration for the modes enabled here:
@@ -119,7 +120,7 @@ trait SaslSetup {
def kafkaServerSaslProperties(serverSaslMechanisms: Seq[String], interBrokerSaslMechanism: String): Properties = {
val props = new Properties
props.put(KafkaConfig.SaslMechanismInterBrokerProtocolProp, interBrokerSaslMechanism)
- props.put(SaslConfigs.SASL_ENABLED_MECHANISMS, serverSaslMechanisms.mkString(","))
+ props.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, serverSaslMechanisms.mkString(","))
props
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
index caa988d..8d3f9b2 100644
--- a/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
@@ -17,19 +17,42 @@
package kafka.api
-import org.apache.kafka.common.config.SslConfigs
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
import org.apache.kafka.common.protocol.SecurityProtocol
+import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder, SslAuthenticationContext}
import org.junit.Before
+object SslEndToEndAuthorizationTest {
+ class TestPrincipalBuilder extends KafkaPrincipalBuilder {
+ private val Pattern = "O=A (.*?),CN=localhost".r
+
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
+ context match {
+ case ctx: SslAuthenticationContext =>
+ ctx.session.getPeerPrincipal.getName match {
+ case Pattern(name) =>
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, name)
+ case _ =>
+ KafkaPrincipal.ANONYMOUS
+ }
+ }
+ }
+ }
+}
+
class SslEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
+ import kafka.api.SslEndToEndAuthorizationTest.TestPrincipalBuilder
+
override protected def securityProtocol = SecurityProtocol.SSL
- this.serverConfig.setProperty(SslConfigs.SSL_CLIENT_AUTH_CONFIG, "required")
- override val clientPrincipal = "O=A client,CN=localhost"
- override val kafkaPrincipal = "O=A server,CN=localhost"
+ this.serverConfig.setProperty(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "required")
+ this.serverConfig.setProperty(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, classOf[TestPrincipalBuilder].getName)
+ override val clientPrincipal = "client"
+ override val kafkaPrincipal = "server"
@Before
- override def setUp {
+ override def setUp() {
startSasl(jaasSections(List.empty, None, ZkSasl))
- super.setUp
+ super.setUp()
}
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 c005c72..1b5841d 100755
--- a/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
+++ b/core/src/test/scala/unit/kafka/integration/KafkaServerTestHarness.scala
@@ -80,7 +80,7 @@ abstract class KafkaServerTestHarness extends ZooKeeperTestHarness {
@Before
override def setUp() {
- super.setUp
+ super.setUp()
if (configs.isEmpty)
throw new KafkaException("Must supply at least one server config.")
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala b/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
index 5dbd1a8..ca17b9a 100644
--- a/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
+++ b/core/src/test/scala/unit/kafka/security/auth/SimpleAclAuthorizerTest.scala
@@ -34,7 +34,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
val simpleAclAuthorizer2 = new SimpleAclAuthorizer
val testPrincipal = Acl.WildCardPrincipal
val testHostName = InetAddress.getByName("192.168.0.1")
- val session = new Session(testPrincipal, testHostName)
+ val session = Session(testPrincipal, testHostName)
var resource: Resource = null
val superUsers = "User:superuser1; User:superuser2"
val username = "alice"
@@ -95,8 +95,8 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
changeAclAndVerify(Set.empty[Acl], acls, Set.empty[Acl])
- val host1Session = new Session(user1, host1)
- val host2Session = new Session(user1, host2)
+ val host1Session = Session(user1, host1)
+ val host2Session = Session(user1, host2)
assertTrue("User1 should have READ access from host2", simpleAclAuthorizer.authorize(host2Session, Read, resource))
assertFalse("User1 should not have READ access from host1 due to denyAcl", simpleAclAuthorizer.authorize(host1Session, Read, resource))
@@ -108,8 +108,8 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
assertFalse("User1 should not have edit access from host2", simpleAclAuthorizer.authorize(host2Session, Alter, resource))
//test if user has READ and write access they also get describe access
- val user2Session = new Session(user2, host1)
- val user3Session = new Session(user3, host1)
+ val user2Session = Session(user2, host1)
+ val user3Session = Session(user3, host1)
assertTrue("User2 should have DESCRIBE access from host1", simpleAclAuthorizer.authorize(user2Session, Describe, resource))
assertTrue("User3 should have DESCRIBE access from host2", simpleAclAuthorizer.authorize(user3Session, Describe, resource))
assertTrue("User2 should have READ access from host1", simpleAclAuthorizer.authorize(user2Session, Read, resource))
@@ -120,7 +120,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
def testDenyTakesPrecedence() {
val user = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, username)
val host = InetAddress.getByName("192.168.2.1")
- val session = new Session(user, host)
+ val session = Session(user, host)
val allowAll = Acl.AllowAllAcl
val denyAcl = new Acl(user, Deny, host.getHostAddress, All)
@@ -137,7 +137,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
changeAclAndVerify(Set.empty[Acl], Set[Acl](allowAllAcl), Set.empty[Acl])
- val session = new Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "random"), InetAddress.getByName("192.0.4.4"))
+ val session = Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "random"), InetAddress.getByName("192.0.4.4"))
assertTrue("allow all acl should allow access to all.", simpleAclAuthorizer.authorize(session, Read, resource))
}
@@ -147,8 +147,8 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
changeAclAndVerify(Set.empty[Acl], Set[Acl](denyAllAcl), Set.empty[Acl])
- val session1 = new Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "superuser1"), InetAddress.getByName("192.0.4.4"))
- val session2 = new Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "superuser2"), InetAddress.getByName("192.0.4.4"))
+ val session1 = Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "superuser1"), InetAddress.getByName("192.0.4.4"))
+ val session2 = Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "superuser2"), InetAddress.getByName("192.0.4.4"))
assertTrue("superuser always has access, no matter what acls.", simpleAclAuthorizer.authorize(session1, Read, resource))
assertTrue("superuser always has access, no matter what acls.", simpleAclAuthorizer.authorize(session2, Read, resource))
@@ -165,7 +165,7 @@ class SimpleAclAuthorizerTest extends ZooKeeperTestHarness {
val acls = changeAclAndVerify(Set.empty[Acl], Set[Acl](readAcl), Set.empty[Acl], wildCardResource)
- val host1Session = new Session(user1, host1)
+ val host1Session = Session(user1, host1)
assertTrue("User1 should have Read access from host1", simpleAclAuthorizer.authorize(host1Session, Read, resource))
//allow Write to specific topic.
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
index aec68e2..3ec03c3 100644
--- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
@@ -28,20 +28,19 @@ import org.apache.kafka.common.acl.{AccessControlEntry, AccessControlEntryFilter
import org.apache.kafka.common.resource.{ResourceFilter, Resource => AdminResource, ResourceType => AdminResourceType}
import org.apache.kafka.common.{Node, TopicPartition}
import org.apache.kafka.common.metrics.{KafkaMetric, Quota, Sensor}
-import org.apache.kafka.common.network.{Authenticator, ListenerName, TransportLayer}
+import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.{ApiKeys, SecurityProtocol}
import org.apache.kafka.common.protocol.types.Struct
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation
import org.apache.kafka.common.requests.{Resource => RResource, ResourceType => RResourceType, _}
-import org.apache.kafka.common.security.auth.{DefaultPrincipalBuilder, KafkaPrincipal}
+import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder}
import org.junit.Assert._
import org.junit.{After, Before, Test}
import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
-
class RequestQuotaTest extends BaseRequestTest {
override def numBrokers: Int = 1
@@ -85,7 +84,7 @@ class RequestQuotaTest extends BaseRequestTest {
AdminUtils.changeClientIdConfig(zkUtils, unthrottledClientId, quotaProps)
TestUtils.retry(10000) {
- val quotaManager = servers(0).apis.quotas.request
+ val quotaManager = servers.head.apis.quotas.request
assertEquals(s"Default request quota not set", Quota.upperBound(0.01), quotaManager.quota("some-user", "some-client"))
assertEquals(s"Request quota override not set", Quota.upperBound(2000), quotaManager.quota("some-user", unthrottledClientId))
}
@@ -445,8 +444,8 @@ object RequestQuotaTest {
session.principal != UnauthorizedPrincipal
}
}
- class TestPrincipalBuilder extends DefaultPrincipalBuilder {
- override def buildPrincipal(transportLayer: TransportLayer, authenticator: Authenticator) = {
+ class TestPrincipalBuilder extends KafkaPrincipalBuilder {
+ override def build(context: AuthenticationContext): KafkaPrincipal = {
principal
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 9517789..cd31f2c 100644
--- a/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/JaasTestUtils.scala
@@ -158,9 +158,9 @@ object JaasTestUtils {
kafkaClientModule(mechanism, keytabLocation, KafkaClientPrincipal, KafkaPlainUser, KafkaPlainPassword, KafkaScramUser, KafkaScramPassword).toString
def zkSections: Seq[JaasSection] = Seq(
- new JaasSection(ZkServerContextName, Seq(ZkDigestModule(debug = false,
+ JaasSection(ZkServerContextName, Seq(ZkDigestModule(debug = false,
Map("user_super" -> ZkUserSuperPasswd, s"user_$ZkUser" -> ZkUserPassword)))),
- new JaasSection(ZkClientContextName, Seq(ZkDigestModule(debug = false,
+ JaasSection(ZkClientContextName, Seq(ZkDigestModule(debug = false,
Map("username" -> ZkUser, "password" -> ZkUserPassword))))
)
@@ -191,7 +191,7 @@ object JaasTestUtils {
debug = false)
case mechanism => throw new IllegalArgumentException("Unsupported server mechanism " + mechanism)
}
- new JaasSection(contextName, modules)
+ JaasSection(contextName, modules)
}
// consider refactoring if more mechanisms are added
@@ -227,7 +227,7 @@ object JaasTestUtils {
* Used for the static JAAS configuration and it uses the credentials for client#2
*/
def kafkaClientSection(mechanism: Option[String], keytabLocation: Option[File]): JaasSection = {
- new JaasSection(KafkaClientContextName, mechanism.map(m =>
+ JaasSection(KafkaClientContextName, mechanism.map(m =>
kafkaClientModule(m, keytabLocation, KafkaClientPrincipal2, KafkaPlainUser2, KafkaPlainPassword2, KafkaScramUser2, KafkaScramPassword2)).toSeq)
}
[2/2] kafka git commit: KAFKA-5783;
Add KafkaPrincipalBuilder with support for SASL (KIP-189)
Posted by rs...@apache.org.
KAFKA-5783; Add KafkaPrincipalBuilder with support for SASL (KIP-189)
Author: Jason Gustafson <ja...@confluent.io>
Reviewers: Ismael Juma <is...@juma.me.uk>, Rajini Sivaram <ra...@googlemail.com>, Manikumar Reddy <ma...@gmail.com>
Closes #3795 from hachikuji/KAFKA-5783
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3b5d88fe
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3b5d88fe
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3b5d88fe
Branch: refs/heads/trunk
Commit: 3b5d88febb186c4b92cc514fa2338de7bc338f67
Parents: f7b1add
Author: Jason Gustafson <ja...@confluent.io>
Authored: Thu Sep 14 10:16:00 2017 +0100
Committer: Rajini Sivaram <ra...@googlemail.com>
Committed: Thu Sep 14 10:16:00 2017 +0100
----------------------------------------------------------------------
checkstyle/import-control.xml | 5 +-
.../apache/kafka/common/config/SaslConfigs.java | 44 +++--
.../apache/kafka/common/config/SslConfigs.java | 42 +++--
.../config/internals/BrokerSecurityConfigs.java | 70 ++++++++
.../kafka/common/network/Authenticator.java | 19 +--
.../kafka/common/network/ChannelBuilder.java | 3 +-
.../kafka/common/network/ChannelBuilders.java | 48 ++++--
.../common/network/DefaultAuthenticator.java | 63 -------
.../kafka/common/network/KafkaChannel.java | 12 +-
.../common/network/PlaintextChannelBuilder.java | 64 +++++--
.../common/network/SaslChannelBuilder.java | 43 ++---
.../kafka/common/network/SslChannelBuilder.java | 81 +++++++--
.../security/auth/AuthenticationContext.java | 36 ++++
.../security/auth/DefaultPrincipalBuilder.java | 6 +-
.../common/security/auth/KafkaPrincipal.java | 70 +++++---
.../security/auth/KafkaPrincipalBuilder.java | 36 ++++
.../auth/PlaintextAuthenticationContext.java | 40 +++++
.../common/security/auth/PrincipalBuilder.java | 5 +-
.../auth/SaslAuthenticationContext.java | 48 ++++++
.../security/auth/SslAuthenticationContext.java | 46 +++++
.../DefaultKafkaPrincipalBuilder.java | 149 ++++++++++++++++
.../authenticator/SaslClientAuthenticator.java | 33 ++--
.../authenticator/SaslServerAuthenticator.java | 67 +++++---
.../SaslServerCallbackHandler.java | 21 +--
.../kafka/common/security/ssl/SslFactory.java | 3 +-
.../kafka/common/utils/SecurityUtils.java | 37 ++++
.../org/apache/kafka/common/utils/Utils.java | 4 +-
.../apache/kafka/common/network/CertStores.java | 8 +-
.../common/network/ChannelBuildersTest.java | 107 ++++++++++++
.../kafka/common/network/SslSelectorTest.java | 37 ++--
.../common/network/SslTransportLayerTest.java | 62 +++----
.../common/security/TestSecurityConfig.java | 16 +-
.../auth/DefaultKafkaPrincipalBuilderTest.java | 170 +++++++++++++++++++
.../security/auth/KafkaPrincipalTest.java | 9 -
.../authenticator/SaslAuthenticatorTest.java | 9 +-
.../SaslServerAuthenticatorTest.java | 22 +--
.../kafka/common/utils/SecurityUtilsTest.java | 43 +++++
.../main/scala/kafka/network/SocketServer.scala | 6 +-
.../main/scala/kafka/server/KafkaConfig.scala | 24 +--
.../kafka/api/EndToEndAuthorizationTest.scala | 18 +-
.../kafka/api/IntegrationTestHarness.scala | 14 +-
.../PlaintextEndToEndAuthorizationTest.scala | 71 ++++++++
.../SaslPlainSslEndToEndAuthorizationTest.scala | 29 +++-
.../scala/integration/kafka/api/SaslSetup.scala | 3 +-
.../api/SslEndToEndAuthorizationTest.scala | 35 +++-
.../integration/KafkaServerTestHarness.scala | 2 +-
.../security/auth/SimpleAclAuthorizerTest.scala | 20 +--
.../unit/kafka/server/RequestQuotaTest.scala | 11 +-
.../scala/unit/kafka/utils/JaasTestUtils.scala | 8 +-
49 files changed, 1398 insertions(+), 421 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 489c004..bea7e20 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -89,16 +89,15 @@
<allow pkg="org.apache.kafka.common.annotation" />
<allow pkg="org.apache.kafka.common.network" />
<allow pkg="org.apache.kafka.common.config" />
+ <allow pkg="org.apache.kafka.common.protocol" />
+ <allow pkg="org.apache.kafka.common.errors" />
<subpackage name="authenticator">
- <allow pkg="org.apache.kafka.common.protocol" />
<allow pkg="org.apache.kafka.common.protocol.types" />
<allow pkg="org.apache.kafka.common.requests" />
- <allow pkg="org.apache.kafka.common.errors" />
<allow pkg="org.apache.kafka.clients" />
</subpackage>
<subpackage name="scram">
<allow pkg="javax.crypto" />
- <allow pkg="org.apache.kafka.common.errors" />
</subpackage>
</subpackage>
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
index 7dc5012..f61b7dd 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java
@@ -16,7 +16,8 @@
*/
package org.apache.kafka.common.config;
-import java.util.Collections;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+
import java.util.List;
public class SaslConfigs {
@@ -29,11 +30,21 @@ public class SaslConfigs {
public static final String GSSAPI_MECHANISM = "GSSAPI";
public static final String DEFAULT_SASL_MECHANISM = GSSAPI_MECHANISM;
- public static final String SASL_ENABLED_MECHANISMS = "sasl.enabled.mechanisms";
- public static final String SASL_ENABLED_MECHANISMS_DOC = "The list of SASL mechanisms enabled in the Kafka server. "
- + "The list may contain any mechanism for which a security provider is available. "
- + "Only GSSAPI is enabled by default.";
- public static final List<String> DEFAULT_SASL_ENABLED_MECHANISMS = Collections.singletonList(GSSAPI_MECHANISM);
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SASL_ENABLED_MECHANISMS = BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SASL_ENABLED_MECHANISMS_DOC = BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_DOC;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final List<String> DEFAULT_SASL_ENABLED_MECHANISMS = BrokerSecurityConfigs.DEFAULT_SASL_ENABLED_MECHANISMS;
public static final String SASL_JAAS_CONFIG = "sasl.jaas.config";
public static final String SASL_JAAS_CONFIG_DOC = "JAAS login context parameters for SASL connections in the format used by JAAS configuration files. "
@@ -61,12 +72,21 @@ public class SaslConfigs {
public static final String SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC = "Login thread sleep time between refresh attempts.";
public static final long DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN = 1 * 60 * 1000L;
- public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES = "sasl.kerberos.principal.to.local.rules";
- public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC = "A list of rules for mapping from principal names to short names (typically operating system usernames). " +
- "The rules are evaluated in order and the first rule that matches a principal name is used to map it to a short name. Any later rules in the list are ignored. " +
- "By default, principal names of the form {username}/{hostname}@{REALM} are mapped to {username}. " +
- "For more details on the format please see <a href=\"#security_authz\"> security authorization and acls</a>.";
- public static final List<String> DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES = Collections.singletonList("DEFAULT");
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES = BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC = BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final List<String> DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES = BrokerSecurityConfigs.DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES;
public static void addClientSaslSupport(ConfigDef config) {
config.define(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM, SaslConfigs.SASL_KERBEROS_SERVICE_NAME_DOC)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
index 8cae00d..5e8d304 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
@@ -16,6 +16,9 @@
*/
package org.apache.kafka.common.config;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.security.auth.DefaultPrincipalBuilder;
+
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.TrustManagerFactory;
@@ -24,10 +27,22 @@ public class SslConfigs {
* NOTE: DO NOT CHANGE EITHER CONFIG NAMES AS THESE ARE PART OF THE PUBLIC API AND CHANGE WILL BREAK USER CODE.
*/
- public static final String PRINCIPAL_BUILDER_CLASS_CONFIG = "principal.builder.class";
- public static final String PRINCIPAL_BUILDER_CLASS_DOC = "The fully qualified name of a class that implements the PrincipalBuilder interface, " +
- "which is currently used to build the Principal for connections with the SSL SecurityProtocol.";
- public static final String DEFAULT_PRINCIPAL_BUILDER_CLASS = "org.apache.kafka.common.security.auth.DefaultPrincipalBuilder";
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String PRINCIPAL_BUILDER_CLASS_CONFIG = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String PRINCIPAL_BUILDER_CLASS_DOC = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_DOC;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release. In recent versions,
+ * the config is optional and there is no default.
+ */
+ @Deprecated
+ public static final String DEFAULT_PRINCIPAL_BUILDER_CLASS = DefaultPrincipalBuilder.class.getName();
public static final String SSL_PROTOCOL_CONFIG = "ssl.protocol";
public static final String SSL_PROTOCOL_DOC = "The SSL protocol used to generate the SSLContext. "
@@ -91,15 +106,16 @@ public class SslConfigs {
public static final String SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG = "ssl.secure.random.implementation";
public static final String SSL_SECURE_RANDOM_IMPLEMENTATION_DOC = "The SecureRandom PRNG implementation to use for SSL cryptography operations. ";
- public static final String SSL_CLIENT_AUTH_CONFIG = "ssl.client.auth";
- public static final String SSL_CLIENT_AUTH_DOC = "Configures kafka broker to request client authentication."
- + " The following settings are common: "
- + " <ul>"
- + " <li><code>ssl.client.auth=required</code> If set to required"
- + " client authentication is required."
- + " <li><code>ssl.client.auth=requested</code> This means client authentication is optional."
- + " unlike requested , if this option is set client can choose not to provide authentication information about itself"
- + " <li><code>ssl.client.auth=none</code> This means client authentication is not needed.";
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SSL_CLIENT_AUTH_CONFIG = BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG;
+ /**
+ * @deprecated As of 1.0.0. This field will be removed in a future major release.
+ */
+ @Deprecated
+ public static final String SSL_CLIENT_AUTH_DOC = BrokerSecurityConfigs.SSL_CLIENT_AUTH_DOC;
public static void addClientSslSupport(ConfigDef config) {
config.define(SslConfigs.SSL_PROTOCOL_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_PROTOCOL, ConfigDef.Importance.MEDIUM, SslConfigs.SSL_PROTOCOL_DOC)
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
new file mode 100644
index 0000000..18616ec
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java
@@ -0,0 +1,70 @@
+/*
+ * 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.config.internals;
+
+import org.apache.kafka.common.config.SaslConfigs;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Common home for broker-side security configs which need to be accessible from the libraries shared
+ * between the broker and the client.
+ *
+ * Note this is an internal API and subject to change without notice.
+ */
+public class BrokerSecurityConfigs {
+
+ public static final String PRINCIPAL_BUILDER_CLASS_CONFIG = "principal.builder.class";
+ public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG = "sasl.kerberos.principal.to.local.rules";
+ public static final String SSL_CLIENT_AUTH_CONFIG = "ssl.client.auth";
+ public static final String SASL_ENABLED_MECHANISMS_CONFIG = "sasl.enabled.mechanisms";
+
+ public static final String PRINCIPAL_BUILDER_CLASS_DOC = "The fully qualified name of a class that implements the " +
+ "KafkaPrincipalBuilder interface, which is used to build the KafkaPrincipal object used during " +
+ "authorization. This config also supports the deprecated PrincipalBuilder interface which was previously " +
+ "used for client authentication over SSL. If no principal builder is defined, the default behavior depends " +
+ "on the security protocol in use. For SSL authentication, the principal name will be the distinguished " +
+ "name from the client certificate if one is provided; otherwise, if client authentication is not required, " +
+ "the principal name will be ANONYMOUS. For SASL authentication, the principal will be derived using the " +
+ "rules defined by <code>" + SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG + "</code> if GSSAPI is in use, " +
+ "and the SASL authentication ID for other mechanisms. For PLAINTEXT, the principal will be ANONYMOUS.";
+
+ public static final String SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_DOC = "A list of rules for mapping from principal " +
+ "names to short names (typically operating system usernames). The rules are evaluated in order and the " +
+ "first rule that matches a principal name is used to map it to a short name. Any later rules in the list are " +
+ "ignored. By default, principal names of the form {username}/{hostname}@{REALM} are mapped to {username}. " +
+ "For more details on the format please see <a href=\"#security_authz\"> security authorization and acls</a>. " +
+ "Note that this configuration is ignored if an extension of KafkaPrincipalBuilder is provided by the " +
+ "<code>" + PRINCIPAL_BUILDER_CLASS_CONFIG + "</code> configuration.";
+ public static final List<String> DEFAULT_SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES = Collections.singletonList("DEFAULT");
+
+ public static final String SSL_CLIENT_AUTH_DOC = "Configures kafka broker to request client authentication."
+ + " The following settings are common: "
+ + " <ul>"
+ + " <li><code>ssl.client.auth=required</code> If set to required"
+ + " client authentication is required."
+ + " <li><code>ssl.client.auth=requested</code> This means client authentication is optional."
+ + " unlike requested , if this option is set client can choose not to provide authentication information about itself"
+ + " <li><code>ssl.client.auth=none</code> This means client authentication is not needed.";
+
+ public static final String SASL_ENABLED_MECHANISMS_DOC = "The list of SASL mechanisms enabled in the Kafka server. "
+ + "The list may contain any mechanism for which a security provider is available. "
+ + "Only GSSAPI is enabled by default.";
+ public static final List<String> DEFAULT_SASL_ENABLED_MECHANISMS = Collections.singletonList(SaslConfigs.GSSAPI_MECHANISM);
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/Authenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Authenticator.java b/clients/src/main/java/org/apache/kafka/common/network/Authenticator.java
index b242967..1fe3beb 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Authenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Authenticator.java
@@ -16,28 +16,15 @@
*/
package org.apache.kafka.common.network;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+
import java.io.Closeable;
import java.io.IOException;
-import java.util.Map;
-import java.security.Principal;
-
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
-import org.apache.kafka.common.KafkaException;
/**
* Authentication for Channel
*/
public interface Authenticator extends Closeable {
-
- /**
- * Configures Authenticator using the provided parameters.
- *
- * @param transportLayer The transport layer used to read or write tokens
- * @param principalBuilder The builder used to construct `Principal`
- * @param configs Additional configuration parameters as key/value pairs
- */
- void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs);
-
/**
* Implements any authentication mechanism. Use transportLayer to read or write tokens.
* If no further authentication needs to be done returns.
@@ -47,7 +34,7 @@ public interface Authenticator extends Closeable {
/**
* Returns Principal using PrincipalBuilder
*/
- Principal principal() throws KafkaException;
+ KafkaPrincipal principal();
/**
* returns true if authentication is complete otherwise returns false;
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
index 6c8890a..54689f3 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilder.java
@@ -33,7 +33,6 @@ public interface ChannelBuilder extends AutoCloseable {
*/
void configure(Map<String, ?> configs) throws KafkaException;
-
/**
* returns a Channel with TransportLayer and Authenticator configured.
* @param id channel id
@@ -44,10 +43,10 @@ public interface ChannelBuilder extends AutoCloseable {
*/
KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize, MemoryPool memoryPool) throws KafkaException;
-
/**
* Closes ChannelBuilder
*/
+ @Override
void close();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 225f5a8..6dd3ddd 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
@@ -16,13 +16,18 @@
*/
package org.apache.kafka.common.network;
+import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.config.AbstractConfig;
-import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.security.JaasContext;
+import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
import org.apache.kafka.common.security.auth.DefaultPrincipalBuilder;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
import org.apache.kafka.common.security.auth.PrincipalBuilder;
import org.apache.kafka.common.security.authenticator.CredentialCache;
+import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
import org.apache.kafka.common.utils.Utils;
import java.util.Map;
@@ -113,12 +118,13 @@ public class ChannelBuilders {
return channelBuilder;
}
- /**
- * Returns a configured `PrincipalBuilder`.
- */
- static PrincipalBuilder createPrincipalBuilder(Map<String, ?> configs) {
- // this is a server-only config so it will always be null on the client
- Class<?> principalBuilderClass = (Class<?>) configs.get(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG);
+ private static void requireNonNullMode(Mode mode, SecurityProtocol securityProtocol) {
+ if (mode == null)
+ throw new IllegalArgumentException("`mode` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
+ }
+
+ @SuppressWarnings("deprecation")
+ private static PrincipalBuilder createPrincipalBuilder(Class<?> principalBuilderClass, Map<String, ?> configs) {
PrincipalBuilder principalBuilder;
if (principalBuilderClass == null)
principalBuilder = new DefaultPrincipalBuilder();
@@ -128,9 +134,31 @@ public class ChannelBuilders {
return principalBuilder;
}
- private static void requireNonNullMode(Mode mode, SecurityProtocol securityProtocol) {
- if (mode == null)
- throw new IllegalArgumentException("`mode` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
+ @SuppressWarnings("deprecation")
+ public static KafkaPrincipalBuilder createPrincipalBuilder(Map<String, ?> configs,
+ TransportLayer transportLayer,
+ Authenticator authenticator,
+ KerberosShortNamer kerberosShortNamer) {
+ Class<?> principalBuilderClass = (Class<?>) configs.get(BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG);
+ final KafkaPrincipalBuilder builder;
+
+ if (principalBuilderClass == null || principalBuilderClass == DefaultKafkaPrincipalBuilder.class) {
+ builder = new DefaultKafkaPrincipalBuilder(kerberosShortNamer);
+ } else if (KafkaPrincipalBuilder.class.isAssignableFrom(principalBuilderClass)) {
+ builder = (KafkaPrincipalBuilder) Utils.newInstance(principalBuilderClass);
+ } else if (PrincipalBuilder.class.isAssignableFrom(principalBuilderClass)) {
+ PrincipalBuilder oldPrincipalBuilder = createPrincipalBuilder(principalBuilderClass, configs);
+ builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer,
+ oldPrincipalBuilder, kerberosShortNamer);
+ } else {
+ throw new InvalidConfigurationException("Type " + principalBuilderClass.getName() + " is not " +
+ "an instance of " + PrincipalBuilder.class.getName() + " or " + KafkaPrincipalBuilder.class.getName());
+ }
+
+ if (builder instanceof Configurable)
+ ((Configurable) builder).configure(configs);
+
+ return builder;
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/DefaultAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/DefaultAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/network/DefaultAuthenticator.java
deleted file mode 100644
index 77d94a6..0000000
--- a/clients/src/main/java/org/apache/kafka/common/network/DefaultAuthenticator.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.network;
-
-import java.security.Principal;
-import java.io.IOException;
-import java.util.Map;
-
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
-import org.apache.kafka.common.KafkaException;
-
-public class DefaultAuthenticator implements Authenticator {
-
- private TransportLayer transportLayer;
- private PrincipalBuilder principalBuilder;
- private Principal principal;
-
- public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs) {
- this.transportLayer = transportLayer;
- this.principalBuilder = principalBuilder;
- }
-
- /**
- * No-Op for default authenticator
- */
- public void authenticate() throws IOException {}
-
- /**
- * Constructs Principal using configured principalBuilder.
- * @return Principal
- * @throws KafkaException
- */
- public Principal principal() throws KafkaException {
- if (principal == null)
- principal = principalBuilder.buildPrincipal(transportLayer, this);
- return principal;
- }
-
- public void close() throws IOException {}
-
- /**
- * DefaultAuthenticator doesn't implement any additional authentication mechanism.
- * @return true
- */
- public boolean complete() {
- return true;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java b/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
index 1e76c43..c759f00 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
@@ -17,17 +17,15 @@
package org.apache.kafka.common.network;
-import java.io.IOException;
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.utils.Utils;
+import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.channels.SelectionKey;
-
-import java.security.Principal;
-
import java.util.Objects;
-import org.apache.kafka.common.memory.MemoryPool;
-import org.apache.kafka.common.utils.Utils;
public class KafkaChannel {
private final String id;
@@ -66,7 +64,7 @@ public class KafkaChannel {
/**
* Returns the principal returned by `authenticator.principal()`.
*/
- public Principal principal() throws IOException {
+ public KafkaPrincipal principal() {
return authenticator.principal();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
index ad63671..c0d1059 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
@@ -16,45 +16,73 @@
*/
package org.apache.kafka.common.network;
-import java.nio.channels.SelectionKey;
-import java.util.Map;
-
-import org.apache.kafka.common.memory.MemoryPool;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
import org.apache.kafka.common.KafkaException;
-
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
+import org.apache.kafka.common.security.auth.PlaintextAuthenticationContext;
+import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.nio.channels.SelectionKey;
+import java.util.Map;
+
public class PlaintextChannelBuilder implements ChannelBuilder {
private static final Logger log = LoggerFactory.getLogger(PlaintextChannelBuilder.class);
- private PrincipalBuilder principalBuilder;
private Map<String, ?> configs;
public void configure(Map<String, ?> configs) throws KafkaException {
- try {
- this.configs = configs;
- principalBuilder = ChannelBuilders.createPrincipalBuilder(configs);
- } catch (Exception e) {
- throw new KafkaException(e);
- }
+ this.configs = configs;
}
@Override
public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize, MemoryPool memoryPool) throws KafkaException {
try {
PlaintextTransportLayer transportLayer = new PlaintextTransportLayer(key);
- Authenticator authenticator = new DefaultAuthenticator();
- authenticator.configure(transportLayer, this.principalBuilder, this.configs);
- return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE);
+ PlaintextAuthenticator authenticator = new PlaintextAuthenticator(configs, transportLayer);
+ return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize,
+ memoryPool != null ? memoryPool : MemoryPool.NONE);
} catch (Exception e) {
log.warn("Failed to create channel due to ", e);
throw new KafkaException(e);
}
}
- public void close() {
- this.principalBuilder.close();
+ @Override
+ public void close() {}
+
+ private static class PlaintextAuthenticator implements Authenticator {
+ private final PlaintextTransportLayer transportLayer;
+ private final KafkaPrincipalBuilder principalBuilder;
+
+ private PlaintextAuthenticator(Map<String, ?> configs, PlaintextTransportLayer transportLayer) {
+ this.transportLayer = transportLayer;
+ this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, transportLayer, this, null);
+ }
+
+ @Override
+ public void authenticate() throws IOException {}
+
+ @Override
+ public KafkaPrincipal principal() {
+ InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress();
+ return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress));
+ }
+
+ @Override
+ public boolean complete() {
+ return true;
+ }
+
+ @Override
+ public void close() {
+ if (principalBuilder instanceof Closeable)
+ Utils.closeQuietly((Closeable) principalBuilder, "principal builder");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 0f98463..9d7eac0 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
@@ -16,29 +16,30 @@
*/
package org.apache.kafka.common.network;
-import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-import java.util.List;
-import java.util.Map;
-
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.protocol.SecurityProtocol;
import org.apache.kafka.common.security.JaasContext;
-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;
+import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
import org.apache.kafka.common.security.ssl.SslFactory;
import org.apache.kafka.common.utils.Java;
-import org.apache.kafka.common.protocol.SecurityProtocol;
-import org.apache.kafka.common.KafkaException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+import java.util.List;
+import java.util.Map;
+
public class SaslChannelBuilder implements ChannelBuilder {
private static final Logger log = LoggerFactory.getLogger(SaslChannelBuilder.class);
@@ -71,12 +72,13 @@ public class SaslChannelBuilder implements ChannelBuilder {
this.credentialCache = credentialCache;
}
+ @Override
public void configure(Map<String, ?> configs) throws KafkaException {
try {
this.configs = configs;
boolean hasKerberos;
if (mode == Mode.SERVER) {
- List<String> enabledMechanisms = (List<String>) this.configs.get(SaslConfigs.SASL_ENABLED_MECHANISMS);
+ List<String> enabledMechanisms = (List<String>) this.configs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
hasKerberos = enabledMechanisms == null || enabledMechanisms.contains(SaslConfigs.GSSAPI_MECHANISM);
} else {
hasKerberos = clientSaslMechanism.equals(SaslConfigs.GSSAPI_MECHANISM);
@@ -90,7 +92,7 @@ public class SaslChannelBuilder implements ChannelBuilder {
defaultRealm = "";
}
@SuppressWarnings("unchecked")
- List<String> principalToLocalRules = (List<String>) configs.get(SaslConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES);
+ List<String> principalToLocalRules = (List<String>) configs.get(BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG);
if (principalToLocalRules != null)
kerberosShortNamer = KerberosShortNamer.fromUnparsedRules(defaultRealm, principalToLocalRules);
}
@@ -114,14 +116,12 @@ public class SaslChannelBuilder implements ChannelBuilder {
TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel);
Authenticator authenticator;
if (mode == Mode.SERVER)
- authenticator = new SaslServerAuthenticator(id, jaasContext, loginManager.subject(),
- kerberosShortNamer, socketChannel.socket().getLocalAddress(), credentialCache,
- listenerName, securityProtocol);
+ authenticator = new SaslServerAuthenticator(configs, id, jaasContext, loginManager.subject(),
+ kerberosShortNamer, credentialCache, listenerName, securityProtocol, transportLayer);
else
- authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(),
- socketChannel.socket().getInetAddress().getHostName(), clientSaslMechanism, handshakeRequestEnable);
- // Both authenticators don't use `PrincipalBuilder`, so we pass `null` for now. Reconsider if this changes.
- authenticator.configure(transportLayer, null, this.configs);
+ authenticator = new SaslClientAuthenticator(configs, id, loginManager.subject(), loginManager.serviceName(),
+ socketChannel.socket().getInetAddress().getHostName(), clientSaslMechanism,
+ handshakeRequestEnable, transportLayer);
return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE);
} catch (Exception e) {
log.info("Failed to create channel due to ", e);
@@ -129,6 +129,7 @@ public class SaslChannelBuilder implements ChannelBuilder {
}
}
+ @Override
public void close() {
if (loginManager != null) {
loginManager.release();
@@ -136,7 +137,7 @@ public class SaslChannelBuilder implements ChannelBuilder {
}
}
- protected TransportLayer buildTransportLayer(String id, SelectionKey key, SocketChannel socketChannel) throws IOException {
+ private TransportLayer buildTransportLayer(String id, SelectionKey key, SocketChannel socketChannel) throws IOException {
if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
return SslTransportLayer.create(id, key,
sslFactory.createSslEngine(socketChannel.socket().getInetAddress().getHostName(), socketChannel.socket().getPort()));
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
index 01b72fe..b6ef625 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
@@ -16,23 +16,27 @@
*/
package org.apache.kafka.common.network;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.memory.MemoryPool;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
+import org.apache.kafka.common.security.auth.SslAuthenticationContext;
+import org.apache.kafka.common.security.ssl.SslFactory;
+import org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
import java.io.IOException;
+import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Map;
-import org.apache.kafka.common.memory.MemoryPool;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
-import org.apache.kafka.common.security.ssl.SslFactory;
-import org.apache.kafka.common.KafkaException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
public class SslChannelBuilder implements ChannelBuilder {
private static final Logger log = LoggerFactory.getLogger(SslChannelBuilder.class);
private SslFactory sslFactory;
- private PrincipalBuilder principalBuilder;
private Mode mode;
private Map<String, ?> configs;
@@ -45,7 +49,6 @@ public class SslChannelBuilder implements ChannelBuilder {
this.configs = configs;
this.sslFactory = new SslFactory(mode);
this.sslFactory.configure(this.configs);
- this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs);
} catch (Exception e) {
throw new KafkaException(e);
}
@@ -55,23 +58,21 @@ public class SslChannelBuilder implements ChannelBuilder {
public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize, MemoryPool memoryPool) throws KafkaException {
try {
SslTransportLayer transportLayer = buildTransportLayer(sslFactory, id, key, peerHost(key));
- Authenticator authenticator = new DefaultAuthenticator();
- authenticator.configure(transportLayer, this.principalBuilder, this.configs);
- return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE);
+ Authenticator authenticator = new SslAuthenticator(configs, transportLayer);
+ return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize,
+ memoryPool != null ? memoryPool : MemoryPool.NONE);
} catch (Exception e) {
log.info("Failed to create channel due to ", e);
throw new KafkaException(e);
}
}
- public void close() {
- this.principalBuilder.close();
- }
+ @Override
+ public void close() {}
protected SslTransportLayer buildTransportLayer(SslFactory sslFactory, String id, SelectionKey key, String host) throws IOException {
SocketChannel socketChannel = (SocketChannel) key.channel();
- return SslTransportLayer.create(id, key,
- sslFactory.createSslEngine(host, socketChannel.socket().getPort()));
+ return SslTransportLayer.create(id, key, sslFactory.createSslEngine(host, socketChannel.socket().getPort()));
}
/**
@@ -112,4 +113,50 @@ public class SslChannelBuilder implements ChannelBuilder {
SocketChannel socketChannel = (SocketChannel) key.channel();
return new InetSocketAddress(socketChannel.socket().getInetAddress(), 0).getHostString();
}
+
+ /**
+ * Note that client SSL authentication is handled in {@link SslTransportLayer}. This class is only used
+ * to transform the derived principal using a {@link KafkaPrincipalBuilder} configured by the user.
+ */
+ private static class SslAuthenticator implements Authenticator {
+ private final SslTransportLayer transportLayer;
+ private final KafkaPrincipalBuilder principalBuilder;
+
+ private SslAuthenticator(Map<String, ?> configs, SslTransportLayer transportLayer) {
+ this.transportLayer = transportLayer;
+ this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, transportLayer, this, null);
+ }
+ /**
+ * No-Op for plaintext authenticator
+ */
+ @Override
+ public void authenticate() throws IOException {}
+
+ /**
+ * Constructs Principal using configured principalBuilder.
+ * @return the built principal
+ */
+ @Override
+ public KafkaPrincipal principal() {
+ InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress();
+ SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress);
+ return principalBuilder.build(context);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (principalBuilder instanceof Closeable)
+ Utils.closeQuietly((Closeable) principalBuilder, "principal builder");
+ }
+
+ /**
+ * SslAuthenticator doesn't implement any additional authentication mechanism.
+ * @return true
+ */
+ @Override
+ public boolean complete() {
+ return true;
+ }
+
+ }
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/AuthenticationContext.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/AuthenticationContext.java b/clients/src/main/java/org/apache/kafka/common/security/auth/AuthenticationContext.java
new file mode 100644
index 0000000..8c82954
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/AuthenticationContext.java
@@ -0,0 +1,36 @@
+/*
+ * 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.auth;
+
+import java.net.InetAddress;
+
+/**
+ * An object representing contextual information from the authentication session. See
+ * {@link SaslAuthenticationContext} and {@link SslAuthenticationContext}.
+ */
+public interface AuthenticationContext {
+ /**
+ * Underlying security protocol of the authentication session.
+ * @return The name of the security protocol (i.e. PLAINTEXT, SASL_PLAINTEXT, SASL_SSL, SSL)
+ */
+ String securityProtocolName();
+
+ /**
+ * Address of the authenticated client
+ */
+ InetAddress clientAddress();
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java b/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java
index 04e4c90..424cfaa 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java
@@ -23,7 +23,11 @@ import org.apache.kafka.common.network.TransportLayer;
import org.apache.kafka.common.network.Authenticator;
import org.apache.kafka.common.KafkaException;
-/** DefaultPrincipalBuilder which return transportLayer's peer Principal **/
+/**
+ * DefaultPrincipalBuilder which return transportLayer's peer Principal
+ * @deprecated As of Kafka 1.0.0. This will be removed in a future major release.
+ **/
+@Deprecated
public class DefaultPrincipalBuilder implements PrincipalBuilder {
public void configure(Map<String, ?> configs) {}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java
index 4bacdcb..ed3c956 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipal.java
@@ -16,53 +16,69 @@
*/
package org.apache.kafka.common.security.auth;
+import org.apache.kafka.common.utils.SecurityUtils;
+
import java.security.Principal;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Principals in Kafka are defined by a type and a name. The principal type will always be "User"
+ * for the simple authorizer that is enabled by default, but custom authorizers can leverage different
+ * principal types (such as to enable group or role-based ACLs). The {@link KafkaPrincipalBuilder} interface
+ * is used when you need to derive a different principal type from the authentication context, or when
+ * you need to represent relations between different principals. For example, you could extend
+ * {@link KafkaPrincipal} in order to link a user principal to one or more role principals.
+ *
+ * For custom extensions of {@link KafkaPrincipal}, there two key points to keep in mind:
+ *
+ * 1. To be compatible with the ACL APIs provided by Kafka (including the command line tool), each ACL
+ * can only represent a permission granted to a single principal (consisting of a principal type and name).
+ * It is possible to use richer ACL semantics, but you must implement your own mechanisms for adding
+ * and removing ACLs.
+ * 2. In general, {@link KafkaPrincipal} extensions are only useful when the corresponding Authorizer
+ * is also aware of the extension. If you have a {@link KafkaPrincipalBuilder} which derives user groups
+ * from the authentication context (e.g. from an SSL client certificate), then you need a custom
+ * authorizer which is capable of using the additional group information.
+ */
public class KafkaPrincipal implements Principal {
- public static final String SEPARATOR = ":";
public static final String USER_TYPE = "User";
public final static KafkaPrincipal ANONYMOUS = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "ANONYMOUS");
- private String principalType;
- private String name;
+ private final String principalType;
+ private final String name;
public KafkaPrincipal(String principalType, String name) {
- if (principalType == null || name == null) {
- throw new IllegalArgumentException("principalType and name can not be null");
- }
- this.principalType = principalType;
- this.name = name;
+ this.principalType = requireNonNull(principalType, "Principal type cannot be null");
+ this.name = requireNonNull(name, "Principal name cannot be null");
}
+ /**
+ * Parse a {@link KafkaPrincipal} instance from a string. This method cannot be used for {@link KafkaPrincipal}
+ * extensions.
+ *
+ * @param str The input string formatted as "{principalType}:{principalName}"
+ * @return The parsed {@link KafkaPrincipal} instance
+ * @deprecated As of 1.0.0. This method will be removed in a future major release.
+ */
+ @Deprecated
public static KafkaPrincipal fromString(String str) {
- if (str == null || str.isEmpty()) {
- throw new IllegalArgumentException("expected a string in format principalType:principalName but got " + str);
- }
-
- String[] split = str.split(SEPARATOR, 2);
-
- if (split == null || split.length != 2) {
- throw new IllegalArgumentException("expected a string in format principalType:principalName but got " + str);
- }
-
- return new KafkaPrincipal(split[0], split[1]);
+ return SecurityUtils.parseKafkaPrincipal(str);
}
@Override
public String toString() {
- return principalType + SEPARATOR + name;
+ return principalType + ":" + name;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
- if (!(o instanceof KafkaPrincipal)) return false;
+ if (o == null) return false;
+ if (getClass() != o.getClass()) return false;
KafkaPrincipal that = (KafkaPrincipal) o;
-
- if (!principalType.equals(that.principalType)) return false;
- return name.equals(that.name);
-
+ return principalType.equals(that.principalType) && name.equals(that.name);
}
@Override
@@ -80,7 +96,5 @@ public class KafkaPrincipal implements Principal {
public String getPrincipalType() {
return principalType;
}
-}
-
-
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipalBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipalBuilder.java b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipalBuilder.java
new file mode 100644
index 0000000..941d3b1
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/KafkaPrincipalBuilder.java
@@ -0,0 +1,36 @@
+/*
+ * 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.auth;
+
+/**
+ * Pluggable principal builder interface which supports both SSL authentication through
+ * {@link SslAuthenticationContext} and SASL through {@link SaslAuthenticationContext}.
+ *
+ * Note that the {@link org.apache.kafka.common.Configurable} and {@link java.io.Closeable}
+ * interfaces are respected if implemented. Additionally, implementations must provide a
+ * default no-arg constructor.
+ */
+public interface KafkaPrincipalBuilder {
+ /**
+ * Build a kafka principal from the authentication context.
+ * @param context The authentication context (either {@link SslAuthenticationContext} or
+ * {@link SaslAuthenticationContext})
+ * @return The built principal which may provide additional enrichment through a subclass of
+ * {@link KafkaPrincipalBuilder}.
+ */
+ KafkaPrincipal build(AuthenticationContext context);
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/PlaintextAuthenticationContext.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/PlaintextAuthenticationContext.java b/clients/src/main/java/org/apache/kafka/common/security/auth/PlaintextAuthenticationContext.java
new file mode 100644
index 0000000..96b8376
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/PlaintextAuthenticationContext.java
@@ -0,0 +1,40 @@
+/*
+ * 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.auth;
+
+import org.apache.kafka.common.protocol.SecurityProtocol;
+
+import java.net.InetAddress;
+
+public class PlaintextAuthenticationContext implements AuthenticationContext {
+ private final InetAddress clientAddress;
+
+ public PlaintextAuthenticationContext(InetAddress clientAddress) {
+ this.clientAddress = clientAddress;
+ }
+
+ @Override
+ public String securityProtocolName() {
+ return SecurityProtocol.PLAINTEXT.name;
+ }
+
+ @Override
+ public InetAddress clientAddress() {
+ return clientAddress;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java b/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
index fc2a930..d58170a 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
@@ -25,10 +25,13 @@ import org.apache.kafka.common.Configurable;
import java.util.Map;
import java.security.Principal;
-/*
+/**
* PrincipalBuilder for Authenticator
+ * @deprecated As of Kafka 1.0.0, use {@link KafkaPrincipalBuilder} instead. This will be removed in
+ * a future major release.
*/
@InterfaceStability.Unstable
+@Deprecated
public interface PrincipalBuilder extends Configurable {
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/SaslAuthenticationContext.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/SaslAuthenticationContext.java b/clients/src/main/java/org/apache/kafka/common/security/auth/SaslAuthenticationContext.java
new file mode 100644
index 0000000..f98164b
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/SaslAuthenticationContext.java
@@ -0,0 +1,48 @@
+/*
+ * 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.auth;
+
+import org.apache.kafka.common.protocol.SecurityProtocol;
+
+import javax.security.sasl.SaslServer;
+import java.net.InetAddress;
+
+public class SaslAuthenticationContext implements AuthenticationContext {
+ private final SaslServer server;
+ private final SecurityProtocol securityProtocol;
+ private final InetAddress clientAddress;
+
+ public SaslAuthenticationContext(SaslServer server, SecurityProtocol securityProtocol, InetAddress clientAddress) {
+ this.server = server;
+ this.securityProtocol = securityProtocol;
+ this.clientAddress = clientAddress;
+ }
+
+ public SaslServer server() {
+ return server;
+ }
+
+ @Override
+ public String securityProtocolName() {
+ return securityProtocol.name;
+ }
+
+ @Override
+ public InetAddress clientAddress() {
+ return clientAddress;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/auth/SslAuthenticationContext.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/SslAuthenticationContext.java b/clients/src/main/java/org/apache/kafka/common/security/auth/SslAuthenticationContext.java
new file mode 100644
index 0000000..325c282
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/SslAuthenticationContext.java
@@ -0,0 +1,46 @@
+/*
+ * 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.auth;
+
+import org.apache.kafka.common.protocol.SecurityProtocol;
+
+import javax.net.ssl.SSLSession;
+import java.net.InetAddress;
+
+public class SslAuthenticationContext implements AuthenticationContext {
+ private final SSLSession session;
+ private final InetAddress clientAddress;
+
+ public SslAuthenticationContext(SSLSession session, InetAddress clientAddress) {
+ this.session = session;
+ this.clientAddress = clientAddress;
+ }
+
+ public SSLSession session() {
+ return session;
+ }
+
+ @Override
+ public String securityProtocolName() {
+ return SecurityProtocol.SSL.name;
+ }
+
+ @Override
+ public InetAddress clientAddress() {
+ return clientAddress;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/authenticator/DefaultKafkaPrincipalBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/DefaultKafkaPrincipalBuilder.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/DefaultKafkaPrincipalBuilder.java
new file mode 100644
index 0000000..1d0295a
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/DefaultKafkaPrincipalBuilder.java
@@ -0,0 +1,149 @@
+/*
+ * 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 org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.network.Authenticator;
+import org.apache.kafka.common.network.TransportLayer;
+import org.apache.kafka.common.security.auth.AuthenticationContext;
+import org.apache.kafka.common.security.auth.KafkaPrincipal;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
+import org.apache.kafka.common.security.auth.PlaintextAuthenticationContext;
+import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.apache.kafka.common.security.auth.SaslAuthenticationContext;
+import org.apache.kafka.common.security.auth.SslAuthenticationContext;
+import org.apache.kafka.common.security.kerberos.KerberosName;
+import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
+
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSession;
+import javax.security.sasl.SaslServer;
+import java.io.Closeable;
+import java.io.IOException;
+import java.security.Principal;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Default implementation of {@link KafkaPrincipalBuilder} which provides basic support for
+ * SSL authentication and SASL authentication. In the latter case, when GSSAPI is used, this
+ * class applies {@link org.apache.kafka.common.security.kerberos.KerberosShortNamer} to transform
+ * the name.
+ *
+ * NOTE: This is an internal class and can change without notice. Unlike normal implementations
+ * of {@link KafkaPrincipalBuilder}, there is no default no-arg constructor since this class
+ * must adapt implementations of the older {@link PrincipalBuilder} interface.
+ */
+public class DefaultKafkaPrincipalBuilder implements KafkaPrincipalBuilder, Closeable {
+ @SuppressWarnings("deprecation")
+ private final PrincipalBuilder oldPrincipalBuilder;
+ private final Authenticator authenticator;
+ private final TransportLayer transportLayer;
+ private final KerberosShortNamer kerberosShortNamer;
+
+ /**
+ * Construct a new instance which wraps an instance of the older {@link PrincipalBuilder}.
+ *
+ * @param authenticator The authenticator in use
+ * @param transportLayer The underlying transport layer
+ * @param oldPrincipalBuilder Instance of {@link PrincipalBuilder}
+ * @param kerberosShortNamer Kerberos name rewrite rules or null if none have been configured
+ */
+ @SuppressWarnings("deprecation")
+ public static DefaultKafkaPrincipalBuilder fromOldPrincipalBuilder(Authenticator authenticator,
+ TransportLayer transportLayer,
+ PrincipalBuilder oldPrincipalBuilder,
+ KerberosShortNamer kerberosShortNamer) {
+ return new DefaultKafkaPrincipalBuilder(
+ requireNonNull(authenticator),
+ requireNonNull(transportLayer),
+ requireNonNull(oldPrincipalBuilder),
+ kerberosShortNamer);
+ }
+
+ @SuppressWarnings("deprecation")
+ private DefaultKafkaPrincipalBuilder(Authenticator authenticator,
+ TransportLayer transportLayer,
+ PrincipalBuilder oldPrincipalBuilder,
+ KerberosShortNamer kerberosShortNamer) {
+ this.authenticator = authenticator;
+ this.transportLayer = transportLayer;
+ this.oldPrincipalBuilder = oldPrincipalBuilder;
+ this.kerberosShortNamer = kerberosShortNamer;
+ }
+
+ /**
+ * Construct a new instance.
+ *
+ * @param kerberosShortNamer Kerberos name rewrite rules or null if none have been configured
+ */
+ public DefaultKafkaPrincipalBuilder(KerberosShortNamer kerberosShortNamer) {
+ this(null, null, null, kerberosShortNamer);
+ }
+
+ @Override
+ public KafkaPrincipal build(AuthenticationContext context) {
+ if (context instanceof PlaintextAuthenticationContext) {
+ if (oldPrincipalBuilder != null)
+ return convertToKafkaPrincipal(oldPrincipalBuilder.buildPrincipal(transportLayer, authenticator));
+
+ return KafkaPrincipal.ANONYMOUS;
+ } else if (context instanceof SslAuthenticationContext) {
+ SSLSession sslSession = ((SslAuthenticationContext) context).session();
+
+ if (oldPrincipalBuilder != null)
+ return convertToKafkaPrincipal(oldPrincipalBuilder.buildPrincipal(transportLayer, authenticator));
+
+ try {
+ return convertToKafkaPrincipal(sslSession.getPeerPrincipal());
+ } catch (SSLPeerUnverifiedException se) {
+ return KafkaPrincipal.ANONYMOUS;
+ }
+ } else if (context instanceof SaslAuthenticationContext) {
+ SaslServer saslServer = ((SaslAuthenticationContext) context).server();
+ if (SaslConfigs.GSSAPI_MECHANISM.equals(saslServer.getMechanismName()))
+ return applyKerberosShortNamer(saslServer.getAuthorizationID());
+ else
+ return new KafkaPrincipal(KafkaPrincipal.USER_TYPE, saslServer.getAuthorizationID());
+ } else {
+ throw new IllegalArgumentException("Unhandled authentication context type: " + context.getClass().getName());
+ }
+ }
+
+ private KafkaPrincipal applyKerberosShortNamer(String authorizationId) {
+ KerberosName kerberosName = KerberosName.parse(authorizationId);
+ try {
+ String shortName = kerberosShortNamer.shortName(kerberosName);
+ return new KafkaPrincipal(KafkaPrincipal.USER_TYPE, shortName);
+ } catch (IOException e) {
+ throw new KafkaException("Failed to set name for '" + kerberosName +
+ "' based on Kerberos authentication rules.", e);
+ }
+ }
+
+ private KafkaPrincipal convertToKafkaPrincipal(Principal principal) {
+ return new KafkaPrincipal(KafkaPrincipal.USER_TYPE, principal.getName());
+ }
+
+ @Override
+ public void close() {
+ if (oldPrincipalBuilder != null)
+ oldPrincipalBuilder.close();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
index 6bf9b2a..7b68abe 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
@@ -38,7 +38,6 @@ import org.apache.kafka.common.requests.SaslHandshakeRequest;
import org.apache.kafka.common.requests.SaslHandshakeResponse;
import org.apache.kafka.common.security.auth.AuthCallbackHandler;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,14 +69,11 @@ public class SaslClientAuthenticator implements Authenticator {
private final String host;
private final String node;
private final String mechanism;
- private final boolean handshakeRequestEnable;
-
- // assigned in `configure`
- private SaslClient saslClient;
- private Map<String, ?> configs;
- private String clientPrincipalName;
- private AuthCallbackHandler callbackHandler;
- private TransportLayer transportLayer;
+ private final TransportLayer transportLayer;
+ private final SaslClient saslClient;
+ private final Map<String, ?> configs;
+ private final String clientPrincipalName;
+ private final AuthCallbackHandler callbackHandler;
// buffers used in `authenticate`
private NetworkReceive netInBuffer;
@@ -92,21 +88,24 @@ public class SaslClientAuthenticator implements Authenticator {
// Request header for which response from the server is pending
private RequestHeader currentRequestHeader;
- public SaslClientAuthenticator(String node, Subject subject, String servicePrincipal, String host, String mechanism, boolean handshakeRequestEnable) throws IOException {
+ public SaslClientAuthenticator(Map<String, ?> configs,
+ String node,
+ Subject subject,
+ String servicePrincipal,
+ String host,
+ String mechanism,
+ boolean handshakeRequestEnable,
+ TransportLayer transportLayer) throws IOException {
this.node = node;
this.subject = subject;
this.host = host;
this.servicePrincipal = servicePrincipal;
this.mechanism = mechanism;
- this.handshakeRequestEnable = handshakeRequestEnable;
this.correlationId = -1;
- }
+ this.transportLayer = transportLayer;
+ this.configs = configs;
- public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs) throws KafkaException {
try {
- this.transportLayer = transportLayer;
- this.configs = configs;
-
setSaslState(handshakeRequestEnable ? SaslState.SEND_HANDSHAKE_REQUEST : SaslState.INITIAL);
// determine client principal from subject for Kerberos to use as authorization id for the SaslClient.
@@ -252,7 +251,7 @@ public class SaslClientAuthenticator implements Authenticator {
return serverPacket;
}
- public Principal principal() {
+ public KafkaPrincipal principal() {
return new KafkaPrincipal(KafkaPrincipal.USER_TYPE, clientPrincipalName);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3b5d88fe/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 276d067..28aa995 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
@@ -18,11 +18,13 @@ package org.apache.kafka.common.security.authenticator;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.IllegalSaslStateException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.UnsupportedSaslMechanismException;
import org.apache.kafka.common.network.Authenticator;
+import org.apache.kafka.common.network.ChannelBuilders;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.network.NetworkReceive;
@@ -43,12 +45,14 @@ import org.apache.kafka.common.requests.SaslHandshakeResponse;
import org.apache.kafka.common.security.JaasContext;
import org.apache.kafka.common.security.auth.AuthCallbackHandler;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder;
+import org.apache.kafka.common.security.auth.SaslAuthenticationContext;
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.apache.kafka.common.utils.Utils;
import org.ietf.jgss.GSSContext;
import org.ietf.jgss.GSSCredential;
import org.ietf.jgss.GSSException;
@@ -62,11 +66,11 @@ import javax.security.auth.Subject;
import javax.security.sasl.Sasl;
import javax.security.sasl.SaslException;
import javax.security.sasl.SaslServer;
+import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
-import java.security.Principal;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
import java.util.HashSet;
@@ -80,7 +84,7 @@ public class SaslServerAuthenticator implements Authenticator {
static final int MAX_RECEIVE_SIZE = 524288;
private static final Logger LOG = LoggerFactory.getLogger(SaslServerAuthenticator.class);
- public enum SaslState {
+ private enum SaslState {
GSSAPI_OR_HANDSHAKE_REQUEST, HANDSHAKE_REQUEST, AUTHENTICATE, COMPLETE, FAILED
}
@@ -89,9 +93,11 @@ public class SaslServerAuthenticator implements Authenticator {
private final String connectionId;
private final JaasContext jaasContext;
private final Subject subject;
- private final KerberosShortNamer kerberosNamer;
- private final InetAddress clientAddress;
private final CredentialCache credentialCache;
+ private final TransportLayer transportLayer;
+ private final Set<String> enabledMechanisms;
+ private final Map<String, ?> configs;
+ private final KafkaPrincipalBuilder principalBuilder;
// Current SASL state
private SaslState saslState = SaslState.GSSAPI_OR_HANDSHAKE_REQUEST;
@@ -101,48 +107,44 @@ public class SaslServerAuthenticator implements Authenticator {
private String saslMechanism;
private AuthCallbackHandler callbackHandler;
- // assigned in `configure`
- private TransportLayer transportLayer;
- private Set<String> enabledMechanisms;
- private Map<String, ?> configs;
-
// buffers used in `authenticate`
private NetworkReceive netInBuffer;
private Send netOutBuffer;
- public SaslServerAuthenticator(String connectionId,
+ public SaslServerAuthenticator(Map<String, ?> configs,
+ String connectionId,
JaasContext jaasContext,
Subject subject,
KerberosShortNamer kerberosNameParser,
- InetAddress clientAddress,
CredentialCache credentialCache,
ListenerName listenerName,
- SecurityProtocol securityProtocol) throws IOException {
+ SecurityProtocol securityProtocol,
+ TransportLayer transportLayer) throws IOException {
if (subject == null)
throw new IllegalArgumentException("subject cannot be null");
this.connectionId = connectionId;
this.jaasContext = jaasContext;
this.subject = subject;
- this.kerberosNamer = kerberosNameParser;
- this.clientAddress = clientAddress;
this.credentialCache = credentialCache;
this.listenerName = listenerName;
this.securityProtocol = securityProtocol;
- }
-
- public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map<String, ?> configs) {
this.transportLayer = transportLayer;
+
this.configs = configs;
- List<String> enabledMechanisms = (List<String>) this.configs.get(SaslConfigs.SASL_ENABLED_MECHANISMS);
+ List<String> enabledMechanisms = (List<String>) this.configs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
if (enabledMechanisms == null || enabledMechanisms.isEmpty())
throw new IllegalArgumentException("No SASL mechanisms are enabled");
this.enabledMechanisms = new HashSet<>(enabledMechanisms);
+
+ // Note that the old principal builder does not support SASL, so we do not need to pass the
+ // authenticator or the transport layer
+ this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, null, null, kerberosNameParser);
}
private void createSaslServer(String mechanism) throws IOException {
this.saslMechanism = mechanism;
if (!ScramMechanism.isScram(mechanism))
- callbackHandler = new SaslServerCallbackHandler(jaasContext, kerberosNamer);
+ callbackHandler = new SaslServerCallbackHandler(jaasContext);
else
callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism, ScramCredential.class));
callbackHandler.configure(configs, Mode.SERVER, subject, saslMechanism);
@@ -152,7 +154,8 @@ public class SaslServerAuthenticator implements Authenticator {
try {
saslServer = Subject.doAs(subject, new PrivilegedExceptionAction<SaslServer>() {
public SaslServer run() throws SaslException {
- return Sasl.createSaslServer(saslMechanism, "kafka", clientAddress.getHostName(), configs, callbackHandler);
+ return Sasl.createSaslServer(saslMechanism, "kafka", serverAddress().getHostName(),
+ configs, callbackHandler);
}
});
} catch (PrivilegedActionException e) {
@@ -214,6 +217,7 @@ public class SaslServerAuthenticator implements Authenticator {
* The messages are sent and received as size delimited bytes that consists of a 4 byte network-ordered size N
* followed by N bytes representing the opaque payload.
*/
+ @Override
public void authenticate() throws IOException {
if (netOutBuffer != null && !flushNetOutBufferAndUpdateInterestOps())
return;
@@ -263,15 +267,21 @@ public class SaslServerAuthenticator implements Authenticator {
}
}
- public Principal principal() {
- return new KafkaPrincipal(KafkaPrincipal.USER_TYPE, saslServer.getAuthorizationID());
+ @Override
+ public KafkaPrincipal principal() {
+ SaslAuthenticationContext context = new SaslAuthenticationContext(saslServer, securityProtocol, clientAddress());
+ return principalBuilder.build(context);
}
+ @Override
public boolean complete() {
return saslState == SaslState.COMPLETE;
}
+ @Override
public void close() throws IOException {
+ if (principalBuilder instanceof Closeable)
+ Utils.closeQuietly((Closeable) principalBuilder, "principal builder");
if (saslServer != null)
saslServer.dispose();
if (callbackHandler != null)
@@ -305,6 +315,14 @@ public class SaslServerAuthenticator implements Authenticator {
return netOutBuffer.completed();
}
+ private InetAddress serverAddress() {
+ return transportLayer.socketChannel().socket().getLocalAddress();
+ }
+
+ private InetAddress clientAddress() {
+ return transportLayer.socketChannel().socket().getInetAddress();
+ }
+
private boolean handleKafkaRequest(byte[] requestBytes) throws IOException, AuthenticationException {
boolean isKafkaRequest = false;
String clientMechanism = null;
@@ -325,7 +343,8 @@ public class SaslServerAuthenticator implements Authenticator {
LOG.debug("Handling Kafka request {}", apiKey);
- RequestContext requestContext = new RequestContext(header, connectionId, clientAddress,
+
+ RequestContext requestContext = new RequestContext(header, connectionId, clientAddress(),
KafkaPrincipal.ANONYMOUS, listenerName, securityProtocol);
RequestAndSize requestAndSize = requestContext.parseRequest(requestBuffer);
if (apiKey == ApiKeys.API_VERSIONS)