KAFKA-4636; Per listener security settings overrides (KIP-103) Author: Ismael Juma <[email protected]>
Reviewers: Jun Rao <[email protected]>, Rajini Sivaram <[email protected]> Closes #2406 from ijuma/kafka-4636-per-listener-security-settings Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/ca0c071c Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/ca0c071c Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/ca0c071c Branch: refs/heads/trunk Commit: ca0c071c108c9fd31a759e1cd1c4f89bdc5ac47e Parents: 1eac3f3 Author: Ismael Juma <[email protected]> Authored: Fri Jan 27 01:24:39 2017 +0000 Committer: Ismael Juma <[email protected]> Committed: Fri Jan 27 01:24:56 2017 +0000 ---------------------------------------------------------------------- .../org/apache/kafka/clients/ClientUtils.java | 15 +- .../kafka/clients/consumer/KafkaConsumer.java | 2 +- .../kafka/clients/producer/KafkaProducer.java | 2 +- .../kafka/common/config/AbstractConfig.java | 48 ++- .../apache/kafka/common/config/ConfigDef.java | 38 +-- .../kafka/common/network/ChannelBuilders.java | 56 ++-- .../kafka/common/network/ListenerName.java | 6 + .../apache/kafka/common/network/LoginType.java | 39 --- .../org/apache/kafka/common/network/Mode.java | 3 + .../common/network/SaslChannelBuilder.java | 48 ++- .../kafka/common/security/JaasConfig.java | 7 +- .../kafka/common/security/JaasContext.java | 190 ++++++++++++ .../apache/kafka/common/security/JaasUtils.java | 111 ------- .../security/auth/DefaultPrincipalBuilder.java | 1 - .../kafka/common/security/auth/Login.java | 5 +- .../security/authenticator/AbstractLogin.java | 17 +- .../security/authenticator/LoginManager.java | 54 ++-- .../SaslClientCallbackHandler.java | 2 +- .../authenticator/SaslServerAuthenticator.java | 10 +- .../SaslServerCallbackHandler.java | 15 +- .../common/security/kerberos/KerberosLogin.java | 44 +-- .../common/security/plain/PlainSaslServer.java | 29 +- .../scram/ScramServerCallbackHandler.java | 2 +- .../kafka/common/config/AbstractConfigTest.java | 58 ++++ .../kafka/common/network/NetworkTestUtils.java | 9 +- .../kafka/common/network/NioEchoServer.java | 10 +- .../common/network/SslTransportLayerTest.java | 94 ++++-- .../kafka/common/security/JaasContextTest.java | 308 +++++++++++++++++++ .../kafka/common/security/JaasUtilsTest.java | 255 --------------- .../common/security/TestSecurityConfig.java | 38 +++ .../authenticator/SaslAuthenticatorTest.java | 133 +++++--- .../security/authenticator/TestJaasConfig.java | 10 +- .../org/apache/kafka/test/TestSslUtils.java | 2 +- .../runtime/distributed/WorkerGroupMember.java | 2 +- .../main/scala/kafka/admin/AdminClient.scala | 2 +- .../controller/ControllerChannelManager.scala | 8 +- .../main/scala/kafka/network/SocketServer.scala | 8 +- .../main/scala/kafka/server/KafkaServer.scala | 13 +- .../kafka/server/ReplicaFetcherThread.scala | 8 +- .../kafka/api/IntegrationTestHarness.scala | 6 + .../api/SaslEndToEndAuthorizationTest.scala | 7 +- .../api/SaslPlainPlaintextConsumerTest.scala | 7 + .../scala/integration/kafka/api/SaslSetup.scala | 18 +- .../integration/kafka/api/SaslTestHarness.scala | 6 +- .../api/SslEndToEndAuthorizationTest.scala | 2 +- ...eListenersWithSameSecurityProtocolTest.scala | 10 +- .../integration/KafkaServerTestHarness.scala | 2 +- .../unit/kafka/network/SocketServerTest.scala | 2 +- .../scala/unit/kafka/utils/JaasTestUtils.scala | 20 +- .../test/scala/unit/kafka/utils/TestUtils.scala | 2 +- .../processor/internals/StreamsKafkaClient.java | 2 +- 51 files changed, 1087 insertions(+), 699 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java index 3d54515..28eb72e 100644 --- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java +++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java @@ -16,11 +16,11 @@ import java.io.Closeable; import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.concurrent.atomic.AtomicReference; +import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.network.ChannelBuilders; -import org.apache.kafka.common.network.LoginType; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.protocol.SecurityProtocol; import org.apache.kafka.common.network.ChannelBuilder; import org.apache.kafka.common.config.ConfigException; @@ -73,14 +73,15 @@ public class ClientUtils { } /** - * @param configs client/server configs + * @param config client configs * @return configured ChannelBuilder based on the configs. */ - public static ChannelBuilder createChannelBuilder(Map<String, ?> configs) { - SecurityProtocol securityProtocol = SecurityProtocol.forName((String) configs.get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); + public static ChannelBuilder createChannelBuilder(AbstractConfig config) { + SecurityProtocol securityProtocol = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG)); if (!SecurityProtocol.nonTestingValues().contains(securityProtocol)) throw new ConfigException("Invalid SecurityProtocol " + securityProtocol); - String clientSaslMechanism = (String) configs.get(SaslConfigs.SASL_MECHANISM); - return ChannelBuilders.clientChannelBuilder(securityProtocol, LoginType.CLIENT, configs, clientSaslMechanism, true); + String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM); + return ChannelBuilders.clientChannelBuilder(securityProtocol, JaasContext.Type.CLIENT, config, null, + clientSaslMechanism, true); } } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index 2936f0f..e21c196 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -648,7 +648,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> { List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)); this.metadata.update(Cluster.bootstrap(addresses), 0); String metricGrpPrefix = "consumer"; - ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config.values()); + ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config); NetworkClient netClient = new NetworkClient( new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, channelBuilder), this.metadata, http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index c604daa..4415c64 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -298,7 +298,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> { List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)); this.metadata.update(Cluster.bootstrap(addresses), time.milliseconds()); - ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config.values()); + ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config); NetworkClient client = new NetworkClient( new Selector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), this.metrics, time, "producer", channelBuilder), this.metadata, http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java index 096047f..84047e0 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java +++ b/clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java @@ -46,6 +46,8 @@ public class AbstractConfig { /* the parsed values */ private final Map<String, Object> values; + private final ConfigDef definition; + @SuppressWarnings("unchecked") public AbstractConfig(ConfigDef definition, Map<?, ?> originals, boolean doLog) { /* check that all the keys are really strings */ @@ -55,6 +57,7 @@ public class AbstractConfig { this.originals = (Map<String, ?>) originals; this.values = definition.parse(this.originals); this.used = Collections.synchronizedSet(new HashSet<String>()); + this.definition = definition; if (doLog) logAll(); } @@ -63,12 +66,6 @@ public class AbstractConfig { this(definition, originals, true); } - public AbstractConfig(Map<String, Object> parsedConfig) { - this.values = parsedConfig; - this.originals = new HashMap<>(); - this.used = Collections.synchronizedSet(new HashSet<String>()); - } - protected Object get(String key) { if (!values.containsKey(key)) throw new ConfigException(String.format("Unknown configuration '%s'", key)); @@ -152,7 +149,7 @@ public class AbstractConfig { * @return a Map containing the settings with the prefix */ public Map<String, Object> originalsWithPrefix(String prefix) { - Map<String, Object> result = new RecordingMap<>(prefix); + Map<String, Object> result = new RecordingMap<>(prefix, false); for (Map.Entry<String, ?> entry : originals.entrySet()) { if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) result.put(entry.getKey().substring(prefix.length()), entry.getValue()); @@ -160,6 +157,25 @@ public class AbstractConfig { return result; } + /** + * Put all keys that do not start with {@code prefix} and their parsed values in the result map and then + * put all the remaining keys with the prefix stripped and their parsed values in the result map. + * + * This is useful if one wants to allow prefixed configs to override default ones. + */ + public Map<String, Object> valuesWithPrefixOverride(String prefix) { + Map<String, Object> result = new RecordingMap<>(values(), prefix, true); + for (Map.Entry<String, ?> entry : originals.entrySet()) { + if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) { + String keyWithNoPrefix = entry.getKey().substring(prefix.length()); + ConfigDef.ConfigKey configKey = definition.configKeys().get(keyWithNoPrefix); + if (configKey != null) + result.put(keyWithNoPrefix, definition.parseValue(configKey, entry.getValue(), true)); + } + } + return result; + } + public Map<String, ?> values() { return new RecordingMap<>(values); } @@ -264,34 +280,40 @@ public class AbstractConfig { private class RecordingMap<V> extends HashMap<String, V> { private final String prefix; + private final boolean withIgnoreFallback; RecordingMap() { - this(""); + this("", false); } - RecordingMap(String prefix) { + RecordingMap(String prefix, boolean withIgnoreFallback) { this.prefix = prefix; + this.withIgnoreFallback = withIgnoreFallback; } RecordingMap(Map<String, ? extends V> m) { - this(m, ""); + this(m, "", false); } - RecordingMap(Map<String, ? extends V> m, String prefix) { + RecordingMap(Map<String, ? extends V> m, String prefix, boolean withIgnoreFallback) { super(m); this.prefix = prefix; + this.withIgnoreFallback = withIgnoreFallback; } @Override public V get(Object key) { if (key instanceof String) { + String stringKey = (String) key; String keyWithPrefix; if (prefix.isEmpty()) { - keyWithPrefix = (String) key; + keyWithPrefix = stringKey; } else { - keyWithPrefix = prefix + key; + keyWithPrefix = prefix + stringKey; } ignore(keyWithPrefix); + if (withIgnoreFallback) + ignore(stringKey); } return super.get(key); } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java index 89feb9a..1694614 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java +++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java @@ -426,26 +426,28 @@ public class ConfigDef { } // parse all known keys Map<String, Object> values = new HashMap<>(); - for (ConfigKey key : configKeys.values()) { - Object value; - // props map contains setting - assign ConfigKey value - if (props.containsKey(key.name)) { - value = parseType(key.name, props.get(key.name), key.type); - // props map doesn't contain setting, the key is required because no default value specified - its an error - } else if (key.defaultValue == NO_DEFAULT_VALUE) { - throw new ConfigException("Missing required configuration \"" + key.name + "\" which has no default value."); - } else { - // otherwise assign setting its default value - value = key.defaultValue; - } - if (key.validator != null) { - key.validator.ensureValid(key.name, value); - } - values.put(key.name, value); - } + for (ConfigKey key : configKeys.values()) + values.put(key.name, parseValue(key, props.get(key.name), props.containsKey(key.name))); return values; } + Object parseValue(ConfigKey key, Object value, boolean isSet) { + Object parsedValue; + if (isSet) { + parsedValue = parseType(key.name, value, key.type); + // props map doesn't contain setting, the key is required because no default value specified - its an error + } else if (key.defaultValue == NO_DEFAULT_VALUE) { + throw new ConfigException("Missing required configuration \"" + key.name + "\" which has no default value."); + } else { + // otherwise assign setting its default value + parsedValue = key.defaultValue; + } + if (key.validator != null) { + key.validator.ensureValid(key.name, parsedValue); + } + return parsedValue; + } + /** * Validate the current configuration values with the configuration definition. * @param props the current configuration values @@ -1166,4 +1168,4 @@ public class ConfigDef { }; } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java index 5a1486c..02fb5e8 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 @@ -13,8 +13,10 @@ package org.apache.kafka.common.network; +import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.config.SslConfigs; import org.apache.kafka.common.protocol.SecurityProtocol; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.auth.DefaultPrincipalBuilder; import org.apache.kafka.common.security.auth.PrincipalBuilder; import org.apache.kafka.common.security.authenticator.CredentialCache; @@ -28,8 +30,9 @@ public class ChannelBuilders { /** * @param securityProtocol the securityProtocol - * @param loginType the loginType, it must be non-null if `securityProtocol` is SASL_*; it is ignored otherwise - * @param configs client configs + * @param contextType the contextType, it must be non-null if `securityProtocol` is SASL_*; it is ignored otherwise + * @param config client config + * @param listenerName the listenerName if contextType is SERVER or null otherwise * @param clientSaslMechanism SASL mechanism if mode is CLIENT, ignored otherwise * @param saslHandshakeRequestEnable flag to enable Sasl handshake requests; disabled only for SASL * inter-broker connections with inter-broker protocol version < 0.10 @@ -37,41 +40,52 @@ public class ChannelBuilders { * @throws IllegalArgumentException if `mode` invariants described above is not maintained */ public static ChannelBuilder clientChannelBuilder(SecurityProtocol securityProtocol, - LoginType loginType, - Map<String, ?> configs, + JaasContext.Type contextType, + AbstractConfig config, + ListenerName listenerName, String clientSaslMechanism, boolean saslHandshakeRequestEnable) { if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL) { - if (loginType == null) - throw new IllegalArgumentException("`loginType` must be non-null if `securityProtocol` is `" + securityProtocol + "`"); + if (contextType == null) + throw new IllegalArgumentException("`contextType` must be non-null if `securityProtocol` is `" + securityProtocol + "`"); if (clientSaslMechanism == null) throw new IllegalArgumentException("`clientSaslMechanism` must be non-null in client mode if `securityProtocol` is `" + securityProtocol + "`"); } - return create(securityProtocol, Mode.CLIENT, loginType, configs, clientSaslMechanism, saslHandshakeRequestEnable, null); + return create(securityProtocol, Mode.CLIENT, contextType, config, listenerName, clientSaslMechanism, + saslHandshakeRequestEnable, null); } /** + * @param listenerName the listenerName * @param securityProtocol the securityProtocol - * @param configs server configs + * @param config server config * @param credentialCache Credential cache for SASL/SCRAM if SCRAM is enabled * @return the configured `ChannelBuilder` */ - public static ChannelBuilder serverChannelBuilder(SecurityProtocol securityProtocol, - Map<String, ?> configs, - CredentialCache credentialCache) { - return create(securityProtocol, Mode.SERVER, LoginType.SERVER, configs, null, true, credentialCache); + public static ChannelBuilder serverChannelBuilder(ListenerName listenerName, + SecurityProtocol securityProtocol, + AbstractConfig config, + CredentialCache credentialCache) { + return create(securityProtocol, Mode.SERVER, JaasContext.Type.SERVER, config, listenerName, null, + true, credentialCache); } private static ChannelBuilder create(SecurityProtocol securityProtocol, - Mode mode, - LoginType loginType, - Map<String, ?> configs, - String clientSaslMechanism, - boolean saslHandshakeRequestEnable, - CredentialCache credentialCache) { - ChannelBuilder channelBuilder; + Mode mode, + JaasContext.Type contextType, + AbstractConfig config, + ListenerName listenerName, + String clientSaslMechanism, + boolean saslHandshakeRequestEnable, + CredentialCache credentialCache) { + Map<String, ?> configs; + if (listenerName == null) + configs = config.values(); + else + configs = config.valuesWithPrefixOverride(listenerName.configPrefix()); + ChannelBuilder channelBuilder; switch (securityProtocol) { case SSL: requireNonNullMode(mode, securityProtocol); @@ -80,7 +94,9 @@ public class ChannelBuilders { case SASL_SSL: case SASL_PLAINTEXT: requireNonNullMode(mode, securityProtocol); - channelBuilder = new SaslChannelBuilder(mode, loginType, securityProtocol, clientSaslMechanism, saslHandshakeRequestEnable, credentialCache); + JaasContext jaasContext = JaasContext.load(contextType, listenerName, configs); + channelBuilder = new SaslChannelBuilder(mode, jaasContext, securityProtocol, + clientSaslMechanism, saslHandshakeRequestEnable, credentialCache); break; case PLAINTEXT: case TRACE: http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java b/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java index b376514..fad2ee2 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java +++ b/clients/src/main/java/org/apache/kafka/common/network/ListenerName.java @@ -23,6 +23,8 @@ import java.util.Objects; public final class ListenerName { + private static final String CONFIG_STATIC_PREFIX = "listener.name"; + /** * Create an instance with the security protocol name as the value. */ @@ -65,4 +67,8 @@ public final class ListenerName { public String toString() { return "ListenerName(" + value + ")"; } + + public String configPrefix() { + return CONFIG_STATIC_PREFIX + "." + value.toLowerCase(Locale.ROOT) + "."; + } } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/network/LoginType.java ---------------------------------------------------------------------- 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 deleted file mode 100644 index a3a2b27..0000000 --- a/clients/src/main/java/org/apache/kafka/common/network/LoginType.java +++ /dev/null @@ -1,39 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.kafka.common.network; - -import 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 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; - } -} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/network/Mode.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/network/Mode.java b/clients/src/main/java/org/apache/kafka/common/network/Mode.java index 4d8ef3b..59ef712 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/Mode.java +++ b/clients/src/main/java/org/apache/kafka/common/network/Mode.java @@ -16,4 +16,7 @@ */ package org.apache.kafka.common.network; +/** + * Connection mode for SSL and SASL connections. + */ public enum Mode { CLIENT, SERVER } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java index b556f38..060f833 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 @@ -13,15 +13,15 @@ package org.apache.kafka.common.network; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; import java.util.List; import java.util.Map; -import javax.security.auth.login.Configuration; - import org.apache.kafka.common.config.SaslConfigs; -import org.apache.kafka.common.security.JaasUtils; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.kerberos.KerberosShortNamer; import org.apache.kafka.common.security.authenticator.CredentialCache; import org.apache.kafka.common.security.authenticator.LoginManager; @@ -39,20 +39,20 @@ public class SaslChannelBuilder implements ChannelBuilder { private final SecurityProtocol securityProtocol; private final String clientSaslMechanism; private final Mode mode; - private final LoginType loginType; + private final JaasContext jaasContext; private final boolean handshakeRequestEnable; private final CredentialCache credentialCache; - private Configuration jaasConfig; private LoginManager loginManager; private SslFactory sslFactory; private Map<String, ?> configs; private KerberosShortNamer kerberosShortNamer; - public SaslChannelBuilder(Mode mode, LoginType loginType, SecurityProtocol securityProtocol, - String clientSaslMechanism, boolean handshakeRequestEnable, CredentialCache credentialCache) { + public SaslChannelBuilder(Mode mode, JaasContext jaasContext, SecurityProtocol securityProtocol, + String clientSaslMechanism, + boolean handshakeRequestEnable, CredentialCache credentialCache) { this.mode = mode; - this.loginType = loginType; + this.jaasContext = jaasContext; this.securityProtocol = securityProtocol; this.handshakeRequestEnable = handshakeRequestEnable; this.clientSaslMechanism = clientSaslMechanism; @@ -73,7 +73,7 @@ public class SaslChannelBuilder implements ChannelBuilder { if (hasKerberos) { String defaultRealm; try { - defaultRealm = JaasUtils.defaultKerberosRealm(); + defaultRealm = defaultKerberosRealm(); } catch (Exception ke) { defaultRealm = ""; } @@ -82,8 +82,7 @@ public class SaslChannelBuilder implements ChannelBuilder { if (principalToLocalRules != null) kerberosShortNamer = KerberosShortNamer.fromUnparsedRules(defaultRealm, principalToLocalRules); } - this.jaasConfig = JaasUtils.jaasConfig(loginType, configs); - this.loginManager = LoginManager.acquireLoginManager(loginType, hasKerberos, configs, jaasConfig); + this.loginManager = LoginManager.acquireLoginManager(jaasContext, hasKerberos, configs); if (this.securityProtocol == SecurityProtocol.SASL_SSL) { // Disable SSL client authentication as we are using SASL authentication @@ -101,8 +100,9 @@ public class SaslChannelBuilder implements ChannelBuilder { TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel); Authenticator authenticator; if (mode == Mode.SERVER) - authenticator = new SaslServerAuthenticator(id, jaasConfig, loginManager.subject(), kerberosShortNamer, - socketChannel.socket().getLocalAddress().getHostName(), maxReceiveSize, credentialCache); + authenticator = new SaslServerAuthenticator(id, jaasContext, loginManager.subject(), + kerberosShortNamer, socketChannel.socket().getLocalAddress().getHostName(), maxReceiveSize, + credentialCache); else authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(), socketChannel.socket().getInetAddress().getHostName(), clientSaslMechanism, handshakeRequestEnable); @@ -128,4 +128,26 @@ public class SaslChannelBuilder implements ChannelBuilder { return new PlaintextTransportLayer(key); } } + + private static String defaultKerberosRealm() 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; + Method getDefaultRealmMethod; + if (System.getProperty("java.vendor").contains("IBM")) { + classRef = Class.forName("com.ibm.security.krb5.internal.Config"); + } else { + classRef = Class.forName("sun.security.krb5.Config"); + } + getInstanceMethod = classRef.getMethod("getInstance", new Class[0]); + kerbConf = getInstanceMethod.invoke(classRef, new Object[0]); + getDefaultRealmMethod = classRef.getDeclaredMethod("getDefaultRealm", + new Class[0]); + return (String) getDefaultRealmMethod.invoke(kerbConf, new Object[0]); + } } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/JaasConfig.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasConfig.java b/clients/src/main/java/org/apache/kafka/common/security/JaasConfig.java index 2128c61..d900be2 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/JaasConfig.java +++ b/clients/src/main/java/org/apache/kafka/common/security/JaasConfig.java @@ -32,7 +32,6 @@ import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.config.SaslConfigs; -import org.apache.kafka.common.network.LoginType; /** * JAAS configuration parser that constructs a JAAS configuration object with a single @@ -51,7 +50,7 @@ class JaasConfig extends Configuration { private final String loginContextName; private final List<AppConfigurationEntry> configEntries; - public JaasConfig(LoginType loginType, String jaasConfigParams) { + public JaasConfig(String loginContextName, String jaasConfigParams) { StreamTokenizer tokenizer = new StreamTokenizer(new StringReader(jaasConfigParams)); tokenizer.slashSlashComments(true); tokenizer.slashStarComments(true); @@ -67,7 +66,7 @@ class JaasConfig extends Configuration { if (configEntries.isEmpty()) throw new IllegalArgumentException("Login module not specified in JAAS config"); - this.loginContextName = loginType.contextName(); + this.loginContextName = loginContextName; } catch (IOException e) { throw new KafkaException("Unexpected exception while parsing JAAS config"); @@ -120,4 +119,4 @@ class JaasConfig extends Configuration { throw new IllegalArgumentException("JAAS config entry not terminated by semi-colon"); return new AppConfigurationEntry(loginModule, controlFlag, options); } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java b/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java new file mode 100644 index 0000000..6abeef6 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/security/JaasContext.java @@ -0,0 +1,190 @@ +/** + * 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; + +import org.apache.kafka.common.config.SaslConfigs; +import org.apache.kafka.common.config.types.Password; +import org.apache.kafka.common.network.ListenerName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.security.auth.login.AppConfigurationEntry; +import javax.security.auth.login.Configuration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +public class JaasContext { + + private static final Logger LOG = LoggerFactory.getLogger(JaasUtils.class); + + private static final String GLOBAL_CONTEXT_NAME_SERVER = "KafkaServer"; + private static final String GLOBAL_CONTEXT_NAME_CLIENT = "KafkaClient"; + + /** + * Returns an instance of this class. + * + * For contextType SERVER, the context will contain the default Configuration and the context name will be one of: + * + * 1. Lowercased listener name followed by a period and the string `KafkaServer` + * 2. The string `KafkaServer` + * + * If both are valid entries in the JAAS configuration, the first option is chosen. + * + * For contextType CLIENT, if JAAS configuration property @link SaslConfigs#SASL_JAAS_CONFIG} is specified, + * the configuration object is created by parsing the property value. Otherwise, the default Configuration + * is returned. The context name is always `KafkaClient`. + * + * @throws IllegalArgumentException if JAAS configuration property is specified for contextType SERVER, if + * listenerName is not defined for contextType SERVER of if listenerName is defined for contextType CLIENT. + */ + public static JaasContext load(JaasContext.Type contextType, ListenerName listenerName, + Map<String, ?> configs) { + String listenerContextName; + String globalContextName; + switch (contextType) { + case CLIENT: + if (listenerName != null) + throw new IllegalArgumentException("listenerName should be null for CLIENT"); + globalContextName = GLOBAL_CONTEXT_NAME_CLIENT; + listenerContextName = null; + break; + case SERVER: + if (listenerName == null) + throw new IllegalArgumentException("listenerName should not be null for SERVER"); + globalContextName = GLOBAL_CONTEXT_NAME_SERVER; + listenerContextName = listenerName.value().toLowerCase(Locale.ROOT) + "." + GLOBAL_CONTEXT_NAME_SERVER; + break; + default: + throw new IllegalArgumentException("Unexpected context type " + contextType); + } + return load(contextType, listenerContextName, globalContextName, configs); + } + + static JaasContext load(JaasContext.Type contextType, String listenerContextName, + String globalContextName, Map<String, ?> configs) { + Password jaasConfigArgs = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG); + if (jaasConfigArgs != null) { + if (contextType == JaasContext.Type.SERVER) + throw new IllegalArgumentException("JAAS config property not supported for server"); + else { + JaasConfig jaasConfig = new JaasConfig(globalContextName, jaasConfigArgs.value()); + AppConfigurationEntry[] clientModules = jaasConfig.getAppConfigurationEntry(globalContextName); + int numModules = clientModules == null ? 0 : clientModules.length; + if (numModules != 1) + throw new IllegalArgumentException("JAAS config property contains " + numModules + " login modules, should be 1 module"); + return new JaasContext(globalContextName, contextType, jaasConfig); + } + } else + return defaultContext(contextType, listenerContextName, globalContextName); + } + + private static JaasContext defaultContext(JaasContext.Type contextType, String listenerContextName, + String globalContextName) { + String jaasConfigFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); + if (jaasConfigFile == null) { + if (contextType == Type.CLIENT) { + LOG.debug("System property '" + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' and Kafka SASL property '" + + SaslConfigs.SASL_JAAS_CONFIG + "' are not set, using default JAAS configuration."); + } else { + LOG.debug("System property '" + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' is not set, using default JAAS " + + "configuration."); + } + } + + Configuration jaasConfig = Configuration.getConfiguration(); + + AppConfigurationEntry[] configEntries = null; + String contextName = globalContextName; + + if (listenerContextName != null) { + configEntries = jaasConfig.getAppConfigurationEntry(listenerContextName); + if (configEntries != null) + contextName = listenerContextName; + } + + if (configEntries == null) + configEntries = jaasConfig.getAppConfigurationEntry(globalContextName); + + if (configEntries == null) { + String listenerNameText = listenerContextName == null ? "" : " or '" + listenerContextName + "'"; + String errorMessage = "Could not find a '" + globalContextName + "'" + listenerNameText + " entry in the JAAS " + + "configuration. System property '" + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' is " + + (jaasConfigFile == null ? "not set" : jaasConfigFile); + throw new IllegalArgumentException(errorMessage); + } + + return new JaasContext(contextName, contextType, jaasConfig); + } + + /** + * The type of the SASL login context, it should be SERVER for the broker and CLIENT for the clients (consumer, producer, + * etc.). This is used to validate behaviour (e.g. some functionality is only available in the broker or clients). + */ + public enum Type { CLIENT, SERVER; } + + private final String name; + private final Type type; + private final Configuration configuration; + private final List<AppConfigurationEntry> configurationEntries; + + public JaasContext(String name, Type type, Configuration configuration) { + this.name = name; + this.type = type; + this.configuration = configuration; + AppConfigurationEntry[] entries = configuration.getAppConfigurationEntry(name); + if (entries == null) + throw new IllegalArgumentException("Could not find a '" + name + "' entry in this JAAS configuration."); + this.configurationEntries = Collections.unmodifiableList(new ArrayList<>(Arrays.asList(entries))); + } + + public String name() { + return name; + } + + public Type type() { + return type; + } + + public Configuration configuration() { + return configuration; + } + + public List<AppConfigurationEntry> configurationEntries() { + return configurationEntries; + } + + /** + * Returns the configuration option for <code>key</code> from this context. + * If login module name is specified, return option value only from that module. + */ + public String configEntryOption(String key, String loginModuleName) { + for (AppConfigurationEntry entry : configurationEntries) { + if (loginModuleName != null && !loginModuleName.equals(entry.getLoginModuleName())) + continue; + Object val = entry.getOptions().get(key); + if (val != null) + return (String) val; + } + return null; + } + +} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java b/clients/src/main/java/org/apache/kafka/common/security/JaasUtils.java index e326156..ca6b7f0 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 @@ -17,16 +17,8 @@ package org.apache.kafka.common.security; import javax.security.auth.login.Configuration; -import javax.security.auth.login.AppConfigurationEntry; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.util.Map; -import java.io.IOException; import org.apache.kafka.common.KafkaException; -import org.apache.kafka.common.config.SaslConfigs; -import org.apache.kafka.common.config.types.Password; -import org.apache.kafka.common.network.LoginType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -34,114 +26,11 @@ public class JaasUtils { private static final Logger LOG = LoggerFactory.getLogger(JaasUtils.class); public static final String JAVA_LOGIN_CONFIG_PARAM = "java.security.auth.login.config"; - public static final String LOGIN_CONTEXT_SERVER = "KafkaServer"; - public static final String LOGIN_CONTEXT_CLIENT = "KafkaClient"; public static final String SERVICE_NAME = "serviceName"; public static final String ZK_SASL_CLIENT = "zookeeper.sasl.client"; public static final String ZK_LOGIN_CONTEXT_NAME_KEY = "zookeeper.sasl.clientconfig"; - /** - * Returns a JAAS Configuration object. For loginType SERVER, default Configuration - * is returned. For loginType CLIENT, if JAAS configuration property - * {@link SaslConfigs#SASL_JAAS_CONFIG} is specified, the configuration object - * is created by parsing the property value. Otherwise, the default Configuration - * is returned. - * @throws IllegalArgumentException if JAAS configuration property is specified - * for loginType SERVER - */ - public static Configuration jaasConfig(LoginType loginType, Map<String, ?> configs) { - Password jaasConfigArgs = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG); - if (jaasConfigArgs != null) { - if (loginType == LoginType.SERVER) - throw new IllegalArgumentException("JAAS config property not supported for server"); - else { - JaasConfig jaasConfig = new JaasConfig(loginType, jaasConfigArgs.value()); - AppConfigurationEntry[] clientModules = jaasConfig.getAppConfigurationEntry(LoginType.CLIENT.contextName()); - int numModules = clientModules == null ? 0 : clientModules.length; - if (numModules != 1) - throw new IllegalArgumentException("JAAS config property contains " + numModules + " login modules, should be one module"); - return jaasConfig; - } - } else - return defaultJaasConfig(loginType); - } - - private static Configuration defaultJaasConfig(LoginType loginType) { - String jaasConfigFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); - if (jaasConfigFile == null) { - LOG.debug("System property '" + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' and Kafka SASL property '" + - SaslConfigs.SASL_JAAS_CONFIG + "' are not set, using default JAAS configuration."); - } - - Configuration jaasConfig = Configuration.getConfiguration(); - - String loginContextName = loginType.contextName(); - AppConfigurationEntry[] configEntries = jaasConfig.getAppConfigurationEntry(loginContextName); - if (configEntries == null) { - String errorMessage; - errorMessage = "Could not find a '" + loginContextName + "' entry in the JAAS configuration. System property '" + - JaasUtils.JAVA_LOGIN_CONFIG_PARAM + "' is " + (jaasConfigFile == null ? "not set" : jaasConfigFile); - throw new IllegalArgumentException(errorMessage); - } - return jaasConfig; - } - - /** - * Returns the configuration option for <code>key</code> from the server login context - * of the default JAAS configuration. If login module name is specified, return option value - * only from that module. - */ - public static String defaultServerJaasConfigOption(String key, String loginModuleName) throws IOException { - return jaasConfigOption(Configuration.getConfiguration(), LoginType.SERVER.contextName(), key, loginModuleName); - } - - /** - * Returns the configuration option for <code>key</code> from the login context - * <code>loginContextName</code> of the specified JAAS configuration. - * If login module name is specified, return option value only from that module. - */ - public static String jaasConfigOption(Configuration jaasConfig, String loginContextName, String key, String loginModuleName) throws IOException { - AppConfigurationEntry[] configurationEntries = jaasConfig.getAppConfigurationEntry(loginContextName); - if (configurationEntries == null) { - String errorMessage = "Could not find a '" + loginContextName + "' entry in this JAAS configuration."; - throw new IOException(errorMessage); - } - - for (AppConfigurationEntry entry: configurationEntries) { - if (loginModuleName != null && !loginModuleName.equals(entry.getLoginModuleName())) - continue; - Object val = entry.getOptions().get(key); - if (val != null) - return (String) val; - } - return null; - } - - public static String defaultKerberosRealm() - 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; - Method getDefaultRealmMethod; - if (System.getProperty("java.vendor").contains("IBM")) { - classRef = Class.forName("com.ibm.security.krb5.internal.Config"); - } else { - classRef = Class.forName("sun.security.krb5.Config"); - } - getInstanceMethod = classRef.getMethod("getInstance", new Class[0]); - kerbConf = getInstanceMethod.invoke(classRef, new Object[0]); - getDefaultRealmMethod = classRef.getDeclaredMethod("getDefaultRealm", - new Class[0]); - return (String) getDefaultRealmMethod.invoke(kerbConf, new Object[0]); - } - public static boolean isZkSecurityEnabled() { boolean zkSaslEnabled = Boolean.parseBoolean(System.getProperty(ZK_SASL_CLIENT, "true")); String zkLoginContextName = System.getProperty(ZK_LOGIN_CONTEXT_NAME_KEY, "Client"); http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java b/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java index fbbeb9e..0ea935c 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/security/auth/DefaultPrincipalBuilder.java @@ -25,7 +25,6 @@ import org.apache.kafka.common.network.Authenticator; import org.apache.kafka.common.KafkaException; /** DefaultPrincipalBuilder which return transportLayer's peer Principal **/ - public class DefaultPrincipalBuilder implements PrincipalBuilder { public void configure(Map<String, ?> configs) {} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/auth/Login.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/Login.java b/clients/src/main/java/org/apache/kafka/common/security/auth/Login.java index 2f831c0..bc55094 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/auth/Login.java +++ b/clients/src/main/java/org/apache/kafka/common/security/auth/Login.java @@ -18,10 +18,11 @@ package org.apache.kafka.common.security.auth; +import org.apache.kafka.common.security.JaasContext; + import java.util.Map; import javax.security.auth.Subject; -import javax.security.auth.login.Configuration; import javax.security.auth.login.LoginContext; import javax.security.auth.login.LoginException; @@ -33,7 +34,7 @@ public interface Login { /** * Configures this login instance. */ - void configure(Map<String, ?> configs, Configuration jaasConfig, String loginContextName); + void configure(Map<String, ?> configs, JaasContext jaasContext); /** * Performs login for each login module specified for the login context of this instance. http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/authenticator/AbstractLogin.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/AbstractLogin.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/AbstractLogin.java index e1bbbce..1b9953c 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/AbstractLogin.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/AbstractLogin.java @@ -18,7 +18,6 @@ package org.apache.kafka.common.security.authenticator; -import javax.security.auth.login.Configuration; import javax.security.auth.login.LoginContext; import javax.security.auth.login.LoginException; import javax.security.sasl.RealmCallback; @@ -29,6 +28,7 @@ import javax.security.auth.callback.PasswordCallback; import javax.security.auth.callback.UnsupportedCallbackException; import javax.security.auth.Subject; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.auth.Login; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,20 +41,17 @@ import java.util.Map; public abstract class AbstractLogin implements Login { private static final Logger log = LoggerFactory.getLogger(AbstractLogin.class); - private Configuration jaasConfig; - private String loginContextName; + private JaasContext jaasContext; private LoginContext loginContext; - @Override - public void configure(Map<String, ?> configs, Configuration jaasConfig, String loginContextName) { - this.jaasConfig = jaasConfig; - this.loginContextName = loginContextName; + public void configure(Map<String, ?> configs, JaasContext jaasContext) { + this.jaasContext = jaasContext; } @Override public LoginContext login() throws LoginException { - loginContext = new LoginContext(loginContextName, null, new LoginCallbackHandler(), jaasConfig); + loginContext = new LoginContext(jaasContext.name(), null, new LoginCallbackHandler(), jaasContext.configuration()); loginContext.login(); log.info("Successfully logged in."); return loginContext; @@ -65,8 +62,8 @@ public abstract class AbstractLogin implements Login { return loginContext.getSubject(); } - protected Configuration jaasConfig() { - return jaasConfig; + protected JaasContext jaasContext() { + return jaasContext; } /** http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java index a28afae..55b561c 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/LoginManager.java @@ -19,39 +19,36 @@ package org.apache.kafka.common.security.authenticator; import javax.security.auth.Subject; -import javax.security.auth.login.Configuration; import javax.security.auth.login.LoginException; import java.io.IOException; import java.util.ArrayList; -import java.util.EnumMap; import java.util.HashMap; import java.util.Map; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.config.types.Password; -import org.apache.kafka.common.network.LoginType; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.auth.Login; import org.apache.kafka.common.security.kerberos.KerberosLogin; public class LoginManager { // static configs (broker or client) - private static final EnumMap<LoginType, LoginManager> LOGIN_TYPE_INSTANCES = new EnumMap<>(LoginType.class); + private static final Map<String, LoginManager> STATIC_INSTANCES = new HashMap<>(); // dynamic configs (client-only) - private static final Map<Password, LoginManager> JAAS_CONF_INSTANCES = new HashMap<>(); + private static final Map<Password, LoginManager> DYNAMIC_INSTANCES = new HashMap<>(); private final Login login; private final Object cacheKey; private int refCount; - private LoginManager(LoginType loginType, boolean hasKerberos, Map<String, ?> configs, Configuration jaasConfig, + private LoginManager(JaasContext jaasContext, boolean hasKerberos, Map<String, ?> configs, Password jaasConfigValue) throws IOException, LoginException { - this.cacheKey = jaasConfigValue != null ? jaasConfigValue : loginType; - String loginContext = loginType.contextName(); + this.cacheKey = jaasConfigValue != null ? jaasConfigValue : jaasContext.name(); login = hasKerberos ? new KerberosLogin() : new DefaultLogin(); - login.configure(configs, jaasConfig, loginContext); + login.configure(configs, jaasContext); login.login(); } @@ -59,35 +56,30 @@ public class LoginManager { * Returns an instance of `LoginManager` and increases its reference count. * * `release()` should be invoked when the `LoginManager` is no longer needed. This method will try to reuse an - * existing `LoginManager` for the provided `loginType` and `SaslConfigs.SASL_JAAS_CONFIG` in `configs`, if available. + * existing `LoginManager` for the provided context type and `SaslConfigs.SASL_JAAS_CONFIG` in `configs`, + * if available. * * 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 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 hasKerberos - * @param configs configuration as key/value pairs - * @param jaasConfig JAAS Configuration object */ - public static LoginManager acquireLoginManager(LoginType loginType, boolean hasKerberos, Map<String, ?> configs, - Configuration jaasConfig) throws IOException, LoginException { + public static LoginManager acquireLoginManager(JaasContext jaasContext, boolean hasKerberos, + Map<String, ?> configs) throws IOException, LoginException { synchronized (LoginManager.class) { // SASL_JAAS_CONFIG is only supported by clients LoginManager loginManager; Password jaasConfigValue = (Password) configs.get(SaslConfigs.SASL_JAAS_CONFIG); - if (loginType == LoginType.CLIENT && jaasConfigValue != null) { - loginManager = JAAS_CONF_INSTANCES.get(jaasConfigValue); + if (jaasContext.type() == JaasContext.Type.CLIENT && jaasConfigValue != null) { + loginManager = DYNAMIC_INSTANCES.get(jaasConfigValue); if (loginManager == null) { - loginManager = new LoginManager(loginType, hasKerberos, configs, jaasConfig, jaasConfigValue); - JAAS_CONF_INSTANCES.put(jaasConfigValue, loginManager); + loginManager = new LoginManager(jaasContext, hasKerberos, configs, jaasConfigValue); + DYNAMIC_INSTANCES.put(jaasConfigValue, loginManager); } } else { - loginManager = LOGIN_TYPE_INSTANCES.get(loginType); + loginManager = STATIC_INSTANCES.get(jaasContext.name()); if (loginManager == null) { - loginManager = new LoginManager(loginType, hasKerberos, configs, jaasConfig, jaasConfigValue); - LOGIN_TYPE_INSTANCES.put(loginType, loginManager); + loginManager = new LoginManager(jaasContext, hasKerberos, configs, jaasConfigValue); + STATIC_INSTANCES.put(jaasContext.name(), loginManager); } } return loginManager.acquire(); @@ -116,9 +108,9 @@ public class LoginManager { throw new IllegalStateException("release called on LoginManager with refCount == 0"); else if (refCount == 1) { if (cacheKey instanceof Password) { - JAAS_CONF_INSTANCES.remove(cacheKey); + DYNAMIC_INSTANCES.remove(cacheKey); } else { - LOGIN_TYPE_INSTANCES.remove(cacheKey); + STATIC_INSTANCES.remove(cacheKey); } login.close(); } @@ -129,10 +121,10 @@ public class LoginManager { /* Should only be used in tests. */ public static void closeAll() { synchronized (LoginManager.class) { - for (LoginType key : new ArrayList<>(LOGIN_TYPE_INSTANCES.keySet())) - LOGIN_TYPE_INSTANCES.remove(key).login.close(); - for (Password key : new ArrayList<>(JAAS_CONF_INSTANCES.keySet())) - JAAS_CONF_INSTANCES.remove(key).login.close(); + for (String key : new ArrayList<>(STATIC_INSTANCES.keySet())) + STATIC_INSTANCES.remove(key).login.close(); + for (Password key : new ArrayList<>(DYNAMIC_INSTANCES.keySet())) + DYNAMIC_INSTANCES.remove(key).login.close(); } } } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java index 8e0b8db..3a38e24 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java @@ -91,4 +91,4 @@ public class SaslClientCallbackHandler implements AuthCallbackHandler { @Override public void close() { } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java index 069e12f..07792d2 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 @@ -24,6 +24,7 @@ import org.apache.kafka.common.errors.IllegalSaslStateException; import org.apache.kafka.common.errors.UnsupportedSaslMechanismException; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.network.Authenticator; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.network.Mode; import org.apache.kafka.common.network.NetworkReceive; import org.apache.kafka.common.network.NetworkSend; @@ -57,7 +58,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.security.auth.Subject; -import javax.security.auth.login.Configuration; import javax.security.sasl.Sasl; import javax.security.sasl.SaslException; import javax.security.sasl.SaslServer; @@ -81,7 +81,7 @@ public class SaslServerAuthenticator implements Authenticator { } private final String node; - private final Configuration jaasConfig; + private final JaasContext jaasContext; private final Subject subject; private final KerberosShortNamer kerberosNamer; private final int maxReceiveSize; @@ -105,11 +105,11 @@ public class SaslServerAuthenticator implements Authenticator { private NetworkReceive netInBuffer; private Send netOutBuffer; - public SaslServerAuthenticator(String node, Configuration jaasConfig, final Subject subject, KerberosShortNamer kerberosNameParser, String host, int maxReceiveSize, CredentialCache credentialCache) throws IOException { + public SaslServerAuthenticator(String node, JaasContext jaasContext, final Subject subject, KerberosShortNamer kerberosNameParser, String host, int maxReceiveSize, CredentialCache credentialCache) throws IOException { if (subject == null) throw new IllegalArgumentException("subject cannot be null"); this.node = node; - this.jaasConfig = jaasConfig; + this.jaasContext = jaasContext; this.subject = subject; this.kerberosNamer = kerberosNameParser; this.maxReceiveSize = maxReceiveSize; @@ -129,7 +129,7 @@ public class SaslServerAuthenticator implements Authenticator { private void createSaslServer(String mechanism) throws IOException { this.saslMechanism = mechanism; if (!ScramMechanism.isScram(mechanism)) - callbackHandler = new SaslServerCallbackHandler(jaasConfig, kerberosNamer); + callbackHandler = new SaslServerCallbackHandler(jaasContext, kerberosNamer); else callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism, ScramCredential.class)); callbackHandler.configure(configs, Mode.SERVER, subject, saslMechanism); http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java index c01f01d..94083fb 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerCallbackHandler.java @@ -21,6 +21,7 @@ package org.apache.kafka.common.security.authenticator; import java.io.IOException; import java.util.Map; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.auth.AuthCallbackHandler; import org.apache.kafka.common.security.kerberos.KerberosShortNamer; import org.slf4j.Logger; @@ -29,14 +30,11 @@ import org.slf4j.LoggerFactory; import javax.security.auth.Subject; import javax.security.auth.callback.Callback; import javax.security.auth.callback.UnsupportedCallbackException; -import javax.security.auth.login.AppConfigurationEntry; -import javax.security.auth.login.Configuration; import javax.security.sasl.AuthorizeCallback; import javax.security.sasl.RealmCallback; import org.apache.kafka.common.security.kerberos.KerberosName; import org.apache.kafka.common.network.Mode; -import org.apache.kafka.common.security.JaasUtils; /** * Callback handler for Sasl servers. The callbacks required for all the SASL @@ -47,11 +45,10 @@ import org.apache.kafka.common.security.JaasUtils; public class SaslServerCallbackHandler implements AuthCallbackHandler { private static final Logger LOG = LoggerFactory.getLogger(SaslServerCallbackHandler.class); private final KerberosShortNamer kerberosShortNamer; + private final JaasContext jaasContext; - public SaslServerCallbackHandler(Configuration configuration, KerberosShortNamer 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."); + public SaslServerCallbackHandler(JaasContext jaasContext, KerberosShortNamer kerberosNameParser) throws IOException { + this.jaasContext = jaasContext; this.kerberosShortNamer = kerberosNameParser; } @@ -59,6 +56,10 @@ public class SaslServerCallbackHandler implements AuthCallbackHandler { public void configure(Map<String, ?> configs, Mode mode, Subject subject, String saslMechanism) { } + public JaasContext jaasContext() { + return jaasContext; + } + @Override public void handle(Callback[] callbacks) throws UnsupportedCallbackException { for (Callback callback : callbacks) { http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java index 32a0929..23d163c 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java +++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosLogin.java @@ -20,13 +20,12 @@ package org.apache.kafka.common.security.kerberos; import javax.security.auth.kerberos.KerberosPrincipal; import javax.security.auth.login.AppConfigurationEntry; -import javax.security.auth.login.Configuration; import javax.security.auth.login.LoginContext; import javax.security.auth.login.LoginException; import javax.security.auth.kerberos.KerberosTicket; import javax.security.auth.Subject; -import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.security.JaasContext; import org.apache.kafka.common.security.JaasUtils; import org.apache.kafka.common.security.authenticator.AbstractLogin; import org.apache.kafka.common.config.SaslConfigs; @@ -36,8 +35,8 @@ import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; import java.util.Date; +import java.util.List; import java.util.Random; import java.util.Set; import java.util.Map; @@ -56,7 +55,6 @@ public class KerberosLogin extends AbstractLogin { private boolean isKrbTicket; private boolean isUsingTicketCache; - private String loginContextName; private String principal; // LoginThread will sleep until 80% of time from last refresh to @@ -82,26 +80,19 @@ public class KerberosLogin extends AbstractLogin { private String serviceName; private long lastLogin; - /** - * Login constructor. The constructor starts the thread used - * to periodically re-login to the Kerberos Ticket Granting Server. - * @param loginContextName - * name of section in JAAS file that will be use to login. - * Passed as first param to javax.security.auth.login.LoginContext(). - * @param configs configure Login with the given key-value pairs. - * @throws javax.security.auth.login.LoginException - * Thrown if authentication fails. - */ - public void configure(Map<String, ?> configs, Configuration jaasConfig, final String loginContextName) { - super.configure(configs, jaasConfig, loginContextName); - this.loginContextName = loginContextName; + public void configure(Map<String, ?> configs, JaasContext jaasContext) { + super.configure(configs, jaasContext); this.ticketRenewWindowFactor = (Double) configs.get(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR); this.ticketRenewJitter = (Double) configs.get(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER); this.minTimeBeforeRelogin = (Long) configs.get(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN); this.kinitCmd = (String) configs.get(SaslConfigs.SASL_KERBEROS_KINIT_CMD); - this.serviceName = getServiceName(jaasConfig, configs, loginContextName); + this.serviceName = getServiceName(configs, jaasContext); } + /** + * Performs login for each login module specified for the login context of this instance and starts the thread used + * to periodically re-login to the Kerberos Ticket Granting Server. + */ @Override public LoginContext login() throws LoginException { @@ -110,13 +101,13 @@ public class KerberosLogin extends AbstractLogin { subject = loginContext.getSubject(); isKrbTicket = !subject.getPrivateCredentials(KerberosTicket.class).isEmpty(); - AppConfigurationEntry[] entries = jaasConfig().getAppConfigurationEntry(loginContextName); - if (entries.length == 0) { + List<AppConfigurationEntry> entries = jaasContext().configurationEntries(); + if (entries.isEmpty()) { isUsingTicketCache = false; principal = null; } else { // there will only be a single entry - AppConfigurationEntry entry = entries[0]; + AppConfigurationEntry entry = entries.get(0); if (entry.getOptions().get("useTicketCache") != null) { String val = (String) entry.getOptions().get("useTicketCache"); isUsingTicketCache = val.equals("true"); @@ -292,13 +283,8 @@ public class KerberosLogin extends AbstractLogin { return serviceName; } - private String getServiceName(Configuration jaasConfig, Map<String, ?> configs, String loginContext) { - String jaasServiceName; - try { - jaasServiceName = JaasUtils.jaasConfigOption(jaasConfig, loginContext, JaasUtils.SERVICE_NAME, null); - } catch (IOException e) { - throw new KafkaException("JAAS configuration entry not found", e); - } + private static String getServiceName(Map<String, ?> configs, JaasContext jaasContext) { + String jaasServiceName = jaasContext.configEntryOption(JaasUtils.SERVICE_NAME, null); String configServiceName = (String) configs.get(SaslConfigs.SASL_KERBEROS_SERVICE_NAME); if (jaasServiceName != null && configServiceName != null && !jaasServiceName.equals(configServiceName)) { String message = String.format("Conflicting serviceName values found in JAAS and Kafka configs " + @@ -377,7 +363,7 @@ public class KerberosLogin extends AbstractLogin { loginContext.logout(); //login and also update the subject field of this instance to //have the new credentials (pass it to the LoginContext constructor) - loginContext = new LoginContext(loginContextName, subject, null, jaasConfig()); + loginContext = new LoginContext(jaasContext().name(), subject, null, jaasContext().configuration()); log.info("Initiating re-login for {}", principal); loginContext.login(); } http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java b/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java index 0928057..1ad0223 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java +++ b/clients/src/main/java/org/apache/kafka/common/security/plain/PlainSaslServer.java @@ -18,7 +18,6 @@ package org.apache.kafka.common.security.plain; -import java.io.IOException; import java.io.UnsupportedEncodingException; import java.util.Arrays; import java.util.Map; @@ -29,7 +28,8 @@ import javax.security.sasl.SaslException; import javax.security.sasl.SaslServer; import javax.security.sasl.SaslServerFactory; -import org.apache.kafka.common.security.JaasUtils; +import org.apache.kafka.common.security.JaasContext; +import org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler; /** * Simple SaslServer implementation for SASL/PLAIN. In order to make this implementation @@ -49,10 +49,13 @@ public class PlainSaslServer implements SaslServer { public static final String PLAIN_MECHANISM = "PLAIN"; private static final String JAAS_USER_PREFIX = "user_"; + private final JaasContext jaasContext; + private boolean complete; private String authorizationID; - public PlainSaslServer(CallbackHandler callbackHandler) { + public PlainSaslServer(JaasContext jaasContext) { + this.jaasContext = jaasContext; } @Override @@ -91,13 +94,10 @@ public class PlainSaslServer implements SaslServer { if (authorizationID.isEmpty()) authorizationID = username; - try { - String expectedPassword = JaasUtils.defaultServerJaasConfigOption(JAAS_USER_PREFIX + username, PlainLoginModule.class.getName()); - if (!password.equals(expectedPassword)) { - throw new SaslException("Authentication failed: Invalid username or password"); - } - } catch (IOException e) { - throw new SaslException("Authentication failed: Invalid JAAS configuration", e); + String expectedPassword = jaasContext.configEntryOption(JAAS_USER_PREFIX + username, + PlainLoginModule.class.getName()); + if (!password.equals(expectedPassword)) { + throw new SaslException("Authentication failed: Invalid username or password"); } complete = true; return new byte[0]; @@ -151,10 +151,13 @@ public class PlainSaslServer implements SaslServer { public SaslServer createSaslServer(String mechanism, String protocol, String serverName, Map<String, ?> props, CallbackHandler cbh) throws SaslException { - if (!PLAIN_MECHANISM.equals(mechanism)) { + if (!PLAIN_MECHANISM.equals(mechanism)) throw new SaslException(String.format("Mechanism \'%s\' is not supported. Only PLAIN is supported.", mechanism)); - } - return new PlainSaslServer(cbh); + + if (!(cbh instanceof SaslServerCallbackHandler)) + throw new SaslException("CallbackHandler must be of type SaslServerCallbackHandler, but it is: " + cbh.getClass()); + + return new PlainSaslServer(((SaslServerCallbackHandler) cbh).jaasContext()); } @Override http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java index 46bfe57..002489c 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java +++ b/clients/src/main/java/org/apache/kafka/common/security/scram/ScramServerCallbackHandler.java @@ -57,4 +57,4 @@ public class ScramServerCallbackHandler implements AuthCallbackHandler { @Override public void close() { } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java index d483ef0..00c604e 100644 --- a/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java +++ b/clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java @@ -18,6 +18,7 @@ import org.apache.kafka.common.config.ConfigDef.Type; import org.apache.kafka.common.metrics.FakeMetricsReporter; import org.apache.kafka.common.metrics.JmxReporter; import org.apache.kafka.common.metrics.MetricsReporter; +import org.apache.kafka.common.security.TestSecurityConfig; import org.junit.Test; import java.util.Arrays; @@ -27,6 +28,7 @@ import java.util.Map; import java.util.Properties; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.junit.Assert.assertEquals; @@ -62,6 +64,62 @@ public class AbstractConfigTest { } @Test + public void testValuesWithPrefixOverride() { + String prefix = "prefix."; + Properties props = new Properties(); + props.put("sasl.mechanism", "PLAIN"); + props.put("prefix.sasl.mechanism", "GSSAPI"); + props.put("prefix.sasl.kerberos.kinit.cmd", "/usr/bin/kinit2"); + props.put("prefix.ssl.truststore.location", "my location"); + props.put("sasl.kerberos.service.name", "service name"); + props.put("ssl.keymanager.algorithm", "algorithm"); + TestSecurityConfig config = new TestSecurityConfig(props); + Map<String, Object> valuesWithPrefixOverride = config.valuesWithPrefixOverride(prefix); + + // prefix overrides global + assertTrue(config.unused().contains("prefix.sasl.mechanism")); + assertTrue(config.unused().contains("sasl.mechanism")); + assertEquals("GSSAPI", valuesWithPrefixOverride.get("sasl.mechanism")); + assertFalse(config.unused().contains("sasl.mechanism")); + assertFalse(config.unused().contains("prefix.sasl.mechanism")); + + // prefix overrides default + assertTrue(config.unused().contains("prefix.sasl.kerberos.kinit.cmd")); + assertFalse(config.unused().contains("sasl.kerberos.kinit.cmd")); + assertEquals("/usr/bin/kinit2", valuesWithPrefixOverride.get("sasl.kerberos.kinit.cmd")); + assertFalse(config.unused().contains("sasl.kerberos.kinit.cmd")); + assertFalse(config.unused().contains("prefix.sasl.kerberos.kinit.cmd")); + + // prefix override with no default + assertTrue(config.unused().contains("prefix.ssl.truststore.location")); + assertFalse(config.unused().contains("ssl.truststore.location")); + assertEquals("my location", valuesWithPrefixOverride.get("ssl.truststore.location")); + assertFalse(config.unused().contains("ssl.truststore.location")); + assertFalse(config.unused().contains("prefix.ssl.truststore.location")); + + // global overrides default + assertTrue(config.unused().contains("ssl.keymanager.algorithm")); + assertEquals("algorithm", valuesWithPrefixOverride.get("ssl.keymanager.algorithm")); + assertFalse(config.unused().contains("ssl.keymanager.algorithm")); + + // global with no default + assertTrue(config.unused().contains("sasl.kerberos.service.name")); + assertEquals("service name", valuesWithPrefixOverride.get("sasl.kerberos.service.name")); + assertFalse(config.unused().contains("sasl.kerberos.service.name")); + + // unset with default + assertFalse(config.unused().contains("sasl.kerberos.min.time.before.relogin")); + assertEquals(SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN, + valuesWithPrefixOverride.get("sasl.kerberos.min.time.before.relogin")); + assertFalse(config.unused().contains("sasl.kerberos.min.time.before.relogin")); + + // unset with no default + assertFalse(config.unused().contains("ssl.key.password")); + assertNull(valuesWithPrefixOverride.get("ssl.key.password")); + assertFalse(config.unused().contains("ssl.key.password")); + } + + @Test public void testUnused() { Properties props = new Properties(); String configValue = "org.apache.kafka.common.config.AbstractConfigTest$ConfiguredFakeMetricsReporter"; http://git-wip-us.apache.org/repos/asf/kafka/blob/ca0c071c/clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java b/clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java index 969055d..78c08d5 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java +++ b/clients/src/test/java/org/apache/kafka/common/network/NetworkTestUtils.java @@ -14,11 +14,11 @@ package org.apache.kafka.common.network; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.Map; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import org.apache.kafka.common.config.AbstractConfig; import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.protocol.SecurityProtocol; import org.apache.kafka.common.utils.MockTime; @@ -30,8 +30,9 @@ import org.apache.kafka.test.TestUtils; */ public class NetworkTestUtils { - public static NioEchoServer createEchoServer(SecurityProtocol securityProtocol, Map<String, Object> serverConfigs) throws Exception { - NioEchoServer server = new NioEchoServer(securityProtocol, serverConfigs, "localhost"); + public static NioEchoServer createEchoServer(ListenerName listenerName, SecurityProtocol securityProtocol, + AbstractConfig serverConfig) throws Exception { + NioEchoServer server = new NioEchoServer(listenerName, securityProtocol, serverConfig, "localhost"); server.start(); return server; } @@ -81,6 +82,6 @@ public class NetworkTestUtils { break; } } - assertTrue(closed); + assertTrue("Channel was not closed by timeout", closed); } }
