From 9c519d741a84e5783facd796fafa6ce99a660233 Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Fri, 30 Mar 2018 17:38:17 +0100 Subject: [PATCH 1/6] KAFKA-6750 - KIP-282: Add the listener name to the authentication context Co-authored-by: Edoardo Comar Co-authored-by: Mickael Maison --- .../kafka/common/network/ChannelBuilders.java | 2 +- .../network/PlaintextChannelBuilder.java | 17 +++- .../common/network/SslChannelBuilder.java | 8 +- .../security/auth/AuthenticationContext.java | 10 ++- .../auth/PlaintextAuthenticationContext.java | 11 ++- .../auth/SaslAuthenticationContext.java | 12 ++- .../auth/SslAuthenticationContext.java | 12 ++- .../SaslServerAuthenticator.java | 2 +- .../common/network/ChannelBuildersTest.java | 4 +- .../kafka/common/network/SelectorTest.java | 4 +- .../common/network/SslTransportLayerTest.java | 8 +- .../DefaultKafkaPrincipalBuilderTest.java | 19 +++-- .../ListenerAndPrincipalBuilderTest.scala | 81 +++++++++++++++++++ 13 files changed, 165 insertions(+), 25 deletions(-) create mode 100644 core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala 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 078d844e85fce..1a643963fe484 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 @@ -131,7 +131,7 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol, tokenCache); break; case PLAINTEXT: - channelBuilder = new PlaintextChannelBuilder(); + channelBuilder = new PlaintextChannelBuilder(listenerName); break; default: throw new IllegalArgumentException("Unexpected securityProtocol " + securityProtocol); 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 c0d1059aea95a..9e963eae02631 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 @@ -34,6 +34,15 @@ public class PlaintextChannelBuilder implements ChannelBuilder { private static final Logger log = LoggerFactory.getLogger(PlaintextChannelBuilder.class); private Map configs; + private ListenerName listenerName; + + /** + * Constructs a plaintext channel builder. ListenerName is provided only + * for server channel builder and will be null for client channel builder. + */ + public PlaintextChannelBuilder(ListenerName listenerName) { + this.listenerName = listenerName; + } public void configure(Map configs) throws KafkaException { this.configs = configs; @@ -43,7 +52,7 @@ public void configure(Map configs) throws KafkaException { public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize, MemoryPool memoryPool) throws KafkaException { try { PlaintextTransportLayer transportLayer = new PlaintextTransportLayer(key); - PlaintextAuthenticator authenticator = new PlaintextAuthenticator(configs, transportLayer); + PlaintextAuthenticator authenticator = new PlaintextAuthenticator(configs, transportLayer, listenerName); return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE); } catch (Exception e) { @@ -58,10 +67,12 @@ public void close() {} private static class PlaintextAuthenticator implements Authenticator { private final PlaintextTransportLayer transportLayer; private final KafkaPrincipalBuilder principalBuilder; + private final ListenerName listenerName; - private PlaintextAuthenticator(Map configs, PlaintextTransportLayer transportLayer) { + private PlaintextAuthenticator(Map configs, PlaintextTransportLayer transportLayer, ListenerName listenerName) { this.transportLayer = transportLayer; this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, transportLayer, this, null); + this.listenerName = listenerName; } @Override @@ -70,7 +81,7 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress)); + return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerName)); } @Override 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 941c455365d56..b8007a76c4a45 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 @@ -89,7 +89,7 @@ public ListenerName listenerName() { 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 SslAuthenticator(configs, transportLayer); + Authenticator authenticator = new SslAuthenticator(configs, transportLayer, listenerName); return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE); } catch (Exception e) { @@ -152,10 +152,12 @@ private String peerHost(SelectionKey key) { private static class SslAuthenticator implements Authenticator { private final SslTransportLayer transportLayer; private final KafkaPrincipalBuilder principalBuilder; + private final ListenerName listenerName; - private SslAuthenticator(Map configs, SslTransportLayer transportLayer) { + private SslAuthenticator(Map configs, SslTransportLayer transportLayer, ListenerName listenerName) { this.transportLayer = transportLayer; this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, transportLayer, this, null); + this.listenerName = listenerName; } /** * No-Op for plaintext authenticator @@ -170,7 +172,7 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress); + SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress, listenerName); return principalBuilder.build(context); } 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 index b8c08477027cf..b5666a1b0c330 100644 --- 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 @@ -18,9 +18,12 @@ import java.net.InetAddress; +import org.apache.kafka.common.network.ListenerName; + /** * An object representing contextual information from the authentication session. See - * {@link SaslAuthenticationContext} and {@link SslAuthenticationContext}. + * {@link PlaintextAuthenticationContext}, {@link SaslAuthenticationContext} + * and {@link SslAuthenticationContext}. */ public interface AuthenticationContext { /** @@ -32,4 +35,9 @@ public interface AuthenticationContext { * Address of the authenticated client */ InetAddress clientAddress(); + + /** + * Name of the listener used for the connection + */ + ListenerName listenerName(); } 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 index bc14d36aaf540..89dbb1c056d64 100644 --- 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 @@ -18,11 +18,15 @@ import java.net.InetAddress; +import org.apache.kafka.common.network.ListenerName; + public class PlaintextAuthenticationContext implements AuthenticationContext { private final InetAddress clientAddress; + private final ListenerName listenerName; - public PlaintextAuthenticationContext(InetAddress clientAddress) { + public PlaintextAuthenticationContext(InetAddress clientAddress, ListenerName listenerName) { this.clientAddress = clientAddress; + this.listenerName = listenerName; } @Override @@ -35,4 +39,9 @@ public InetAddress clientAddress() { return clientAddress; } + @Override + public ListenerName listenerName() { + return listenerName; + } + } 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 index 89e606377d9eb..d87b3556b2dae 100644 --- 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 @@ -17,17 +17,22 @@ package org.apache.kafka.common.security.auth; import javax.security.sasl.SaslServer; + +import org.apache.kafka.common.network.ListenerName; + import java.net.InetAddress; public class SaslAuthenticationContext implements AuthenticationContext { private final SaslServer server; private final SecurityProtocol securityProtocol; private final InetAddress clientAddress; + private final ListenerName listenerName; - public SaslAuthenticationContext(SaslServer server, SecurityProtocol securityProtocol, InetAddress clientAddress) { + public SaslAuthenticationContext(SaslServer server, SecurityProtocol securityProtocol, InetAddress clientAddress, ListenerName listenerName) { this.server = server; this.securityProtocol = securityProtocol; this.clientAddress = clientAddress; + this.listenerName = listenerName; } public SaslServer server() { @@ -43,4 +48,9 @@ public SecurityProtocol securityProtocol() { public InetAddress clientAddress() { return clientAddress; } + + @Override + public ListenerName listenerName() { + return listenerName; + } } 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 index d87a8920b479e..a370c84551eed 100644 --- 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 @@ -17,15 +17,20 @@ package org.apache.kafka.common.security.auth; import javax.net.ssl.SSLSession; + +import org.apache.kafka.common.network.ListenerName; + import java.net.InetAddress; public class SslAuthenticationContext implements AuthenticationContext { private final SSLSession session; private final InetAddress clientAddress; + private final ListenerName listenerName; - public SslAuthenticationContext(SSLSession session, InetAddress clientAddress) { + public SslAuthenticationContext(SSLSession session, InetAddress clientAddress, ListenerName listenerName) { this.session = session; this.clientAddress = clientAddress; + this.listenerName = listenerName; } public SSLSession session() { @@ -41,4 +46,9 @@ public SecurityProtocol securityProtocol() { public InetAddress clientAddress() { return clientAddress; } + + @Override + public ListenerName listenerName() { + return listenerName; + } } 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 5140afb196d2c..0d8e47bbcfba6 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 @@ -282,7 +282,7 @@ public void authenticate() throws IOException { @Override public KafkaPrincipal principal() { - SaslAuthenticationContext context = new SaslAuthenticationContext(saslServer, securityProtocol, clientAddress()); + SaslAuthenticationContext context = new SaslAuthenticationContext(saslServer, securityProtocol, clientAddress(), listenerName); KafkaPrincipal principal = principalBuilder.build(context); if (ScramMechanism.isScram(saslMechanism) && Boolean.parseBoolean((String) saslServer.getNegotiatedProperty(ScramLoginModule.TOKEN_AUTH_CONFIG))) { principal.tokenAuthenticated(true); 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 index de210e7496cb1..346c7a8b8704a 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java @@ -24,6 +24,7 @@ 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.SecurityProtocol; import org.easymock.EasyMock; import org.junit.Test; @@ -38,7 +39,6 @@ public class ChannelBuildersTest { @Test - @SuppressWarnings("deprecation") public void testCreateOldPrincipalBuilder() throws Exception { TransportLayer transportLayer = EasyMock.mock(TransportLayer.class); Authenticator authenticator = EasyMock.mock(Authenticator.class); @@ -51,7 +51,7 @@ public void testCreateOldPrincipalBuilder() throws Exception { assertTrue(OldPrincipalBuilder.configured); // test delegation - KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost())); + KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); assertEquals(OldPrincipalBuilder.PRINCIPAL_NAME, principal.getName()); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); } diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java index f1c6a5af1420b..8ce8d5043bee7 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java @@ -80,7 +80,7 @@ public void setUp() throws Exception { this.server = new EchoServer(SecurityProtocol.PLAINTEXT, configs); this.server.start(); this.time = new MockTime(); - this.channelBuilder = new PlaintextChannelBuilder(); + this.channelBuilder = new PlaintextChannelBuilder(ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)); this.channelBuilder.configure(configs); this.metrics = new Metrics(); this.selector = new Selector(5000, this.metrics, time, "MetricGroup", channelBuilder, new LogContext()); @@ -305,7 +305,7 @@ public void testMute() throws Exception { @Test public void registerFailure() throws Exception { - ChannelBuilder channelBuilder = new PlaintextChannelBuilder() { + ChannelBuilder channelBuilder = new PlaintextChannelBuilder(null) { @Override public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize, MemoryPool memoryPool) throws KafkaException { 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 d6b4bac48d8e3..64c31575b2fe2 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 @@ -436,6 +436,8 @@ public void testInvalidSecureRandomImplementation() throws Exception { fail("SSL channel configured with invalid SecureRandom implementation"); } catch (KafkaException e) { // Expected exception + } finally { + channelBuilder.close(); } } @@ -451,6 +453,8 @@ public void testInvalidTruststorePassword() throws Exception { fail("SSL channel configured with invalid truststore password"); } catch (KafkaException e) { // Expected exception + } finally { + channelBuilder.close(); } } @@ -466,6 +470,8 @@ public void testInvalidKeystorePassword() throws Exception { fail("SSL channel configured with invalid keystore password"); } catch (KafkaException e) { // Expected exception + } finally { + channelBuilder.close(); } } @@ -767,7 +773,7 @@ public void testCloseSsl() throws Exception { @Test public void testClosePlaintext() throws Exception { - testClose(SecurityProtocol.PLAINTEXT, new PlaintextChannelBuilder()); + testClose(SecurityProtocol.PLAINTEXT, new PlaintextChannelBuilder(null)); } private void testClose(SecurityProtocol securityProtocol, ChannelBuilder clientChannelBuilder) throws Exception { 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 index fdf368788c02e..9baee8be6c6b2 100644 --- 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 @@ -18,6 +18,7 @@ import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.network.Authenticator; +import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.network.TransportLayer; import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder; import org.apache.kafka.common.security.kerberos.KerberosName; @@ -53,19 +54,19 @@ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost())); + KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); 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()))); + assertEquals(KafkaPrincipal.ANONYMOUS, builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)))); + builder.close(); } @Test @@ -86,12 +87,11 @@ public void testUseOldPrincipalBuilderForSslIfProvided() throws Exception { DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost())); + KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); builder.close(); - verifyAll(); } @@ -105,10 +105,11 @@ public void testUseSessionPeerPrincipalForSsl() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost())); + KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); + builder.close(); verifyAll(); } @@ -124,10 +125,11 @@ public void testPrincipalBuilderScram() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server, - SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost())); + SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT))); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); + builder.close(); verifyAll(); } @@ -146,10 +148,11 @@ public void testPrincipalBuilderGssapi() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(kerberosShortNamer); KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server, - SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost())); + SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT))); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); + builder.close(); verifyAll(); } diff --git a/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala b/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala new file mode 100644 index 0000000000000..14861c07b9567 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala @@ -0,0 +1,81 @@ +/** + * Licensed 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.server + +import java.net.Socket +import java.util.Properties + +import kafka.utils.TestUtils +import org.apache.kafka.common.network.ListenerName +import org.apache.kafka.common.protocol.types.Struct +import org.apache.kafka.common.requests._ +import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder} +import org.junit.Assert._ +import org.junit.{Before, Test} + +class ListenerAndPrincipalBuilderTest extends BaseRequestTest { + + override def numBrokers: Int = 1 + + override def propertyOverrides(properties: Properties): Unit = { + properties.put(KafkaConfig.PrincipalBuilderClassProp, classOf[ListenerAndPrincipalBuilderTest.TestPrincipalBuilder].getName) + properties.put(KafkaConfig.ListenersProp, "PLAINTEXT://localhost:0,CUSTOM://localhost:0") + properties.put(KafkaConfig.ListenerSecurityProtocolMapProp, "PLAINTEXT:PLAINTEXT,CUSTOM:PLAINTEXT") + } + + @Before + override def setUp() { + super.setUp() + + TestUtils.retry(10000) { + // controller to node connection using the default security.inter.broker.protocol + assertEquals("PLAINTEXT", ListenerAndPrincipalBuilderTest.lastListenerName) + } + } + + @Test + def testListenerNameFromAuthenticationContext() { + val port = anySocketServer.boundPort(new ListenerName("CUSTOM")) + val socket = new Socket("localhost", port) + try { + // using the simplest possible builder + requestResponse(socket, "clientId", 0, new ListGroupsRequest.Builder()) + } finally { + socket.close() + } + assertEquals("CUSTOM", ListenerAndPrincipalBuilderTest.lastListenerName) + } + + private def requestResponse(socket: Socket, clientId: String, correlationId: Int, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Struct = { + val apiKey = requestBuilder.apiKey + val request = requestBuilder.build() + val header = new RequestHeader(apiKey, request.version, clientId, correlationId) + val response = requestAndReceive(socket, request.serialize(header).array) + val responseBuffer = skipResponseHeader(response) + apiKey.parseResponse(request.version, responseBuffer) + } + +} + +object ListenerAndPrincipalBuilderTest { + var lastListenerName = "" + + class TestPrincipalBuilder extends KafkaPrincipalBuilder { + override def build(context: AuthenticationContext): KafkaPrincipal = { + lastListenerName = context.listenerName.value + KafkaPrincipal.ANONYMOUS + } + } +} From 70bd84e87263f47f33113d6aa76a5997c887bdef Mon Sep 17 00:00:00 2001 From: Edoardo Comar Date: Mon, 30 Apr 2018 14:59:52 +0100 Subject: [PATCH 2/6] Changed listenerName type to be a String on AuthenticationContext public interface --- .../common/network/PlaintextChannelBuilder.java | 4 +++- .../kafka/common/network/SslChannelBuilder.java | 4 +++- .../common/security/auth/AuthenticationContext.java | 3 +-- .../auth/PlaintextAuthenticationContext.java | 8 +++----- .../security/auth/SaslAuthenticationContext.java | 8 +++----- .../security/auth/SslAuthenticationContext.java | 8 +++----- .../authenticator/SaslServerAuthenticator.java | 2 +- .../kafka/common/network/ChannelBuildersTest.java | 2 +- .../auth/DefaultKafkaPrincipalBuilderTest.java | 13 ++++++------- .../server/ListenerAndPrincipalBuilderTest.scala | 12 +----------- 10 files changed, 25 insertions(+), 39 deletions(-) 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 9e963eae02631..ab65f5382c3f1 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 @@ -81,7 +81,9 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerName)); + // listenerName should only be null in Client mode + String listenerNameStr = listenerName != null ? listenerName.value() : null; + return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerNameStr)); } @Override 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 b8007a76c4a45..ae167f57193aa 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 @@ -172,7 +172,9 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress, listenerName); + // listenerName should only be null in Client mode + String listenerNameStr = listenerName != null ? listenerName.value() : null; + SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress, listenerNameStr); return principalBuilder.build(context); } 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 index b5666a1b0c330..3e644b202e06d 100644 --- 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 @@ -18,7 +18,6 @@ import java.net.InetAddress; -import org.apache.kafka.common.network.ListenerName; /** * An object representing contextual information from the authentication session. See @@ -39,5 +38,5 @@ public interface AuthenticationContext { /** * Name of the listener used for the connection */ - ListenerName listenerName(); + String listenerName(); } 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 index 89dbb1c056d64..a111f21ec1302 100644 --- 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 @@ -18,13 +18,11 @@ import java.net.InetAddress; -import org.apache.kafka.common.network.ListenerName; - public class PlaintextAuthenticationContext implements AuthenticationContext { private final InetAddress clientAddress; - private final ListenerName listenerName; + private final String listenerName; - public PlaintextAuthenticationContext(InetAddress clientAddress, ListenerName listenerName) { + public PlaintextAuthenticationContext(InetAddress clientAddress, String listenerName) { this.clientAddress = clientAddress; this.listenerName = listenerName; } @@ -40,7 +38,7 @@ public InetAddress clientAddress() { } @Override - public ListenerName listenerName() { + public String listenerName() { return listenerName; } 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 index d87b3556b2dae..719d041770d13 100644 --- 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 @@ -18,17 +18,15 @@ import javax.security.sasl.SaslServer; -import org.apache.kafka.common.network.ListenerName; - import java.net.InetAddress; public class SaslAuthenticationContext implements AuthenticationContext { private final SaslServer server; private final SecurityProtocol securityProtocol; private final InetAddress clientAddress; - private final ListenerName listenerName; + private final String listenerName; - public SaslAuthenticationContext(SaslServer server, SecurityProtocol securityProtocol, InetAddress clientAddress, ListenerName listenerName) { + public SaslAuthenticationContext(SaslServer server, SecurityProtocol securityProtocol, InetAddress clientAddress, String listenerName) { this.server = server; this.securityProtocol = securityProtocol; this.clientAddress = clientAddress; @@ -50,7 +48,7 @@ public InetAddress clientAddress() { } @Override - public ListenerName listenerName() { + public String listenerName() { return listenerName; } } 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 index a370c84551eed..752041c13d4ef 100644 --- 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 @@ -18,16 +18,14 @@ import javax.net.ssl.SSLSession; -import org.apache.kafka.common.network.ListenerName; - import java.net.InetAddress; public class SslAuthenticationContext implements AuthenticationContext { private final SSLSession session; private final InetAddress clientAddress; - private final ListenerName listenerName; + private final String listenerName; - public SslAuthenticationContext(SSLSession session, InetAddress clientAddress, ListenerName listenerName) { + public SslAuthenticationContext(SSLSession session, InetAddress clientAddress, String listenerName) { this.session = session; this.clientAddress = clientAddress; this.listenerName = listenerName; @@ -48,7 +46,7 @@ public InetAddress clientAddress() { } @Override - public ListenerName listenerName() { + public String listenerName() { return listenerName; } } 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 0d8e47bbcfba6..f2cc621a2b842 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 @@ -282,7 +282,7 @@ public void authenticate() throws IOException { @Override public KafkaPrincipal principal() { - SaslAuthenticationContext context = new SaslAuthenticationContext(saslServer, securityProtocol, clientAddress(), listenerName); + SaslAuthenticationContext context = new SaslAuthenticationContext(saslServer, securityProtocol, clientAddress(), listenerName.value()); KafkaPrincipal principal = principalBuilder.build(context); if (ScramMechanism.isScram(saslMechanism) && Boolean.parseBoolean((String) saslServer.getNegotiatedProperty(ScramLoginModule.TOKEN_AUTH_CONFIG))) { principal.tokenAuthenticated(true); 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 index 346c7a8b8704a..c2b89fe6882f7 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java +++ b/clients/src/test/java/org/apache/kafka/common/network/ChannelBuildersTest.java @@ -51,7 +51,7 @@ public void testCreateOldPrincipalBuilder() throws Exception { assertTrue(OldPrincipalBuilder.configured); // test delegation - KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); + KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(OldPrincipalBuilder.PRINCIPAL_NAME, principal.getName()); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); } 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 index 9baee8be6c6b2..4aac7042ea828 100644 --- 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 @@ -18,7 +18,6 @@ import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.network.Authenticator; -import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.network.TransportLayer; import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder; import org.apache.kafka.common.security.kerberos.KerberosName; @@ -54,7 +53,7 @@ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); + KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -65,7 +64,7 @@ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception @Test public void testReturnAnonymousPrincipalForPlaintext() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); - assertEquals(KafkaPrincipal.ANONYMOUS, builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)))); + assertEquals(KafkaPrincipal.ANONYMOUS, builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name()))); builder.close(); } @@ -87,7 +86,7 @@ public void testUseOldPrincipalBuilderForSslIfProvided() throws Exception { DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); + KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -105,7 +104,7 @@ public void testUseSessionPeerPrincipalForSsl() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT))); + KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -125,7 +124,7 @@ public void testPrincipalBuilderScram() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server, - SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT))); + SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), SecurityProtocol.SASL_PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -148,7 +147,7 @@ public void testPrincipalBuilderGssapi() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(kerberosShortNamer); KafkaPrincipal principal = builder.build(new SaslAuthenticationContext(server, - SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), ListenerName.forSecurityProtocol(SecurityProtocol.SASL_PLAINTEXT))); + SecurityProtocol.SASL_PLAINTEXT, InetAddress.getLocalHost(), SecurityProtocol.SASL_PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); diff --git a/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala b/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala index 14861c07b9567..5b35f4790d716 100644 --- a/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala +++ b/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala @@ -57,16 +57,6 @@ class ListenerAndPrincipalBuilderTest extends BaseRequestTest { } assertEquals("CUSTOM", ListenerAndPrincipalBuilderTest.lastListenerName) } - - private def requestResponse(socket: Socket, clientId: String, correlationId: Int, requestBuilder: AbstractRequest.Builder[_ <: AbstractRequest]): Struct = { - val apiKey = requestBuilder.apiKey - val request = requestBuilder.build() - val header = new RequestHeader(apiKey, request.version, clientId, correlationId) - val response = requestAndReceive(socket, request.serialize(header).array) - val responseBuffer = skipResponseHeader(response) - apiKey.parseResponse(request.version, responseBuffer) - } - } object ListenerAndPrincipalBuilderTest { @@ -74,7 +64,7 @@ object ListenerAndPrincipalBuilderTest { class TestPrincipalBuilder extends KafkaPrincipalBuilder { override def build(context: AuthenticationContext): KafkaPrincipal = { - lastListenerName = context.listenerName.value + lastListenerName = context.listenerName KafkaPrincipal.ANONYMOUS } } From bada354b9f092475020c07dcfecac82b13050fca Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Fri, 1 Jun 2018 11:39:54 +0100 Subject: [PATCH 3/6] Updates following review --- .../network/PlaintextChannelBuilder.java | 2 +- .../common/network/SslTransportLayerTest.java | 15 +--- .../kafka/api/EndToEndAuthorizationTest.scala | 2 +- .../PlaintextEndToEndAuthorizationTest.scala | 20 +++++- .../ListenerAndPrincipalBuilderTest.scala | 71 ------------------- 5 files changed, 24 insertions(+), 86 deletions(-) delete mode 100644 core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala 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 ab65f5382c3f1..bbbd55bb217dd 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 @@ -34,7 +34,7 @@ public class PlaintextChannelBuilder implements ChannelBuilder { private static final Logger log = LoggerFactory.getLogger(PlaintextChannelBuilder.class); private Map configs; - private ListenerName listenerName; + private final ListenerName listenerName; /** * Constructs a plaintext channel builder. ListenerName is provided only 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 64c31575b2fe2..de3da8e439c35 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 @@ -429,15 +429,12 @@ public void testClientAuthenticationRequestedNotProvided() throws Exception { */ @Test public void testInvalidSecureRandomImplementation() throws Exception { - SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false); - try { + try (SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false)) { sslClientConfigs.put(SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG, "invalid"); channelBuilder.configure(sslClientConfigs); fail("SSL channel configured with invalid SecureRandom implementation"); } catch (KafkaException e) { // Expected exception - } finally { - channelBuilder.close(); } } @@ -446,15 +443,12 @@ public void testInvalidSecureRandomImplementation() throws Exception { */ @Test public void testInvalidTruststorePassword() throws Exception { - SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false); - try { + try (SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false)) { sslClientConfigs.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "invalid"); channelBuilder.configure(sslClientConfigs); fail("SSL channel configured with invalid truststore password"); } catch (KafkaException e) { // Expected exception - } finally { - channelBuilder.close(); } } @@ -463,15 +457,12 @@ public void testInvalidTruststorePassword() throws Exception { */ @Test public void testInvalidKeystorePassword() throws Exception { - SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false); - try { + try (SslChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false)) { sslClientConfigs.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "invalid"); channelBuilder.configure(sslClientConfigs); fail("SSL channel configured with invalid keystore password"); } catch (KafkaException e) { // Expected exception - } finally { - channelBuilder.close(); } } diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala index 4500d49e3377b..7a572764797a1 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala @@ -331,7 +331,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas } } - private def sendRecords(numRecords: Int, tp: TopicPartition) { + protected def sendRecords(numRecords: Int, tp: TopicPartition) { val futures = (0 until numRecords).map { i => val record = new ProducerRecord(tp.topic(), tp.partition(), s"$i".getBytes, s"$i".getBytes) debug(s"Sending this record: $record") diff --git a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala index 627934035c230..91ca30dc95897 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala @@ -19,13 +19,18 @@ package kafka.api import org.apache.kafka.common.config.internals.BrokerSecurityConfigs import org.apache.kafka.common.network.ListenerName import org.apache.kafka.common.security.auth._ -import org.junit.Before +import org.junit.{Before, Test} +import org.junit.Assert._ +import org.apache.kafka.common.errors.TopicAuthorizationException // This test case uses a separate listener for client and inter-broker communication, from // which we derive corresponding principals object PlaintextEndToEndAuthorizationTest { + var clientListenerName: String = "" + var serverListenerName: String = "" class TestClientPrincipalBuilder extends KafkaPrincipalBuilder { override def build(context: AuthenticationContext): KafkaPrincipal = { + clientListenerName = context.listenerName context match { case ctx: PlaintextAuthenticationContext if ctx.clientAddress != null => new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "client") @@ -37,6 +42,7 @@ object PlaintextEndToEndAuthorizationTest { class TestServerPrincipalBuilder extends KafkaPrincipalBuilder { override def build(context: AuthenticationContext): KafkaPrincipal = { + serverListenerName = context.listenerName context match { case ctx: PlaintextAuthenticationContext => new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "server") @@ -67,4 +73,16 @@ class PlaintextEndToEndAuthorizationTest extends EndToEndAuthorizationTest { super.setUp() } + @Test + def testListenerName() { + try { + sendRecords(1, tp) + fail("Should have thrown a TopicAuthorizationException") + } catch { + case tae: TopicAuthorizationException => //expected + } + assertEquals("CLIENT", PlaintextEndToEndAuthorizationTest.clientListenerName) + assertEquals("SERVER", PlaintextEndToEndAuthorizationTest.serverListenerName) + } + } diff --git a/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala b/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala deleted file mode 100644 index 5b35f4790d716..0000000000000 --- a/core/src/test/scala/unit/kafka/server/ListenerAndPrincipalBuilderTest.scala +++ /dev/null @@ -1,71 +0,0 @@ -/** - * Licensed 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.server - -import java.net.Socket -import java.util.Properties - -import kafka.utils.TestUtils -import org.apache.kafka.common.network.ListenerName -import org.apache.kafka.common.protocol.types.Struct -import org.apache.kafka.common.requests._ -import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, KafkaPrincipalBuilder} -import org.junit.Assert._ -import org.junit.{Before, Test} - -class ListenerAndPrincipalBuilderTest extends BaseRequestTest { - - override def numBrokers: Int = 1 - - override def propertyOverrides(properties: Properties): Unit = { - properties.put(KafkaConfig.PrincipalBuilderClassProp, classOf[ListenerAndPrincipalBuilderTest.TestPrincipalBuilder].getName) - properties.put(KafkaConfig.ListenersProp, "PLAINTEXT://localhost:0,CUSTOM://localhost:0") - properties.put(KafkaConfig.ListenerSecurityProtocolMapProp, "PLAINTEXT:PLAINTEXT,CUSTOM:PLAINTEXT") - } - - @Before - override def setUp() { - super.setUp() - - TestUtils.retry(10000) { - // controller to node connection using the default security.inter.broker.protocol - assertEquals("PLAINTEXT", ListenerAndPrincipalBuilderTest.lastListenerName) - } - } - - @Test - def testListenerNameFromAuthenticationContext() { - val port = anySocketServer.boundPort(new ListenerName("CUSTOM")) - val socket = new Socket("localhost", port) - try { - // using the simplest possible builder - requestResponse(socket, "clientId", 0, new ListGroupsRequest.Builder()) - } finally { - socket.close() - } - assertEquals("CUSTOM", ListenerAndPrincipalBuilderTest.lastListenerName) - } -} - -object ListenerAndPrincipalBuilderTest { - var lastListenerName = "" - - class TestPrincipalBuilder extends KafkaPrincipalBuilder { - override def build(context: AuthenticationContext): KafkaPrincipal = { - lastListenerName = context.listenerName - KafkaPrincipal.ANONYMOUS - } - } -} From d069e0aea6d9239e573adfc26ad180144340332f Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Mon, 4 Jun 2018 17:09:34 +0100 Subject: [PATCH 4/6] Added in comment to testListenerName --- .../kafka/api/PlaintextEndToEndAuthorizationTest.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala index 91ca30dc95897..6c802b98cec43 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala @@ -75,6 +75,7 @@ class PlaintextEndToEndAuthorizationTest extends EndToEndAuthorizationTest { @Test def testListenerName() { + // To check the client listener name, establish a session on the server by sending any request eg sendRecords try { sendRecords(1, tp) fail("Should have thrown a TopicAuthorizationException") From 4dd420366252fff8b0368cb8cf70b0c1ca3d645a Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Tue, 5 Jun 2018 11:21:39 +0100 Subject: [PATCH 5/6] Addressed latest comments --- .../network/PlaintextChannelBuilder.java | 13 ++++++----- .../common/network/SslChannelBuilder.java | 10 ++++++--- .../security/auth/AuthenticationContext.java | 2 +- .../auth/SslAuthenticationContext.java | 1 - .../DefaultKafkaPrincipalBuilderTest.java | 12 ++++++---- .../kafka/api/EndToEndAuthorizationTest.scala | 4 ++-- .../PlaintextEndToEndAuthorizationTest.scala | 22 +++++++++---------- 7 files changed, 35 insertions(+), 29 deletions(-) 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 bbbd55bb217dd..c7b80cbac63ff 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 @@ -33,12 +33,12 @@ public class PlaintextChannelBuilder implements ChannelBuilder { private static final Logger log = LoggerFactory.getLogger(PlaintextChannelBuilder.class); - private Map configs; private final ListenerName listenerName; + private Map configs; /** - * Constructs a plaintext channel builder. ListenerName is provided only - * for server channel builder and will be null for client channel builder. + * Constructs a plaintext channel builder. ListenerName is non-null whenever + * it's instantiated in the broker and null otherwise. */ public PlaintextChannelBuilder(ListenerName listenerName) { this.listenerName = listenerName; @@ -81,9 +81,10 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - // listenerName should only be null in Client mode - String listenerNameStr = listenerName != null ? listenerName.value() : null; - return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerNameStr)); + // listenerName should only be null in Client mode where principal() should not be called + if (listenerName == null) + throw new IllegalStateException("Unexpected call to principal() when listenerName is null"); + return principalBuilder.build(new PlaintextAuthenticationContext(clientAddress, listenerName.value())); } @Override 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 ae167f57193aa..b43e4c3a472fa 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 @@ -172,9 +172,13 @@ public void authenticate() throws IOException {} @Override public KafkaPrincipal principal() { InetAddress clientAddress = transportLayer.socketChannel().socket().getInetAddress(); - // listenerName should only be null in Client mode - String listenerNameStr = listenerName != null ? listenerName.value() : null; - SslAuthenticationContext context = new SslAuthenticationContext(transportLayer.sslSession(), clientAddress, listenerNameStr); + // listenerName should only be null in Client mode where principal() should not be called + if (listenerName == null) + throw new IllegalStateException("Unexpected call to principal() when listenerName is null"); + SslAuthenticationContext context = new SslAuthenticationContext( + transportLayer.sslSession(), + clientAddress, + listenerName.value()); return principalBuilder.build(context); } 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 index 3e644b202e06d..a8abea858960d 100644 --- 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 @@ -22,7 +22,7 @@ /** * An object representing contextual information from the authentication session. See * {@link PlaintextAuthenticationContext}, {@link SaslAuthenticationContext} - * and {@link SslAuthenticationContext}. + * and {@link SslAuthenticationContext}. This class is only used in the broker. */ public interface AuthenticationContext { /** 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 index 752041c13d4ef..88819f91cf82f 100644 --- 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 @@ -17,7 +17,6 @@ package org.apache.kafka.common.security.auth; import javax.net.ssl.SSLSession; - import java.net.InetAddress; public class SslAuthenticationContext implements AuthenticationContext { 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 index 4aac7042ea828..8b2e8b2629645 100644 --- 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 @@ -53,7 +53,8 @@ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); + KafkaPrincipal principal = builder.build(new PlaintextAuthenticationContext( + InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -64,7 +65,8 @@ public void testUseOldPrincipalBuilderForPlaintextIfProvided() throws Exception @Test public void testReturnAnonymousPrincipalForPlaintext() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); - assertEquals(KafkaPrincipal.ANONYMOUS, builder.build(new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name()))); + assertEquals(KafkaPrincipal.ANONYMOUS, builder.build( + new PlaintextAuthenticationContext(InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name()))); builder.close(); } @@ -86,7 +88,8 @@ public void testUseOldPrincipalBuilderForSslIfProvided() throws Exception { DefaultKafkaPrincipalBuilder builder = DefaultKafkaPrincipalBuilder.fromOldPrincipalBuilder(authenticator, transportLayer, oldPrincipalBuilder, null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); + KafkaPrincipal principal = builder.build( + new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); @@ -104,7 +107,8 @@ public void testUseSessionPeerPrincipalForSsl() throws Exception { DefaultKafkaPrincipalBuilder builder = new DefaultKafkaPrincipalBuilder(null); - KafkaPrincipal principal = builder.build(new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); + KafkaPrincipal principal = builder.build( + new SslAuthenticationContext(session, InetAddress.getLocalHost(), SecurityProtocol.PLAINTEXT.name())); assertEquals(KafkaPrincipal.USER_TYPE, principal.getPrincipalType()); assertEquals("foo", principal.getName()); diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala index 7a572764797a1..35f45cd82365c 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala @@ -331,7 +331,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas } } - protected def sendRecords(numRecords: Int, tp: TopicPartition) { + protected final def sendRecords(numRecords: Int, tp: TopicPartition) { val futures = (0 until numRecords).map { i => val record = new ProducerRecord(tp.topic(), tp.partition(), s"$i".getBytes, s"$i".getBytes) debug(s"Sending this record: $record") @@ -344,7 +344,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas } } - protected def consumeRecords(consumer: Consumer[Array[Byte], Array[Byte]], + protected final def consumeRecords(consumer: Consumer[Array[Byte], Array[Byte]], numRecords: Int = 1, startingOffset: Int = 0, topic: String = topic, diff --git a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala index 6c802b98cec43..699d70a8d2771 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala @@ -26,11 +26,13 @@ import org.apache.kafka.common.errors.TopicAuthorizationException // This test case uses a separate listener for client and inter-broker communication, from // which we derive corresponding principals object PlaintextEndToEndAuthorizationTest { - var clientListenerName: String = "" - var serverListenerName: String = "" + @volatile + private var clientListenerName = None: Option[String] + @volatile + private var serverListenerName = None: Option[String] class TestClientPrincipalBuilder extends KafkaPrincipalBuilder { override def build(context: AuthenticationContext): KafkaPrincipal = { - clientListenerName = context.listenerName + clientListenerName = Some(context.listenerName) context match { case ctx: PlaintextAuthenticationContext if ctx.clientAddress != null => new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "client") @@ -42,7 +44,7 @@ object PlaintextEndToEndAuthorizationTest { class TestServerPrincipalBuilder extends KafkaPrincipalBuilder { override def build(context: AuthenticationContext): KafkaPrincipal = { - serverListenerName = context.listenerName + serverListenerName = Some(context.listenerName) context match { case ctx: PlaintextAuthenticationContext => new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "server") @@ -76,14 +78,10 @@ class PlaintextEndToEndAuthorizationTest extends EndToEndAuthorizationTest { @Test def testListenerName() { // To check the client listener name, establish a session on the server by sending any request eg sendRecords - try { - sendRecords(1, tp) - fail("Should have thrown a TopicAuthorizationException") - } catch { - case tae: TopicAuthorizationException => //expected - } - assertEquals("CLIENT", PlaintextEndToEndAuthorizationTest.clientListenerName) - assertEquals("SERVER", PlaintextEndToEndAuthorizationTest.serverListenerName) + intercept[TopicAuthorizationException](sendRecords(1, tp)) + + assertEquals("CLIENT", PlaintextEndToEndAuthorizationTest.clientListenerName.get) + assertEquals("SERVER", PlaintextEndToEndAuthorizationTest.serverListenerName.get) } } From 132d4df346ef127ed8b1fae690b67041c525244e Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Tue, 5 Jun 2018 11:40:46 +0100 Subject: [PATCH 6/6] Fixed asserts on Option --- .../kafka/api/PlaintextEndToEndAuthorizationTest.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala index 699d70a8d2771..0b2fbcaf45908 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextEndToEndAuthorizationTest.scala @@ -80,8 +80,8 @@ class PlaintextEndToEndAuthorizationTest extends EndToEndAuthorizationTest { // To check the client listener name, establish a session on the server by sending any request eg sendRecords intercept[TopicAuthorizationException](sendRecords(1, tp)) - assertEquals("CLIENT", PlaintextEndToEndAuthorizationTest.clientListenerName.get) - assertEquals("SERVER", PlaintextEndToEndAuthorizationTest.serverListenerName.get) + assertEquals(Some("CLIENT"), PlaintextEndToEndAuthorizationTest.clientListenerName) + assertEquals(Some("SERVER"), PlaintextEndToEndAuthorizationTest.serverListenerName) } }