diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java index 79ad10c2f74bb..6ed6a08a91137 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java @@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.common.network.ChannelBuilders; +import org.apache.kafka.common.network.LoginType; import org.apache.kafka.common.network.Mode; import org.apache.kafka.common.protocol.SecurityProtocol; import org.apache.kafka.common.network.ChannelBuilder; @@ -76,7 +77,7 @@ public static ChannelBuilder createChannelBuilder(Map configs) { SecurityProtocol securityProtocol = SecurityProtocol.valueOf((String) configs.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); if (securityProtocol != SecurityProtocol.SSL && securityProtocol != SecurityProtocol.PLAINTEXT && securityProtocol != SecurityProtocol.SASL_PLAINTEXT) throw new ConfigException("Invalid SecurityProtocol " + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG); - return ChannelBuilders.create(securityProtocol, Mode.CLIENT, configs); + return ChannelBuilders.create(securityProtocol, Mode.CLIENT, LoginType.CLIENT, configs); } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java index 1d50e9a339fe7..bfbb47a2998e8 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java @@ -308,6 +308,7 @@ public class ConsumerConfig extends AbstractConfig { .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC) .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC) .define(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Type.LONG, SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Importance.LOW, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC) + .define(SaslConfigs.AUTH_TO_LOCAL, Type.LIST, SaslConfigs.DEFAULT_AUTH_TO_LOCAL, Importance.MEDIUM, SaslConfigs.AUTH_TO_LOCAL_DOC) .define(REQUEST_TIMEOUT_MS_CONFIG, Type.INT, 40 * 1000, diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java index ce19f88df31cf..bad830beca803 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java @@ -285,6 +285,7 @@ public class ProducerConfig extends AbstractConfig { .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC) .define(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER, Type.DOUBLE, SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER, Importance.LOW, SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC) .define(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Type.LONG, SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN, Importance.LOW, SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC) + .define(SaslConfigs.AUTH_TO_LOCAL, Type.LIST, SaslConfigs.DEFAULT_AUTH_TO_LOCAL, Importance.MEDIUM, SaslConfigs.AUTH_TO_LOCAL_DOC) /* default is set to be a bit lower than the server default (10 min), to avoid both client and server closing connection at same time */ .define(CONNECTIONS_MAX_IDLE_MS_CONFIG, Type.LONG, 9 * 60 * 1000, Importance.MEDIUM, CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_DOC) .define(PARTITIONER_CLASS_CONFIG, Type.CLASS, "org.apache.kafka.clients.producer.internals.DefaultPartitioner", Importance.MEDIUM, PARTITIONER_CLASS_DOC); 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 c8cfadf491eda..c4d9e4866dd57 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 @@ -13,6 +13,8 @@ package org.apache.kafka.common.config; +import java.util.Collections; +import java.util.List; public class SaslConfigs { /* @@ -41,4 +43,8 @@ public class SaslConfigs { public static final String SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC = "LoginThread sleep time between refresh attempts"; public static final long DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN = 1 * 60 * 1000L; + public static final String AUTH_TO_LOCAL = "kafka.security.auth.to.local"; + public static final String AUTH_TO_LOCAL_DOC = "Rules for the mapping between principal names and operating system user names"; + public static final List DEFAULT_AUTH_TO_LOCAL = Collections.singletonList("DEFAULT"); + } 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 74dfb1a58a522..1e5d8405045e2 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 @@ -23,13 +23,15 @@ private ChannelBuilders() { } /** * @param securityProtocol the securityProtocol - * @param mode the SSL mode, it must be non-null if `securityProcol` is `SSL` and it is ignored otherwise + * @param mode the mode, it must be non-null if `securityProtocol` is not `PLAINTEXT`; + * it is ignored otherwise + * @param loginType the loginType, it must be non-null if `securityProtocol` is SASL_*; it is ignored otherwise * @param configs client/server configs * @return the configured `ChannelBuilder` * @throws IllegalArgumentException if `mode` invariants described above is not maintained */ - public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode, Map configs) { - ChannelBuilder channelBuilder = null; + public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode, LoginType loginType, Map configs) { + ChannelBuilder channelBuilder; switch (securityProtocol) { case SSL: @@ -39,7 +41,9 @@ public static ChannelBuilder create(SecurityProtocol securityProtocol, Mode mode case SASL_SSL: case SASL_PLAINTEXT: requireNonNullMode(mode, securityProtocol); - channelBuilder = new SaslChannelBuilder(mode, securityProtocol); + if (loginType == null) + throw new IllegalArgumentException("`loginType` must be non-null if `securityProtocol` is `" + securityProtocol + "`"); + channelBuilder = new SaslChannelBuilder(mode, loginType, securityProtocol); break; case PLAINTEXT: case TRACE: 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 1612fe80a7d5a..ac436c37a0922 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 @@ -61,12 +61,12 @@ public Principal principal() throws IOException { } /** - * Does handshake of transportLayer and Authentication using configured authenticator + * Does handshake of transportLayer and authentication using configured authenticator */ public void prepare() throws IOException { if (!transportLayer.ready()) transportLayer.handshake(); - else if (transportLayer.ready() && !authenticator.complete()) + if (transportLayer.ready() && !authenticator.complete()) authenticator.authenticate(); } diff --git a/clients/src/main/java/org/apache/kafka/common/network/LoginType.java b/clients/src/main/java/org/apache/kafka/common/network/LoginType.java new file mode 100644 index 0000000000000..9216cb0c5c08f --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/network/LoginType.java @@ -0,0 +1,39 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.network; + +import org.apache.kafka.common.security.JaasUtils; + +/** + * The type of the login context, it should be SERVER for the broker and CLIENT for the clients (i.e. consumer and + * producer). It provides the the login context name which defines the section of the JAAS configuration file to be used + * for login. + */ +public enum LoginType { + CLIENT(JaasUtils.LOGIN_CONTEXT_CLIENT), + SERVER(JaasUtils.LOGIN_CONTEXT_SERVER); + + private final String contextName; + + LoginType(String contextName) { + this.contextName = contextName; + } + + public String contextName() { + return contextName; + } +} 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 b9ee9c2bd8d19..1dd1ecd11a2d6 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 @@ -68,10 +68,8 @@ public void close() { protected SSLTransportLayer buildTransportLayer(SSLFactory sslFactory, String id, SelectionKey key) throws IOException { SocketChannel socketChannel = (SocketChannel) key.channel(); - SSLTransportLayer transportLayer = new SSLTransportLayer(id, key, - sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(), - socketChannel.socket().getPort())); - transportLayer.startHandshake(); - return transportLayer; + return SSLTransportLayer.create(id, key, + sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(), + socketChannel.socket().getPort())); } } diff --git a/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java index 35ea9aae56abb..e7afa02927a4b 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java +++ b/clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java @@ -58,7 +58,14 @@ public class SSLTransportLayer implements TransportLayer { private ByteBuffer appReadBuffer; private ByteBuffer emptyBuf = ByteBuffer.allocate(0); - public SSLTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException { + public static SSLTransportLayer create(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException { + SSLTransportLayer transportLayer = new SSLTransportLayer(channelId, key, sslEngine); + transportLayer.startHandshake(); + return transportLayer; + } + + // Prefer `create`, only use this in tests + SSLTransportLayer(String channelId, SelectionKey key, SSLEngine sslEngine) throws IOException { this.channelId = channelId; this.key = key; this.socketChannel = (SocketChannel) key.channel(); 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 81998585ebcb4..53953c5504821 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 @@ -12,11 +12,16 @@ */ package org.apache.kafka.common.network; +import java.io.IOException; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; +import java.util.List; import java.util.Map; +import org.apache.kafka.common.config.SaslConfigs; +import org.apache.kafka.common.security.JaasUtils; import org.apache.kafka.common.security.auth.PrincipalBuilder; +import org.apache.kafka.common.security.kerberos.KerberosNameParser; import org.apache.kafka.common.security.kerberos.LoginManager; import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; import org.apache.kafka.common.security.authenticator.SaslServerAuthenticator; @@ -34,23 +39,35 @@ public class SaslChannelBuilder implements ChannelBuilder { private final SecurityProtocol securityProtocol; private final Mode mode; + private final LoginType loginType; private LoginManager loginManager; private PrincipalBuilder principalBuilder; private SSLFactory sslFactory; private Map configs; + private KerberosNameParser kerberosNameParser; - public SaslChannelBuilder(Mode mode, SecurityProtocol securityProtocol) { + public SaslChannelBuilder(Mode mode, LoginType loginType, SecurityProtocol securityProtocol) { this.mode = mode; + this.loginType = loginType; this.securityProtocol = securityProtocol; } public void configure(Map configs) throws KafkaException { try { this.configs = configs; - this.loginManager = LoginManager.acquireLoginManager(mode, configs); + this.loginManager = LoginManager.acquireLoginManager(loginType, configs); this.principalBuilder = (PrincipalBuilder) Utils.newInstance((Class) configs.get(SSLConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG)); this.principalBuilder.configure(configs); + + String defaultRealm; + try { + defaultRealm = JaasUtils.defaultRealm(); + } catch (Exception ke) { + defaultRealm = ""; + } + kerberosNameParser = new KerberosNameParser(defaultRealm, (List) configs.get(SaslConfigs.AUTH_TO_LOCAL)); + if (this.securityProtocol == SecurityProtocol.SASL_SSL) { this.sslFactory = new SSLFactory(mode); this.sslFactory.configure(this.configs); @@ -63,19 +80,13 @@ public void configure(Map configs) throws KafkaException { public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize) throws KafkaException { try { SocketChannel socketChannel = (SocketChannel) key.channel(); - TransportLayer transportLayer; - if (this.securityProtocol == SecurityProtocol.SASL_SSL) { - transportLayer = new SSLTransportLayer(id, key, - sslFactory.createSSLEngine(socketChannel.socket().getInetAddress().getHostName(), - socketChannel.socket().getPort())); - } else { - transportLayer = new PlaintextTransportLayer(key); - } + TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel); Authenticator authenticator; if (mode == Mode.SERVER) - authenticator = new SaslServerAuthenticator(id, loginManager.subject()); + authenticator = new SaslServerAuthenticator(id, loginManager.subject(), kerberosNameParser); else - authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(), socketChannel.socket().getInetAddress().getHostName()); + authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(), + socketChannel.socket().getInetAddress().getHostName(), kerberosNameParser); authenticator.configure(transportLayer, this.principalBuilder, this.configs); return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize); } catch (Exception e) { @@ -88,4 +99,15 @@ public void close() { this.principalBuilder.close(); this.loginManager.release(); } + + protected 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())); + } else { + return new PlaintextTransportLayer(key); + } + } + } diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java index 817049a8fbdf5..56f092fb6b769 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java +++ b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java @@ -23,11 +23,7 @@ import java.lang.reflect.Method; import java.io.IOException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - public class JaasUtils { - private static final Logger LOG = LoggerFactory.getLogger(JaasUtils.class); public static final String LOGIN_CONTEXT_SERVER = "KafkaServer"; public static final String LOGIN_CONTEXT_CLIENT = "KafkaClient"; public static final String SERVICE_NAME = "serviceName"; @@ -58,6 +54,10 @@ public static String defaultRealm() throws ClassNotFoundException, NoSuchMethodException, IllegalArgumentException, IllegalAccessException, InvocationTargetException { + + //TODO Find a way to avoid using these proprietary classes as access to Java 9 will block access by default + //due to the Jigsaw module system + Object kerbConf; Class classRef; Method getInstanceMethod; 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 29fbf8f2262e2..646af199950c0 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 @@ -46,9 +46,9 @@ import org.apache.kafka.common.network.TransportLayer; import org.apache.kafka.common.security.auth.KafkaPrincipal; import org.apache.kafka.common.security.auth.PrincipalBuilder; -import org.apache.kafka.common.security.kerberos.KerberosName; import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.security.kerberos.KerberosNameParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -64,6 +64,7 @@ public enum SaslState { private final String servicePrincipal; private final String host; private final String node; + private final KerberosNameParser kerberosNameParser; // assigned in `configure` private SaslClient saslClient; @@ -76,11 +77,12 @@ public enum SaslState { private SaslState saslState = SaslState.INITIAL; - public SaslClientAuthenticator(String node, Subject subject, String servicePrincipal, String host) throws IOException { + public SaslClientAuthenticator(String node, Subject subject, String servicePrincipal, String host, KerberosNameParser kerberosNameParser) throws IOException { this.node = node; this.subject = subject; this.host = host; this.servicePrincipal = servicePrincipal; + this.kerberosNameParser = kerberosNameParser; } public void configure(TransportLayer transportLayer, PrincipalBuilder principalBuilder, Map configs) throws KafkaException { @@ -89,7 +91,7 @@ public void configure(TransportLayer transportLayer, PrincipalBuilder principalB // determine client principal from subject. Principal clientPrincipal = subject.getPrincipals().iterator().next(); - this.clientPrincipalName = new KerberosName(clientPrincipal.getName()).toString(); + this.clientPrincipalName = kerberosNameParser.parse(clientPrincipal.getName()).toString(); this.saslClient = createSaslClient(); } catch (Exception e) { throw new KafkaException("Failed to configure SaslClientAuthenticator", e); 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 4d160bfd5ffde..fe07f08dccf0a 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 @@ -35,6 +35,7 @@ import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.security.kerberos.KerberosName; +import org.apache.kafka.common.security.kerberos.KerberosNameParser; import org.ietf.jgss.GSSContext; import org.ietf.jgss.GSSCredential; import org.ietf.jgss.GSSException; @@ -58,6 +59,7 @@ public class SaslServerAuthenticator implements Authenticator { private final SaslServer saslServer; private final Subject subject; private final String node; + private final KerberosNameParser kerberosNameParser; // assigned in `configure` private TransportLayer transportLayer; @@ -66,13 +68,14 @@ public class SaslServerAuthenticator implements Authenticator { private NetworkReceive netInBuffer; private NetworkSend netOutBuffer; - public SaslServerAuthenticator(String node, final Subject subject) throws IOException { + public SaslServerAuthenticator(String node, final Subject subject, KerberosNameParser kerberosNameParser) throws IOException { if (subject == null) throw new IllegalArgumentException("subject cannot be null"); if (subject.getPrincipals().isEmpty()) throw new IllegalArgumentException("subject must have at least one principal"); this.node = node; this.subject = subject; + this.kerberosNameParser = kerberosNameParser; saslServer = createSaslServer(); } @@ -82,11 +85,12 @@ public void configure(TransportLayer transportLayer, PrincipalBuilder principalB private SaslServer createSaslServer() throws IOException { // server is using a JAAS-authenticated subject: determine service principal name and hostname from kafka server's subject. - final SaslServerCallbackHandler saslServerCallbackHandler = new SaslServerCallbackHandler(Configuration.getConfiguration()); + final SaslServerCallbackHandler saslServerCallbackHandler = new SaslServerCallbackHandler( + Configuration.getConfiguration(), kerberosNameParser); final Principal servicePrincipal = subject.getPrincipals().iterator().next(); KerberosName kerberosName; try { - kerberosName = new KerberosName(servicePrincipal.getName()); + kerberosName = kerberosNameParser.parse(servicePrincipal.getName()); } catch (IllegalArgumentException e) { throw new KafkaException("Principal has name with unexpected format " + servicePrincipal); } 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 6a06cd910bf36..8474faf491edd 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 @@ -20,6 +20,7 @@ import java.io.IOException; +import org.apache.kafka.common.security.kerberos.KerberosNameParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.security.auth.callback.Callback; @@ -35,12 +36,13 @@ public class SaslServerCallbackHandler implements CallbackHandler { private static final Logger LOG = LoggerFactory.getLogger(SaslServerCallbackHandler.class); + private final KerberosNameParser kerberosNameParser; - public SaslServerCallbackHandler(Configuration configuration) throws IOException { + public SaslServerCallbackHandler(Configuration configuration, KerberosNameParser kerberosNameParser) throws IOException { AppConfigurationEntry[] configurationEntries = configuration.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_SERVER); - if (configurationEntries == null) throw new IOException("Could not find a 'KafkaServer' entry in this configuration: Kafka Server cannot start."); + this.kerberosNameParser = kerberosNameParser; } public void handle(Callback[] callbacks) throws UnsupportedCallbackException { @@ -66,7 +68,7 @@ private void handleAuthorizeCallback(AuthorizeCallback ac) { authorizationID); ac.setAuthorized(true); - KerberosName kerberosName = new KerberosName(authenticationID); + KerberosName kerberosName = kerberosNameParser.parse(authenticationID); try { String userName = kerberosName.shortName(); LOG.info("Setting authorizedID: {}", userName); diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosName.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosName.java index ab85f2ceb94ef..aef10db83ced1 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosName.java +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosName.java @@ -19,19 +19,10 @@ package org.apache.kafka.common.security.kerberos; import java.io.IOException; -import java.util.ArrayList; import java.util.List; -import java.util.regex.Matcher; -import java.util.regex.Pattern; -import org.apache.kafka.common.security.JaasUtils; - -/** - * This class implements parsing and handling of Kerberos principal names. In - * particular, it splits them apart and translates them down into local - * operating system names. - */ public class KerberosName { + /** The first component of the name */ private final String serviceName; /** The second component of the name. It may be null. */ @@ -39,78 +30,19 @@ public class KerberosName { /** The realm of the name. */ private final String realm; - /** - * A pattern that matches a Kerberos name with at most 3 components. - */ - private static final Pattern NAME_PARSER = Pattern.compile("([^/@]*)(/([^/@]*))?@([^/@]*)"); - - /** - * A pattern that matches a string without '$' and then a single - * parameter with $n. - */ - private static final Pattern PARAMETER_PATTERN = Pattern.compile("([^$]*)(\\$(\\d*))?"); - - /** - * A pattern for parsing a auth_to_local rule. - */ - private static final Pattern RULE_PARSER = Pattern.compile("\\s*((DEFAULT)|(RULE:\\[(\\d*):([^\\]]*)](\\(([^)]*)\\))?" + - "(s/([^/]*)/([^/]*)/(g)?)?))"); - - /** - * A pattern that recognizes simple/non-simple names. - */ - private static final Pattern NON_SIMPLE_PATTERN = Pattern.compile("[/@]"); - - /** - * The list of translation rules. - */ - private static final List RULES; - - private static final String DEFAULT_REALM; - - static { - String defaultRealm; - try { - defaultRealm = JaasUtils.defaultRealm(); - } catch (Exception ke) { - defaultRealm = ""; - } - DEFAULT_REALM = defaultRealm; - try { - String ruleString = System.getProperty("kafka.security.auth_to_local", "DEFAULT"); - RULES = parseRules(ruleString); - } catch (Exception e) { - throw new IllegalArgumentException("Could not configure Kerberos principal name mapping."); - } - } - - /** - * Create a name from the full Kerberos principal name. - * @param name - */ - public KerberosName(String name) { - Matcher match = NAME_PARSER.matcher(name); - if (!match.matches()) { - if (name.contains("@")) { - throw new IllegalArgumentException("Malformed Kerberos name: " + name); - } else { - serviceName = name; - hostName = null; - realm = null; - } - } else { - serviceName = match.group(1); - hostName = match.group(3); - realm = match.group(4); - } - } + /* Rules for the translation of the principal name into an operating system name */ + private final List authToLocalRules; /** - * Get the configured default realm. - * @return the default realm from the krb5.conf + * Creates an instance of `KerberosName` with the provided parameters. */ - public String getDefaultRealm() { - return DEFAULT_REALM; + public KerberosName(String serviceName, String hostName, String realm, List authToLocalRules) { + if (serviceName == null) + throw new IllegalArgumentException("serviceName must not be null"); + this.serviceName = serviceName; + this.hostName = hostName; + this.realm = realm; + this.authToLocalRules = authToLocalRules; } /** @@ -155,198 +87,6 @@ public String realm() { return realm; } - /** - * An encoding of a rule for translating kerberos names. - */ - private static class Rule { - private final boolean isDefault; - private final int numOfComponents; - private final String format; - private final Pattern match; - private final Pattern fromPattern; - private final String toPattern; - private final boolean repeat; - - Rule() { - isDefault = true; - numOfComponents = 0; - format = null; - match = null; - fromPattern = null; - toPattern = null; - repeat = false; - } - - Rule(int numOfComponents, String format, String match, String fromPattern, - String toPattern, boolean repeat) { - isDefault = false; - this.numOfComponents = numOfComponents; - this.format = format; - this.match = match == null ? null : Pattern.compile(match); - this.fromPattern = - fromPattern == null ? null : Pattern.compile(fromPattern); - this.toPattern = toPattern; - this.repeat = repeat; - } - - @Override - public String toString() { - StringBuilder buf = new StringBuilder(); - if (isDefault) { - buf.append("DEFAULT"); - } else { - buf.append("RULE:["); - buf.append(numOfComponents); - buf.append(':'); - buf.append(format); - buf.append(']'); - if (match != null) { - buf.append('('); - buf.append(match); - buf.append(')'); - } - if (fromPattern != null) { - buf.append("s/"); - buf.append(fromPattern); - buf.append('/'); - buf.append(toPattern); - buf.append('/'); - if (repeat) { - buf.append('g'); - } - } - } - return buf.toString(); - } - - /** - * Replace the numbered parameters of the form $n where n is from 1 to - * the length of params. Normal text is copied directly and $n is replaced - * by the corresponding parameter. - * @param format the string to replace parameters again - * @param params the list of parameters - * @return the generated string with the parameter references replaced. - * @throws BadFormatString - */ - static String replaceParameters(String format, - String[] params) throws BadFormatString { - Matcher match = PARAMETER_PATTERN.matcher(format); - int start = 0; - StringBuilder result = new StringBuilder(); - while (start < format.length() && match.find(start)) { - result.append(match.group(1)); - String paramNum = match.group(3); - if (paramNum != null) { - try { - int num = Integer.parseInt(paramNum); - if (num < 0 || num > params.length) { - throw new BadFormatString("index " + num + " from " + format + - " is outside of the valid range 0 to " + - (params.length - 1)); - } - result.append(params[num]); - } catch (NumberFormatException nfe) { - throw new BadFormatString("bad format in username mapping in " + - paramNum, nfe); - } - - } - start = match.end(); - } - return result.toString(); - } - - /** - * Replace the matches of the from pattern in the base string with the value - * of the to string. - * @param base the string to transform - * @param from the pattern to look for in the base string - * @param to the string to replace matches of the pattern with - * @param repeat whether the substitution should be repeated - * @return - */ - static String replaceSubstitution(String base, Pattern from, String to, - boolean repeat) { - Matcher match = from.matcher(base); - if (repeat) { - return match.replaceAll(to); - } else { - return match.replaceFirst(to); - } - } - - /** - * Try to apply this rule to the given name represented as a parameter - * array. - * @param params first element is the realm, second and later elements are - * are the components of the name "a/b@FOO" -> {"FOO", "a", "b"} - * @return the short name if this rule applies or null - * @throws IOException throws if something is wrong with the rules - */ - String apply(String[] params) throws IOException { - String result = null; - if (isDefault) { - if (DEFAULT_REALM.equals(params[0])) { - result = params[1]; - } - } else if (params.length - 1 == numOfComponents) { - String base = replaceParameters(format, params); - if (match == null || match.matcher(base).matches()) { - if (fromPattern == null) { - result = base; - } else { - result = replaceSubstitution(base, fromPattern, toPattern, repeat); - } - } - } - if (result != null && NON_SIMPLE_PATTERN.matcher(result).find()) { - throw new NoMatchingRule("Non-simple name " + result + - " after auth_to_local rule " + this); - } - return result; - } - } - - static List parseRules(String rules) { - List result = new ArrayList(); - String remaining = rules.trim(); - while (remaining.length() > 0) { - Matcher matcher = RULE_PARSER.matcher(remaining); - if (!matcher.lookingAt()) { - throw new IllegalArgumentException("Invalid rule: " + remaining); - } - if (matcher.group(2) != null) { - result.add(new Rule()); - } else { - result.add(new Rule(Integer.parseInt(matcher.group(4)), - matcher.group(5), - matcher.group(7), - matcher.group(9), - matcher.group(10), - "g".equals(matcher.group(11)))); - } - remaining = remaining.substring(matcher.end()); - } - return result; - } - - @SuppressWarnings("serial") - public static class BadFormatString extends IOException { - BadFormatString(String msg) { - super(msg); - } - BadFormatString(String msg, Throwable err) { - super(msg, err); - } - } - - @SuppressWarnings("serial") - public static class NoMatchingRule extends IOException { - NoMatchingRule(String msg) { - super(msg); - } - } - /** * Get the translation of the principal name into an operating system * user name. @@ -363,7 +103,7 @@ public String shortName() throws IOException { } else { params = new String[]{realm, serviceName, hostName}; } - for (Rule r : RULES) { + for (KerberosRule r : authToLocalRules) { String result = r.apply(params); if (result != null) return result; diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosNameParser.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosNameParser.java new file mode 100644 index 0000000000000..95eb170e90599 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosNameParser.java @@ -0,0 +1,103 @@ +/** + * 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.kerberos; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * This class implements parsing and handling of Kerberos principal names. In + * particular, it splits them apart and translates them down into local + * operating system names. + */ +public class KerberosNameParser { + + /** + * A pattern that matches a Kerberos name with at most 3 components. + */ + private static final Pattern NAME_PARSER = Pattern.compile("([^/@]*)(/([^/@]*))?@([^/@]*)"); + + /** + * A pattern for parsing a auth_to_local rule. + */ + private static final Pattern RULE_PARSER = Pattern.compile("((DEFAULT)|(RULE:\\[(\\d*):([^\\]]*)](\\(([^)]*)\\))?(s/([^/]*)/([^/]*)/(g)?)?))"); + + /** + * The list of translation rules. + */ + private final List authToLocalRules; + + public KerberosNameParser(String defaultRealm, List authToLocalRules) { + this.authToLocalRules = parseRules(defaultRealm, authToLocalRules); + } + + /** + * Create a name from the full Kerberos principal name. + */ + public KerberosName parse(String principalName) { + Matcher match = NAME_PARSER.matcher(principalName); + if (!match.matches()) { + if (principalName.contains("@")) { + throw new IllegalArgumentException("Malformed Kerberos name: " + principalName); + } else { + return new KerberosName(principalName, null, null, authToLocalRules); + } + } else { + return new KerberosName(match.group(1), match.group(3), match.group(4), authToLocalRules); + } + } + + private static List parseRules(String defaultRealm, List rules) { + List result = new ArrayList<>(); + for (String rule : rules) { + Matcher matcher = RULE_PARSER.matcher(rule); + if (!matcher.lookingAt()) { + throw new IllegalArgumentException("Invalid rule: " + rule); + } + if (rule.length() != matcher.end()) + throw new IllegalArgumentException("Invalid rule: `" + rule + "`, unmatched substring: `" + rule.substring(matcher.end()) + "`"); + if (matcher.group(2) != null) { + result.add(new KerberosRule(defaultRealm)); + } else { + result.add(new KerberosRule(defaultRealm, + Integer.parseInt(matcher.group(4)), + matcher.group(5), + matcher.group(7), + matcher.group(9), + matcher.group(10), + "g".equals(matcher.group(11)))); + + } + } + return result; + } + + public static class BadFormatString extends IOException { + BadFormatString(String msg) { + super(msg); + } + BadFormatString(String msg, Throwable err) { + super(msg, err); + } + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java new file mode 100644 index 0000000000000..c1789db400532 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java @@ -0,0 +1,189 @@ +/** + * 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.kerberos; + +import java.io.IOException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * An encoding of a rule for translating kerberos names. + */ +class KerberosRule { + + /** + * A pattern that matches a string without '$' and then a single + * parameter with $n. + */ + private static final Pattern PARAMETER_PATTERN = Pattern.compile("([^$]*)(\\$(\\d*))?"); + + /** + * A pattern that recognizes simple/non-simple names. + */ + private static final Pattern NON_SIMPLE_PATTERN = Pattern.compile("[/@]"); + + private final String defaultRealm; + private final boolean isDefault; + private final int numOfComponents; + private final String format; + private final Pattern match; + private final Pattern fromPattern; + private final String toPattern; + private final boolean repeat; + + KerberosRule(String defaultRealm) { + this.defaultRealm = defaultRealm; + isDefault = true; + numOfComponents = 0; + format = null; + match = null; + fromPattern = null; + toPattern = null; + repeat = false; + } + + KerberosRule(String defaultRealm, int numOfComponents, String format, String match, String fromPattern, + String toPattern, boolean repeat) { + this.defaultRealm = defaultRealm; + isDefault = false; + this.numOfComponents = numOfComponents; + this.format = format; + this.match = match == null ? null : Pattern.compile(match); + this.fromPattern = + fromPattern == null ? null : Pattern.compile(fromPattern); + this.toPattern = toPattern; + this.repeat = repeat; + } + + @Override + public String toString() { + StringBuilder buf = new StringBuilder(); + if (isDefault) { + buf.append("DEFAULT"); + } else { + buf.append("RULE:["); + buf.append(numOfComponents); + buf.append(':'); + buf.append(format); + buf.append(']'); + if (match != null) { + buf.append('('); + buf.append(match); + buf.append(')'); + } + if (fromPattern != null) { + buf.append("s/"); + buf.append(fromPattern); + buf.append('/'); + buf.append(toPattern); + buf.append('/'); + if (repeat) { + buf.append('g'); + } + } + } + return buf.toString(); + } + + /** + * Replace the numbered parameters of the form $n where n is from 1 to + * the length of params. Normal text is copied directly and $n is replaced + * by the corresponding parameter. + * @param format the string to replace parameters again + * @param params the list of parameters + * @return the generated string with the parameter references replaced. + * @throws KerberosNameParser.BadFormatString + */ + static String replaceParameters(String format, + String[] params) throws KerberosNameParser.BadFormatString { + Matcher match = PARAMETER_PATTERN.matcher(format); + int start = 0; + StringBuilder result = new StringBuilder(); + while (start < format.length() && match.find(start)) { + result.append(match.group(1)); + String paramNum = match.group(3); + if (paramNum != null) { + try { + int num = Integer.parseInt(paramNum); + if (num < 0 || num > params.length) { + throw new KerberosNameParser.BadFormatString("index " + num + " from " + format + + " is outside of the valid range 0 to " + + (params.length - 1)); + } + result.append(params[num]); + } catch (NumberFormatException nfe) { + throw new KerberosNameParser.BadFormatString("bad format in username mapping in " + + paramNum, nfe); + } + + } + start = match.end(); + } + return result.toString(); + } + + /** + * Replace the matches of the from pattern in the base string with the value + * of the to string. + * @param base the string to transform + * @param from the pattern to look for in the base string + * @param to the string to replace matches of the pattern with + * @param repeat whether the substitution should be repeated + * @return + */ + static String replaceSubstitution(String base, Pattern from, String to, + boolean repeat) { + Matcher match = from.matcher(base); + if (repeat) { + return match.replaceAll(to); + } else { + return match.replaceFirst(to); + } + } + + /** + * Try to apply this rule to the given name represented as a parameter + * array. + * @param params first element is the realm, second and later elements are + * are the components of the name "a/b@FOO" -> {"FOO", "a", "b"} + * @return the short name if this rule applies or null + * @throws IOException throws if something is wrong with the rules + */ + String apply(String[] params) throws IOException { + String result = null; + if (isDefault) { + if (defaultRealm.equals(params[0])) { + result = params[1]; + } + } else if (params.length - 1 == numOfComponents) { + String base = replaceParameters(format, params); + if (match == null || match.matcher(base).matches()) { + if (fromPattern == null) { + result = base; + } else { + result = replaceSubstitution(base, fromPattern, toPattern, repeat); + } + } + } + if (result != null && NON_SIMPLE_PATTERN.matcher(result).find()) { + throw new NoMatchingRule("Non-simple name " + result + " after auth_to_local rule " + this); + } + return result; + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java index 950413177e540..66d3169fec9e4 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java @@ -21,28 +21,24 @@ import javax.security.auth.Subject; import javax.security.auth.login.LoginException; import java.io.IOException; +import java.util.EnumMap; import java.util.Map; -import org.apache.kafka.common.network.Mode; +import org.apache.kafka.common.network.LoginType; import org.apache.kafka.common.security.JaasUtils; public class LoginManager { - private static LoginManager serverInstance; - private static LoginManager clientInstance; + private static final EnumMap CACHED_INSTANCES = new EnumMap(LoginType.class); private final Login login; private final String serviceName; - private final Mode mode; + private final LoginType loginType; private int refCount; - private LoginManager(Mode mode, Map configs) throws IOException, LoginException { - String loginContext; - if (mode == Mode.SERVER) - loginContext = JaasUtils.LOGIN_CONTEXT_SERVER; - else - loginContext = JaasUtils.LOGIN_CONTEXT_CLIENT; - this.mode = mode; + private LoginManager(LoginType loginType, Map configs) throws IOException, LoginException { + this.loginType = loginType; + String loginContext = loginType.contextName(); this.serviceName = JaasUtils.jaasConfig(loginContext, JaasUtils.SERVICE_NAME); login = new Login(loginContext, configs); login.startThreadIfNeeded(); @@ -57,22 +53,20 @@ private LoginManager(Mode mode, Map configs) throws IOException, Logi * * This is a bit ugly and it would be nicer if we could pass the `LoginManager` to `ChannelBuilders.create` and * shut it down when the broker or clients are closed. It's straightforward to do the former, but it's more - * complicated to do the latter without making the consumer API a bit more complex. + * complicated to do the latter without making the consumer API more complex. + * + * @param loginType the type of the login context, it should be SERVER for the broker and CLIENT for the clients + * (i.e. consumer and producer) + * @param configs configuration as key/value pairs */ - public static final LoginManager acquireLoginManager(Mode mode, Map configs) throws IOException, LoginException { + public static final LoginManager acquireLoginManager(LoginType loginType, Map configs) throws IOException, LoginException { synchronized (LoginManager.class) { - switch (mode) { - case SERVER: - if (serverInstance == null) - serverInstance = new LoginManager(mode, configs); - return serverInstance.acquire(); - case CLIENT: - if (clientInstance == null) - clientInstance = new LoginManager(mode, configs); - return clientInstance.acquire(); - default: - throw new IllegalArgumentException("Unsupported `mode` received " + mode); + LoginManager loginManager = CACHED_INSTANCES.get(loginType); + if (loginManager == null) { + loginManager = new LoginManager(loginType, configs); + CACHED_INSTANCES.put(loginType, loginManager); } + return loginManager.acquire(); } } @@ -97,13 +91,7 @@ public void release() { if (refCount == 0) throw new IllegalStateException("release called on LoginManager with refCount == 0"); else if (refCount == 1) { - if (mode == Mode.SERVER) - serverInstance = null; - else if (mode == Mode.CLIENT) - clientInstance = null; - else - throw new IllegalStateException("Unsupported `mode` " + mode); - + CACHED_INSTANCES.remove(loginType); login.shutdown(); } --refCount; diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java new file mode 100644 index 0000000000000..6c2d2677c31d7 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java @@ -0,0 +1,27 @@ +/** + * 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.kerberos; + +import java.io.IOException; + +public class NoMatchingRule extends IOException { + NoMatchingRule(String msg) { + super(msg); + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java b/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java new file mode 100644 index 0000000000000..9781f6d159d52 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java @@ -0,0 +1,59 @@ +/** + * 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.kerberos; + +import org.junit.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class KerberosNameTest { + + @Test + public void testParse() throws IOException { + List rules = new ArrayList<>(Arrays.asList( + "RULE:[1:$1](App\\..*)s/App\\.(.*)/$1/g", + "RULE:[2:$1](App\\..*)s/App\\.(.*)/$1/g", + "DEFAULT" + )); + KerberosNameParser parser = new KerberosNameParser("REALM.COM", rules); + + KerberosName name = parser.parse("App.service-name/example.com@REALM.COM"); + assertEquals("App.service-name", name.serviceName()); + assertEquals("example.com", name.hostName()); + assertEquals("REALM.COM", name.realm()); + assertEquals("service-name", name.shortName()); + + name = parser.parse("App.service-name@REALM.COM"); + assertEquals("App.service-name", name.serviceName()); + assertNull(name.hostName()); + assertEquals("REALM.COM", name.realm()); + assertEquals("service-name", name.shortName()); + + name = parser.parse("user/host@REALM.COM"); + assertEquals("user", name.serviceName()); + assertEquals("host", name.hostName()); + assertEquals("REALM.COM", name.realm()); + assertEquals("user", name.shortName()); + } +} diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 5a26698e5e4ba..dae20efa1be48 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -21,7 +21,7 @@ import kafka.utils._ import org.apache.kafka.clients.{ClientResponse, ClientRequest, ManualMetadataUpdater, NetworkClient} import org.apache.kafka.common.{TopicPartition, Node} import org.apache.kafka.common.metrics.Metrics -import org.apache.kafka.common.network.{Selectable, ChannelBuilders, Selector, NetworkReceive, Mode} +import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, Selector, NetworkReceive, Mode} import org.apache.kafka.common.protocol.{SecurityProtocol, ApiKeys} import org.apache.kafka.common.requests._ import org.apache.kafka.common.utils.Time @@ -96,7 +96,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf "controller-channel", Map("broker-id" -> broker.id.toString).asJava, false, - ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, config.channelConfigs) + ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, config.channelConfigs) ) new NetworkClient( selector, diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala index f09272fe45621..1066fbee1a5c0 100644 --- a/core/src/main/scala/kafka/network/SocketServer.scala +++ b/core/src/main/scala/kafka/network/SocketServer.scala @@ -33,7 +33,7 @@ import kafka.server.KafkaConfig import kafka.utils._ import org.apache.kafka.common.MetricName import org.apache.kafka.common.metrics._ -import org.apache.kafka.common.network.{Mode, Selector => KSelector, ChannelBuilders, InvalidReceiveException} +import org.apache.kafka.common.network.{Selector => KSelector, LoginType, Mode, ChannelBuilders, InvalidReceiveException} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.protocol.SecurityProtocol import org.apache.kafka.common.protocol.types.SchemaException @@ -377,7 +377,7 @@ private[kafka] class Processor(val id: Int, private val newConnections = new ConcurrentLinkedQueue[SocketChannel]() private val inflightResponses = mutable.Map[String, RequestChannel.Response]() - private val channelBuilder = ChannelBuilders.create(protocol, Mode.SERVER, channelConfigs) + private val channelBuilder = ChannelBuilders.create(protocol, Mode.SERVER, LoginType.SERVER, channelConfigs) private val metricTags = new util.HashMap[String, String]() metricTags.put("networkProcessor", id.toString) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 729fc3841b9d8..95949d4371484 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -180,6 +180,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 AuthToLocal = SaslConfigs.DEFAULT_AUTH_TO_LOCAL } @@ -334,6 +335,7 @@ object KafkaConfig { 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 AuthToLocalProp = SaslConfigs.AUTH_TO_LOCAL /* Documentation */ /** ********* Zookeeper Configuration ***********/ @@ -510,6 +512,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 AuthToLocalDoc = SaslConfigs.AUTH_TO_LOCAL_DOC private val configDef = { import ConfigDef.Importance._ @@ -672,6 +675,7 @@ object KafkaConfig { .define(SaslKerberosTicketRenewWindowFactorProp, DOUBLE, Defaults.SaslKerberosTicketRenewWindowFactor, MEDIUM, SaslKerberosTicketRenewWindowFactorDoc) .define(SaslKerberosTicketRenewJitterProp, DOUBLE, Defaults.SaslKerberosTicketRenewJitter, MEDIUM, SaslKerberosTicketRenewJitterDoc) .define(SaslKerberosMinTimeBeforeReloginProp, LONG, Defaults.SaslKerberosMinTimeBeforeRelogin, MEDIUM, SaslKerberosMinTimeBeforeReloginDoc) + .define(AuthToLocalProp, LIST, Defaults.AuthToLocal, MEDIUM, AuthToLocalDoc) } @@ -838,6 +842,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka val saslKerberosTicketRenewWindowFactor = getDouble(KafkaConfig.SaslKerberosTicketRenewWindowFactorProp) val saslKerberosTicketRenewJitter = getDouble(KafkaConfig.SaslKerberosTicketRenewJitterProp) val saslKerberosMinTimeBeforeRelogin = getLong(KafkaConfig.SaslKerberosMinTimeBeforeReloginProp) + val authToLocal = getList(KafkaConfig.AuthToLocalProp) /** ********* Quota Configuration **************/ val producerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp) @@ -982,6 +987,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka channelConfigs.put(SaslKerberosTicketRenewWindowFactorProp, saslKerberosTicketRenewWindowFactor) channelConfigs.put(SaslKerberosTicketRenewJitterProp, saslKerberosTicketRenewJitter) channelConfigs.put(SaslKerberosMinTimeBeforeReloginProp, saslKerberosMinTimeBeforeRelogin) + channelConfigs.put(AuthToLocalProp, authToLocal) channelConfigs } diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 9ad8e6b469f96..1e1b1a18cb06c 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -21,7 +21,7 @@ import java.net.{SocketTimeoutException} import java.util import kafka.admin._ -import kafka.api.{KAFKA_090, ApiVersion} +import kafka.api.KAFKA_090 import kafka.log.LogConfig import kafka.log.CleanerConfig import kafka.log.LogManager @@ -34,11 +34,10 @@ import kafka.utils._ import org.apache.kafka.clients.{ManualMetadataUpdater, ClientRequest, NetworkClient} import org.apache.kafka.common.Node import org.apache.kafka.common.metrics._ -import org.apache.kafka.common.network.{Selectable, ChannelBuilders, NetworkReceive, Selector, Mode} +import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, NetworkReceive, Selector, Mode} import org.apache.kafka.common.protocol.{Errors, ApiKeys, SecurityProtocol} import org.apache.kafka.common.metrics.{JmxReporter, Metrics} import org.apache.kafka.common.requests.{ControlledShutdownResponse, ControlledShutdownRequest, RequestSend} -import org.apache.kafka.common.security.ssl.SSLFactory import org.apache.kafka.common.utils.AppInfoParser import scala.collection.mutable @@ -304,7 +303,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr "kafka-server-controlled-shutdown", Map.empty.asJava, false, - ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, config.channelConfigs) + ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, config.channelConfigs) ) new NetworkClient( selector, diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala index 7123ded20a788..a80366097a05f 100644 --- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala +++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala @@ -27,7 +27,7 @@ import kafka.api.KAFKA_090 import kafka.common.{KafkaStorageException, TopicAndPartition} import ReplicaFetcherThread._ import org.apache.kafka.clients.{ManualMetadataUpdater, NetworkClient, ClientRequest, ClientResponse} -import org.apache.kafka.common.network.{Selectable, ChannelBuilders, NetworkReceive, Selector, Mode} +import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, NetworkReceive, Selector, Mode} import org.apache.kafka.common.requests.{ListOffsetResponse, FetchResponse, RequestSend, AbstractRequest, ListOffsetRequest} import org.apache.kafka.common.requests.{FetchRequest => JFetchRequest} import org.apache.kafka.common.{Node, TopicPartition} @@ -74,7 +74,7 @@ class ReplicaFetcherThread(name: String, "replica-fetcher", Map("broker-id" -> sourceBroker.id.toString).asJava, false, - ChannelBuilders.create(brokerConfig.interBrokerSecurityProtocol, Mode.CLIENT, brokerConfig.channelConfigs) + ChannelBuilders.create(brokerConfig.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, brokerConfig.channelConfigs) ) new NetworkClient( selector, diff --git a/core/src/test/scala/integration/kafka/api/SSLConsumerTest.scala b/core/src/test/scala/integration/kafka/api/SSLConsumerTest.scala index 69b2958b5cde1..15b91dd9df385 100644 --- a/core/src/test/scala/integration/kafka/api/SSLConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/SSLConsumerTest.scala @@ -61,7 +61,7 @@ class SSLConsumerTest extends KafkaServerTestHarness with Logging { val producers = Buffer[KafkaProducer[Array[Byte], Array[Byte]]]() def generateConfigs() = - TestUtils.createBrokerConfigs(numServers, zkConnect, false, enableSSL=true, trustStoreFile=Some(trustStoreFile)).map(KafkaConfig.fromProps(_, overridingProps)) + TestUtils.createBrokerConfigs(numServers, zkConnect, false, enableSsl=true, trustStoreFile=Some(trustStoreFile)).map(KafkaConfig.fromProps(_, overridingProps)) val topic = "topic" val part = 0 diff --git a/core/src/test/scala/integration/kafka/api/SSLProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/SSLProducerSendTest.scala index 967c9f443a612..aef1842c39485 100644 --- a/core/src/test/scala/integration/kafka/api/SSLProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/SSLProducerSendTest.scala @@ -43,7 +43,7 @@ class SSLProducerSendTest extends KafkaServerTestHarness { overridingProps.put(KafkaConfig.NumPartitionsProp, 4.toString) def generateConfigs() = - TestUtils.createBrokerConfigs(numServers, zkConnect, false, enableSSL=true, trustStoreFile=Some(trustStoreFile)).map(KafkaConfig.fromProps(_, overridingProps)) + TestUtils.createBrokerConfigs(numServers, zkConnect, false, enableSsl=true, trustStoreFile=Some(trustStoreFile)).map(KafkaConfig.fromProps(_, overridingProps)) private var consumer1: SimpleConsumer = null private var consumer2: SimpleConsumer = null diff --git a/core/src/test/scala/integration/kafka/api/SaslIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/SaslIntegrationTest.scala index afa6db50f0b3b..c908b890a2718 100644 --- a/core/src/test/scala/integration/kafka/api/SaslIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/SaslIntegrationTest.scala @@ -29,7 +29,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.protocol.SecurityProtocol -import kafka.integration.KafkaServerTestHarness import kafka.utils.{TestUtils, Logging} import kafka.server.{KafkaConfig, KafkaServer} @@ -73,7 +72,7 @@ class SaslIntegrationTest extends SaslTestHarness with Logging { @Before override def setUp() { super.setUp() - val props = TestUtils.createBrokerConfig(numServers, zkConnect, false, enableSasl=true) + val props = TestUtils.createBrokerConfig(numServers, zkConnect, false, enableSaslPlaintext=true) val config = KafkaConfig.fromProps(props, overridingProps) servers = Buffer(TestUtils.createServer(config)) diff --git a/core/src/test/scala/integration/kafka/api/SaslTestHarness.scala b/core/src/test/scala/integration/kafka/api/SaslTestHarness.scala index 27dce9f04eb24..6dce776a3efa8 100644 --- a/core/src/test/scala/integration/kafka/api/SaslTestHarness.scala +++ b/core/src/test/scala/integration/kafka/api/SaslTestHarness.scala @@ -18,6 +18,7 @@ import javax.security.auth.login.Configuration import kafka.utils.TestUtils import kafka.zk.ZooKeeperTestHarness import org.apache.hadoop.minikdc.MiniKdc +import org.apache.kafka.common.security.JaasUtils import org.junit.{After, Before} trait SaslTestHarness extends ZooKeeperTestHarness { @@ -27,6 +28,8 @@ trait SaslTestHarness extends ZooKeeperTestHarness { @Before override def setUp() { + // Clean-up global configuration set by other tests + Configuration.setConfiguration(null) val keytabFile = TestUtils.tempFile() val jaasFile = TestUtils.tempFile() @@ -43,7 +46,7 @@ trait SaslTestHarness extends ZooKeeperTestHarness { kdc.start() kdc.createPrincipal(keytabFile, "client", "kafka/localhost") - System.setProperty("java.security.auth.login.config", jaasFile.getAbsolutePath) + System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasFile.getAbsolutePath) super.setUp } @@ -51,7 +54,7 @@ trait SaslTestHarness extends ZooKeeperTestHarness { override def tearDown() { super.tearDown kdc.stop() - System.clearProperty("java.security.auth.login.config") + System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) Configuration.setConfiguration(null) } } diff --git a/core/src/test/scala/unit/kafka/integration/BaseTopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/BaseTopicMetadataTest.scala index 20a4068228395..4430da7a08172 100644 --- a/core/src/test/scala/unit/kafka/integration/BaseTopicMetadataTest.scala +++ b/core/src/test/scala/unit/kafka/integration/BaseTopicMetadataTest.scala @@ -39,17 +39,23 @@ abstract class BaseTopicMetadataTest extends ZooKeeperTestHarness { var adHocConfigs: Seq[KafkaConfig] = null val numConfigs: Int = 4 - /* If this is `Some`, SSL will be enabled */ + // This should be defined if `securityProtocol` uses SSL (eg SSL, SASL_SSL) protected def trustStoreFile: Option[File] + protected def securityProtocol: SecurityProtocol @Before override def setUp() { super.setUp() - val props = createBrokerConfigs(numConfigs, zkConnect, enableSSL = trustStoreFile.isDefined, trustStoreFile = trustStoreFile) + val props = createBrokerConfigs(numConfigs, zkConnect, interBrokerSecurityProtocol = Some(securityProtocol), + trustStoreFile = trustStoreFile) val configs: Seq[KafkaConfig] = props.map(KafkaConfig.fromProps) adHocConfigs = configs.takeRight(configs.size - 1) // Started and stopped by individual test cases server1 = TestUtils.createServer(configs.head) - brokerEndPoints = Seq(new Broker(server1.config.brokerId, server1.config.hostName, server1.boundPort()).getBrokerEndPoint(SecurityProtocol.PLAINTEXT)) + brokerEndPoints = Seq( + // We are using the Scala clients and they don't support SSL. Once we move to the Java ones, we should use + // `securityProtocol` instead of PLAINTEXT below + new BrokerEndPoint(server1.config.brokerId, server1.config.hostName, server1.boundPort(SecurityProtocol.PLAINTEXT)) + ) } @After diff --git a/core/src/test/scala/unit/kafka/integration/PlaintextTopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/PlaintextTopicMetadataTest.scala index 176d251993205..55c12b5c8a86c 100644 --- a/core/src/test/scala/unit/kafka/integration/PlaintextTopicMetadataTest.scala +++ b/core/src/test/scala/unit/kafka/integration/PlaintextTopicMetadataTest.scala @@ -17,7 +17,10 @@ package kafka.integration +import org.apache.kafka.common.protocol.SecurityProtocol + class PlaintextTopicMetadataTest extends BaseTopicMetadataTest { + protected def securityProtocol = SecurityProtocol.PLAINTEXT protected def trustStoreFile = None } diff --git a/core/src/test/scala/unit/kafka/integration/SaslPlaintextTopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/SaslPlaintextTopicMetadataTest.scala new file mode 100644 index 0000000000000..11d6da4b48b3c --- /dev/null +++ b/core/src/test/scala/unit/kafka/integration/SaslPlaintextTopicMetadataTest.scala @@ -0,0 +1,26 @@ +/** + * 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.integration + +import kafka.api.SaslTestHarness +import org.apache.kafka.common.protocol.SecurityProtocol + +class SaslPlaintextTopicMetadataTest extends BaseTopicMetadataTest with SaslTestHarness { + protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT + protected def trustStoreFile = None +} diff --git a/core/src/test/scala/unit/kafka/integration/SaslSslTopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/SaslSslTopicMetadataTest.scala new file mode 100644 index 0000000000000..9174637d26e4c --- /dev/null +++ b/core/src/test/scala/unit/kafka/integration/SaslSslTopicMetadataTest.scala @@ -0,0 +1,28 @@ +/** + * 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.integration + +import java.io.File + +import kafka.api.SaslTestHarness +import org.apache.kafka.common.protocol.SecurityProtocol + +class SaslSslTopicMetadataTest extends BaseTopicMetadataTest with SaslTestHarness { + protected def securityProtocol = SecurityProtocol.SASL_SSL + protected def trustStoreFile = Some(File.createTempFile("truststore", ".jks")) +} diff --git a/core/src/test/scala/unit/kafka/integration/SslTopicMetadataTest.scala b/core/src/test/scala/unit/kafka/integration/SslTopicMetadataTest.scala index 5ff9f35b27831..f010d45bdb31e 100644 --- a/core/src/test/scala/unit/kafka/integration/SslTopicMetadataTest.scala +++ b/core/src/test/scala/unit/kafka/integration/SslTopicMetadataTest.scala @@ -19,6 +19,9 @@ package kafka.integration import java.io.File +import org.apache.kafka.common.protocol.SecurityProtocol + class SslTopicMetadataTest extends BaseTopicMetadataTest { + protected def securityProtocol = SecurityProtocol.SSL protected def trustStoreFile = Some(File.createTempFile("truststore", ".jks")) } diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala index 6f07a7a2ef93c..b0cb97ee7c608 100644 --- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala +++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala @@ -221,7 +221,8 @@ class SocketServerTest extends JUnitSuite { @Test def testSSLSocketServer(): Unit = { val trustStoreFile = File.createTempFile("truststore", ".jks") - val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0, enableSSL = true, trustStoreFile = Some(trustStoreFile)) + val overrideProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, enableSsl = true, + trustStoreFile = Some(trustStoreFile)) overrideProps.put("listeners", "SSL://localhost:0") val serverMetrics = new Metrics diff --git a/core/src/test/scala/unit/kafka/server/BaseReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/BaseReplicaFetchTest.scala index b744b94cf4e5a..41698d8fcf7da 100644 --- a/core/src/test/scala/unit/kafka/server/BaseReplicaFetchTest.scala +++ b/core/src/test/scala/unit/kafka/server/BaseReplicaFetchTest.scala @@ -19,6 +19,7 @@ package kafka.server import java.io.File +import org.apache.kafka.common.protocol.SecurityProtocol import org.junit.{Test, After, Before} import kafka.zk.ZooKeeperTestHarness import kafka.utils.TestUtils._ @@ -32,15 +33,16 @@ abstract class BaseReplicaFetchTest extends ZooKeeperTestHarness { val topic1 = "foo" val topic2 = "bar" - /* If this is `Some`, SSL will be enabled */ + // This should be defined if `securityProtocol` uses SSL (eg SSL, SASL_SSL) protected def trustStoreFile: Option[File] + protected def securityProtocol: SecurityProtocol @Before override def setUp() { super.setUp() - brokers = createBrokerConfigs(2, zkConnect, enableControlledShutdown = false, enableSSL = trustStoreFile.isDefined, trustStoreFile = trustStoreFile) - .map(KafkaConfig.fromProps) - .map(TestUtils.createServer(_)) + val props = createBrokerConfigs(2, zkConnect, interBrokerSecurityProtocol = Some(securityProtocol), + trustStoreFile = trustStoreFile) + brokers = props.map(KafkaConfig.fromProps).map(TestUtils.createServer(_)) } @After diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index e664a3226cabe..17d66a47529e0 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -512,6 +512,7 @@ class KafkaConfigTest { case KafkaConfig.SaslKerberosTicketRenewWindowFactorProp => case KafkaConfig.SaslKerberosTicketRenewJitterProp => case KafkaConfig.SaslKerberosMinTimeBeforeReloginProp => + case KafkaConfig.AuthToLocalProp => // ignore string case nonNegativeIntProperty => assertPropertyInvalid(getBaseProperties(), name, "not_a_number", "-1") } diff --git a/core/src/test/scala/unit/kafka/server/PlaintextReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/PlaintextReplicaFetchTest.scala index 871e49b6f8d83..b160481ae4e33 100644 --- a/core/src/test/scala/unit/kafka/server/PlaintextReplicaFetchTest.scala +++ b/core/src/test/scala/unit/kafka/server/PlaintextReplicaFetchTest.scala @@ -17,6 +17,9 @@ package kafka.server +import org.apache.kafka.common.protocol.SecurityProtocol + class PlaintextReplicaFetchTest extends BaseReplicaFetchTest { + protected def securityProtocol = SecurityProtocol.PLAINTEXT protected def trustStoreFile = None } diff --git a/core/src/test/scala/unit/kafka/server/SaslPlaintextReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/SaslPlaintextReplicaFetchTest.scala new file mode 100644 index 0000000000000..740db374b01b7 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/SaslPlaintextReplicaFetchTest.scala @@ -0,0 +1,26 @@ +/** + * 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.server + +import kafka.api.SaslTestHarness +import org.apache.kafka.common.protocol.SecurityProtocol + +class SaslPlaintextReplicaFetchTest extends BaseReplicaFetchTest with SaslTestHarness { + protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT + protected def trustStoreFile = None +} diff --git a/core/src/test/scala/unit/kafka/server/SaslSslReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/SaslSslReplicaFetchTest.scala new file mode 100644 index 0000000000000..7291967821215 --- /dev/null +++ b/core/src/test/scala/unit/kafka/server/SaslSslReplicaFetchTest.scala @@ -0,0 +1,28 @@ +/** + * 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.server + +import java.io.File + +import kafka.api.SaslTestHarness +import org.apache.kafka.common.protocol.SecurityProtocol + +class SaslSslReplicaFetchTest extends BaseReplicaFetchTest with SaslTestHarness { + protected def securityProtocol = SecurityProtocol.SASL_SSL + protected def trustStoreFile = Some(File.createTempFile("truststore", ".jks")) +} diff --git a/core/src/test/scala/unit/kafka/server/SslReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/SslReplicaFetchTest.scala index 98580528fb15e..62ca702878939 100644 --- a/core/src/test/scala/unit/kafka/server/SslReplicaFetchTest.scala +++ b/core/src/test/scala/unit/kafka/server/SslReplicaFetchTest.scala @@ -19,6 +19,9 @@ package kafka.server import java.io.File +import org.apache.kafka.common.protocol.SecurityProtocol + class SslReplicaFetchTest extends BaseReplicaFetchTest { + protected def securityProtocol = SecurityProtocol.SSL protected def trustStoreFile = Some(File.createTempFile("truststore", ".jks")) } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 78c12eb1aa34d..dba1ba79e8278 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -137,17 +137,25 @@ object TestUtils extends Logging { } /** - * Create a test config for the given node id + * Create a test config for the provided parameters. + * + * Note that if `interBrokerSecurityProtocol` is defined, the listener for the `SecurityProtocol` will be enabled. */ def createBrokerConfigs(numConfigs: Int, zkConnect: String, enableControlledShutdown: Boolean = true, enableDeleteTopic: Boolean = false, - enableSSL: Boolean = false, - enableSasl: Boolean = false, - trustStoreFile: Option[File] = None): Seq[Properties] = { - (0 until numConfigs).map(node => createBrokerConfig(node, zkConnect, enableControlledShutdown, enableDeleteTopic, - enableSSL = enableSSL, enableSasl = enableSasl, trustStoreFile = trustStoreFile)) + interBrokerSecurityProtocol: Option[SecurityProtocol] = None, + trustStoreFile: Option[File] = None, + enablePlaintext: Boolean = true, + enableSsl: Boolean = false, + enableSaslPlaintext: Boolean = false, + enableSaslSsl: Boolean = false): Seq[Properties] = { + (0 until numConfigs).map { node => + createBrokerConfig(node, zkConnect, enableControlledShutdown, enableDeleteTopic, RandomPort, + interBrokerSecurityProtocol, trustStoreFile, enablePlaintext = enablePlaintext, enableSsl = enableSsl, + enableSaslPlaintext = enableSaslPlaintext, enableSaslSsl = enableSaslSsl) + } } def getBrokerListStrFromServers(servers: Seq[KafkaServer], protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): String = { @@ -155,20 +163,38 @@ object TestUtils extends Logging { } /** - * Create a test config for the given node id - */ + * Create a test config for the provided parameters. + * + * Note that if `interBrokerSecurityProtocol` is defined, the listener for the `SecurityProtocol` will be enabled. + */ def createBrokerConfig(nodeId: Int, zkConnect: String, enableControlledShutdown: Boolean = true, enableDeleteTopic: Boolean = false, - port: Int = RandomPort, enableSasl: Boolean = false, saslPort: Int = RandomPort, enableSSL: Boolean = false, - sslPort: Int = RandomPort, trustStoreFile: Option[File] = None): Properties = { + port: Int = RandomPort, + interBrokerSecurityProtocol: Option[SecurityProtocol] = None, + trustStoreFile: Option[File] = None, + enablePlaintext: Boolean = true, + enableSaslPlaintext: Boolean = false, saslPlaintextPort: Int = RandomPort, + enableSsl: Boolean = false, sslPort: Int = RandomPort, + enableSaslSsl: Boolean = false, saslSslPort: Int = RandomPort) + : Properties = { + + def shouldEnable(protocol: SecurityProtocol) = interBrokerSecurityProtocol.fold(false)(_ == protocol) + + val sslEnabled = enableSsl || shouldEnable(SecurityProtocol.SSL) + val saslSslEnabled = enableSaslSsl || shouldEnable(SecurityProtocol.SASL_SSL) val listeners = { - val protocolAndPorts = ArrayBuffer("PLAINTEXT" -> port) - if (enableSSL) + + val protocolAndPorts = ArrayBuffer[(String, Int)]() + if (enablePlaintext || shouldEnable(SecurityProtocol.PLAINTEXT)) + protocolAndPorts += ("PLAINTEXT" -> port) + if (sslEnabled) protocolAndPorts += "SSL" -> sslPort - if (enableSasl) - protocolAndPorts += "SASL_PLAINTEXT" -> saslPort + if (enableSaslPlaintext || shouldEnable(SecurityProtocol.SASL_PLAINTEXT)) + protocolAndPorts += "SASL_PLAINTEXT" -> saslPlaintextPort + if (saslSslEnabled) + protocolAndPorts += "SASL_SSL" -> saslSslPort protocolAndPorts.map { case (protocol, port) => s"$protocol://localhost:$port" }.mkString(",") @@ -184,8 +210,14 @@ object TestUtils extends Logging { props.put("controlled.shutdown.enable", enableControlledShutdown.toString) props.put("delete.topic.enable", enableDeleteTopic.toString) props.put("controlled.shutdown.retry.backoff.ms", "100") - if (enableSSL) + + if (sslEnabled || saslSslEnabled) props.putAll(addSSLConfigs(Mode.SERVER, true, trustStoreFile, s"server$nodeId")) + + interBrokerSecurityProtocol.foreach { protocol => + props.put(KafkaConfig.InterBrokerSecurityProtocolProp, protocol.name) + } + props.put("port", port.toString) props } @@ -923,19 +955,18 @@ object TestUtils extends Logging { new String(bytes, encoding) } - def addSSLConfigs(mode: Mode, clientCert: Boolean, trustStoreFile: Option[File], certAlias: String): Properties = { - if (!trustStoreFile.isDefined) { + def addSSLConfigs(mode: Mode, clientCert: Boolean, trustStoreFile: Option[File], certAlias: String): Properties = { + + val trustStore = trustStoreFile.getOrElse { throw new Exception("enableSSL set to true but no trustStoreFile provided") } + val sslConfigs = { - if (mode == Mode.SERVER) { - val sslConfigs = TestSSLUtils.createSSLConfig(true, true, mode, trustStoreFile.get, certAlias) - sslConfigs.put(KafkaConfig.InterBrokerSecurityProtocolProp, SecurityProtocol.SSL.name) - sslConfigs - } + if (mode == Mode.SERVER) + TestSSLUtils.createSSLConfig(true, true, mode, trustStore, certAlias) else - TestSSLUtils.createSSLConfig(clientCert, false, mode, trustStoreFile.get, certAlias) + TestSSLUtils.createSSLConfig(clientCert, false, mode, trustStore, certAlias) } val sslProps = new Properties()