KAFKA-4292: Configurable SASL callback handlers #2022
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,22 +21,35 @@ | |
| import org.apache.kafka.common.config.SslConfigs; | ||
| import org.apache.kafka.common.config.internals.BrokerSecurityConfigs; | ||
| import org.apache.kafka.common.memory.MemoryPool; | ||
| import org.apache.kafka.common.security.JaasContext; | ||
| import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler; | ||
| import org.apache.kafka.common.security.auth.Login; | ||
| import org.apache.kafka.common.security.auth.SecurityProtocol; | ||
| import org.apache.kafka.common.security.authenticator.CredentialCache; | ||
| import org.apache.kafka.common.security.authenticator.DefaultLogin; | ||
| import org.apache.kafka.common.security.authenticator.LoginManager; | ||
| import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; | ||
| import org.apache.kafka.common.security.authenticator.SaslClientCallbackHandler; | ||
| import org.apache.kafka.common.security.authenticator.SaslServerAuthenticator; | ||
| import org.apache.kafka.common.security.authenticator.SaslServerCallbackHandler; | ||
| import org.apache.kafka.common.security.kerberos.KerberosClientCallbackHandler; | ||
| import org.apache.kafka.common.security.kerberos.KerberosLogin; | ||
| import org.apache.kafka.common.security.kerberos.KerberosShortNamer; | ||
| import org.apache.kafka.common.security.plain.internal.PlainSaslServer; | ||
| import org.apache.kafka.common.security.plain.internal.PlainServerCallbackHandler; | ||
| import org.apache.kafka.common.security.scram.ScramCredential; | ||
| import org.apache.kafka.common.security.scram.internal.ScramMechanism; | ||
| import org.apache.kafka.common.security.scram.internal.ScramServerCallbackHandler; | ||
| import org.apache.kafka.common.security.ssl.SslFactory; | ||
| import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; | ||
| import org.apache.kafka.common.utils.Java; | ||
| import org.apache.kafka.common.utils.Utils; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.lang.reflect.InvocationTargetException; | ||
| import java.lang.reflect.Method; | ||
| import java.io.IOException; | ||
| import java.net.Socket; | ||
| import java.nio.channels.SelectionKey; | ||
| import java.nio.channels.SocketChannel; | ||
|
|
@@ -66,6 +79,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl | |
| private SslFactory sslFactory; | ||
| private Map<String, ?> configs; | ||
| private KerberosShortNamer kerberosShortNamer; | ||
| private Map<String, AuthenticateCallbackHandler> saslCallbackHandlers; | ||
|
|
||
| public SaslChannelBuilder(Mode mode, | ||
| Map<String, JaasContext> jaasContexts, | ||
|
|
@@ -87,15 +101,25 @@ public SaslChannelBuilder(Mode mode, | |
| this.clientSaslMechanism = clientSaslMechanism; | ||
| this.credentialCache = credentialCache; | ||
| this.tokenCache = tokenCache; | ||
| this.saslCallbackHandlers = new HashMap<>(); | ||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| @Override | ||
| public void configure(Map<String, ?> configs) throws KafkaException { | ||
| try { | ||
| this.configs = configs; | ||
| if (mode == Mode.SERVER) | ||
| createServerCallbackHandlers(configs); | ||
| else | ||
| createClientCallbackHandler(configs); | ||
| for (Map.Entry<String, AuthenticateCallbackHandler> entry : saslCallbackHandlers.entrySet()) { | ||
| String mechanism = entry.getKey(); | ||
| entry.getValue().configure(configs, mechanism, jaasContexts.get(mechanism).configurationEntries()); | ||
| } | ||
|
|
||
| Class<? extends Login> defaultLoginClass = DefaultLogin.class; | ||
| if (jaasContexts.containsKey(SaslConfigs.GSSAPI_MECHANISM)) { | ||
| String defaultRealm; | ||
| try { | ||
| defaultRealm = defaultKerberosRealm(); | ||
|
|
@@ -106,12 +130,13 @@ public void configure(Map<String, ?> configs) throws KafkaException { | |
| List<String> principalToLocalRules = (List<String>) configs.get(BrokerSecurityConfigs.SASL_KERBEROS_PRINCIPAL_TO_LOCAL_RULES_CONFIG); | ||
| if (principalToLocalRules != null) | ||
| kerberosShortNamer = KerberosShortNamer.fromUnparsedRules(defaultRealm, principalToLocalRules); | ||
| defaultLoginClass = KerberosLogin.class; | ||
| } | ||
| for (Map.Entry<String, JaasContext> entry : jaasContexts.entrySet()) { | ||
| String mechanism = entry.getKey(); | ||
| // With static JAAS configuration, use KerberosLogin if Kerberos is enabled. With dynamic JAAS configuration, | ||
| // use KerberosLogin only for the LoginContext corresponding to GSSAPI | ||
| LoginManager loginManager = LoginManager.acquireLoginManager(entry.getValue(), mechanism, defaultLoginClass, configs); | ||
| loginManagers.put(mechanism, loginManager); | ||
| subjects.put(mechanism, loginManager.subject()); | ||
| } | ||
|
|
@@ -120,7 +145,7 @@ public void configure(Map<String, ?> configs) throws KafkaException { | |
| this.sslFactory = new SslFactory(mode, "none", isInterBrokerListener); | ||
| this.sslFactory.configure(configs); | ||
| } | ||
| } catch (Throwable e) { | ||
| close(); | ||
| throw new KafkaException(e); | ||
| } | ||
|
|
@@ -156,11 +181,20 @@ public KafkaChannel buildChannel(String id, SelectionKey key, int maxReceiveSize | |
| TransportLayer transportLayer = buildTransportLayer(id, key, socketChannel); | ||
| Authenticator authenticator; | ||
| if (mode == Mode.SERVER) { | ||
| authenticator = buildServerAuthenticator(configs, | ||
| saslCallbackHandlers, | ||
| id, | ||
| transportLayer, | ||
| subjects); | ||
| } else { | ||
| LoginManager loginManager = loginManagers.get(clientSaslMechanism); | ||
| authenticator = buildClientAuthenticator(configs, | ||
| saslCallbackHandlers.get(clientSaslMechanism), | ||
| id, | ||
| socket.getInetAddress().getHostName(), | ||
| loginManager.serviceName(), | ||
| transportLayer, | ||
| subjects.get(clientSaslMechanism)); | ||
| } | ||
| return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize, memoryPool != null ? memoryPool : MemoryPool.NONE); | ||
| } catch (Exception e) { | ||
|
|
@@ -174,6 +208,8 @@ public void close() { | |
| for (LoginManager loginManager : loginManagers.values()) | ||
| loginManager.release(); | ||
| loginManagers.clear(); | ||
| for (AuthenticateCallbackHandler handler : saslCallbackHandlers.values()) | ||
| handler.close(); | ||
| } | ||
|
|
||
| private TransportLayer buildTransportLayer(String id, SelectionKey key, SocketChannel socketChannel) throws IOException { | ||
|
|
@@ -186,16 +222,23 @@ private TransportLayer buildTransportLayer(String id, SelectionKey key, SocketCh | |
| } | ||
|
|
||
| // Visible to override for testing | ||
| protected SaslServerAuthenticator buildServerAuthenticator(Map<String, ?> configs, | ||
| Map<String, AuthenticateCallbackHandler> callbackHandlers, | ||
| String id, | ||
| TransportLayer transportLayer, | ||
| Map<String, Subject> subjects) throws IOException { | ||
| return new SaslServerAuthenticator(configs, callbackHandlers, id, subjects, | ||
| kerberosShortNamer, listenerName, securityProtocol, transportLayer); | ||
| } | ||
|
|
||
| // Visible to override for testing | ||
| protected SaslClientAuthenticator buildClientAuthenticator(Map<String, ?> configs, | ||
| AuthenticateCallbackHandler callbackHandler, | ||
| String id, | ||
| String serverHost, | ||
| String servicePrincipal, | ||
| TransportLayer transportLayer, Subject subject) throws IOException { | ||
| return new SaslClientAuthenticator(configs, callbackHandler, id, subject, servicePrincipal, | ||
| serverHost, clientSaslMechanism, handshakeRequestEnable, transportLayer); | ||
| } | ||
|
|
||
|
|
@@ -224,4 +267,30 @@ private static String defaultKerberosRealm() throws ClassNotFoundException, NoSu | |
| getDefaultRealmMethod = classRef.getDeclaredMethod("getDefaultRealm", new Class[0]); | ||
| return (String) getDefaultRealmMethod.invoke(kerbConf, new Object[0]); | ||
| } | ||
|
|
||
| private void createClientCallbackHandler(Map<String, ?> configs) { | ||
| Class<? extends AuthenticateCallbackHandler> clazz = (Class<? extends AuthenticateCallbackHandler>) configs.get(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS); | ||
|
Contributor
There was a problem hiding this comment. See comment below about the order of built-in mechanisms vs. non-built-in ones.
Contributor
Author
There was a problem hiding this comment. Same as before, callback handlers can be overridden for built-in mechanisms too. |
||
| if (clazz == null) | ||
| clazz = clientSaslMechanism.equals(SaslConfigs.GSSAPI_MECHANISM) ? KerberosClientCallbackHandler.class : SaslClientCallbackHandler.class; | ||
| AuthenticateCallbackHandler callbackHandler = Utils.newInstance(clazz); | ||
| saslCallbackHandlers.put(clientSaslMechanism, callbackHandler); | ||
| } | ||
|
|
||
| private void createServerCallbackHandlers(Map<String, ?> configs) throws ClassNotFoundException { | ||
| for (String mechanism : jaasContexts.keySet()) { | ||
| AuthenticateCallbackHandler callbackHandler; | ||
| String prefix = ListenerName.saslMechanismPrefix(mechanism); | ||
| Class<? extends AuthenticateCallbackHandler> clazz = | ||
| (Class<? extends AuthenticateCallbackHandler>) configs.get(prefix + BrokerSecurityConfigs.SASL_SERVER_CALLBACK_HANDLER_CLASS); | ||
| if (clazz != null) | ||
| callbackHandler = Utils.newInstance(clazz); | ||
| else if (mechanism.equals(PlainSaslServer.PLAIN_MECHANISM)) | ||
|
Contributor
There was a problem hiding this comment. Would it ever be acceptable to use the config map to override the callback handlers for the built-in mechanism implementations (GSSAPI, PLAIN, and the two SCRAM-related ones)? If so then the code is fine, but if not then the built-in mechanism names should be explicitly checked for first. Same goes for the client side checks above.
Contributor
Author
There was a problem hiding this comment. @rondagostino Thanks for the review. Yes, it is acceptable to override callback handlers for built-in mechanisms. One of the motivations for the KIP is to enable customization of built-in mechanisms to integrate with existing authentication servers (e.g. use an existing server or database to verify passwords for PLAIN). |
||
| callbackHandler = new PlainServerCallbackHandler(); | ||
| else if (ScramMechanism.isScram(mechanism)) | ||
| callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism, ScramCredential.class), tokenCache); | ||
| else | ||
| callbackHandler = new SaslServerCallbackHandler(); | ||
| saslCallbackHandlers.put(mechanism, callbackHandler); | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think this documentation should state that "For brokers, login callback handler config must be prefixed with listener prefix and SASL mechanism name in lower-case" because this is valid to use on both the client-side and the broker side.