Status

Current state: "Under DiscussionAccepted"

Discussion thread: here

JIRA: KAFKA-4292

...

To enable custom SASL mechanisms to be plugged in easily, the login interface for clients and servers will also be made configurable. This will enable custom login logic to be added for new SASL mechanisms. This is particularly useful for mechanisms like Kerberos that require periodic token refresh.

...

• Name: sasl.client.callback.handler.class
• Type: CLASS
• Doc:A  The fully qualified name of a Sasl client callback handler class that implements the org.apache.kafka.common.security.auth.AuthenticateCallbackHandler interface.
• Default: null (by default, the appropriate internal default callback handlers for the mechanism will be used)
  

Server callback handler classes (for brokers only)

• Name: sasl.server.callback.handler.class.map
• Type: STRING
• Doc: A map between Sasl mechanisms and Sasl The fully qualified name of a SASL server callback handler classes class that implement implements the AuthenticateCallbackHandler interface. Key and value are separated by a colon and map entries are separated by commas. For example, PLAIN=CustomPlainCallbackHandler,SCRAM-SHA-256=CustomScramCallbackHandler.The config name must be prefixed by the listener prefix and mechanism name in lower case. For example, listener.name.sasl_ssl.plain.sasl.server.callback.handler.class=com.example.CustomPlainCallbackHandler.
• Default: null (by default, the appropriate internal default callback handlers for each mechanism will be used)
  

Login class (for clients and brokers)

• Name: sasl.login.class
• Type: CLASS
• Doc: A class that implements the org.apache.kafka.common.security.auth.Login interface. For brokers, the config name must be prefixed by the listener prefix and mechanism name in lower case. For example, listener.name.sasl_ssl.plain.sasl.login.class=com.example.PlainServerLogin for brokers and sasl.login.class=com.example.KerberosClientLogin for clients.
  
• Default: null (by default, the internal class KerberosLogin will be used if Kerberos is enabled on the listener and DefaultLogin otherwise)

 Login callback handler class (for clients and brokers)

• Name: sasl.login.callback.handler.class
• Type: CLASS
• Doc: The fully qualified name of a Sasl login callback handler class that implements the org.apache.kafka.common.security.auth.AuthenticateCallbackHandler interface. For servers, the config name must be prefixed by the listener prefix and mechanism name in lower case. For example, listener.name.sasl_ssl.plain.sasl.login.callback.handler.class=com.example.PlainLoginCallbackHandler for brokers and sasl.login.callback.handler.class=com.example.PlainLoginCallbackHandler for clients.
  
• Default: null (by default, the internal class AbstractLogin.DefaultLoginCallbackHandler will be used): A class that implements the org.apache.kafka.common.security.auth.Login interface.
  

Callback Handler

The callback handler interface AuthenticateCallbackHandler will extend the standard javax.security.auth.callback.CallbackHandler interface, enabling the handler to be passed directly to SaslServer/SaslClient implementations. The callback handler configured for a mechanism must include the callbacks as described below:

...

package org.apache.kafka.common.security.auth;

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;

/**
 * Login interface for authentication.
 */
public interface Login {

    /**
     * Configures this login instance.
     */
    void configure(Map<String, ?> configs, String contextName, Configuration configuration,
                   AuthenticateCallbackHandler loginCallbackHandler);

    /**
     * Performs login for each login module specified for the login context of this instance.
     */
    LoginContext login() throws LoginException;

    /**
     * Returns the authenticated subject of this login context.
     */
    Subject subject();

    /**
     * Returns the service name to be used for SASL.
     */
    String serviceName();

    /**
     * Closes this instance.
     */
    void close();
}

...

 Define a new class that implements AuthenticateCallbackHandler  which handles NameCallback and PlainAuthenticateCallback and add the class to the broker's sasl.server.callback.handler.class.map property. A single instance of this callback handler will be created for the broker. The configured callback handler is responsible for validating the password provided by clients and this may use an external authentication server.

...

If a custom SaslServer implementation is used instead of the one included in Kafka, the custom implementation may require a different set of callbacks. A callback handler for these callbacks may be specified in sasl.server.callback.handler.class.map.

Configure a new mechanism not included in Kafka using custom SaslServer/SaslClient

A handler that handles any callbacks required for these server/client implementations may be specified in sasl.server.callback.handler.class .map and sasl.client.callback.handler.class for brokers and clients respectively.

...

KIP-103 introduced support for multiple listeners in the broker for the same security protocol. This allows brokers to configure different SASL mechanisms for internal and external traffic. The listener name prefix can be applied to sasl.server.callback.handler.class.map to define different callback handlers for each of the listeners.

...

public class PlainServerCallbackHandler implements AuthenticateCallbackHandler {
    private List<AppConfigurationEntry> jaasConfigEntries;
    @Override
    public void configure(Map<String, ?> configs, String mechanism, List<AppConfigurationEntry> jaasConfigEntries) {
        this.jaasConfigEntries = jaasConfigEntries;
    }
    @Override
    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
        String username = null;
        for (Callback callback: callbacks) {
            if (callback instanceof NameCallback)
                username = ((NameCallback) callback).getDefaultName();
            else if (callback instanceof PlainAuthenticateCallback) {
                PlainAuthenticateCallback plainCallback = (PlainAuthenticateCallback) callback;
                boolean authenticated = authenticate(username, plainCallback.password());
                plainCallback.authenticated(authenticated);
            } else
                throw new UnsupportedCallbackException(callback);
        }
    }
    protected boolean authenticate(String username, char[] password) throws IOException {
        if (username == null)
            return false;
        else {
            String// expectedPasswordReturn = JaasContext.configEntryOption(jaasConfigEntries, "user_" + username, PlainLoginModule.class.getName());
            return Arrays.equals(password, expectedPassword.toCharArray());true if password matches expected password
        }
    }
    @Override
    public void close() throws KafkaException {
    }
}

...