Kafka Security - Part 3

So far in the series, I have covered how to configure SASL with PlainText (Part 1)  and how to have different listeners for broker and consumers (Part 2)

In this part, I would dissect the org.apache.kafka.common.security.plain.PlainLoginModule which was configured for SASL with PlainText in Part 1 of this series. This would give insight into developing your custom LoginModule over SASL.

PlainLoginModule: 

public class PlainLoginModule implements LoginModule {
    private static final String USERNAME_CONFIG = "username";
    private static final String PASSWORD_CONFIG = "password";

    static {
        PlainSaslServerProvider.initialize();
    }

public PlainLoginModule() {
    }

    public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
        String username = (String)options.get("username");
        if(username != null) {
            subject.getPublicCredentials().add(username);
        }

        String password = (String)options.get("password");
        if(password != null) {
            subject.getPrivateCredentials().add(password);
        }

    }

    public boolean login() throws LoginException {
        return true;
    }

    public boolean logout() throws LoginException {
        return true;
    }

    public boolean commit() throws LoginException {
        return true;
    }

    public boolean abort() throws LoginException {
        return false;
    }

}

The same class PlainLoginModule is used for dual purpose which makes it confusing.

1. The Broker to pass in username and password to authenticate against other Brokers
2. The Broker to initialize a Server Provider which supports authentication using SASL

In the initialize method of the LoginModule, the Broker adds the configured username and password in the Subject. 

This Subject is used by this broker to send in the username/password for authentication by the other user. 

 public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
        String username = (String)options.get("username");
        if(username != null) {
            subject.getPublicCredentials().add(username);
        }

        String password = (String)options.get("password");
        if(password != null) {
            subject.getPrivateCredentials().add(password);
        }

    }

In the static block of the PlainLoginModule class, we are initializing the Sasl Server Provider

    static {
        PlainSaslServerProvider.initialize();
    }

PlainLoginModule is used to configure the SASL server provider in the Kafka Broker. 

The Broker uses the SASL server Provider to authenticate the in-coming call to the broker.

In the initialize method of PlainSaslServerProvider, we are configuring PlainSaslServerProvider as a Security Provider.

The PlainSaslServerProvider configures PlainSaslServerFactory as the SaslServerFactory. 

public class PlainSaslServerProvider extends Provider {
    private static final long serialVersionUID = 1L;

    protected PlainSaslServerProvider() {
        super("Simple SASL/PLAIN Server Provider", 1.0D, "Simple SASL/PLAIN Server Provider for Kafka");
        super.put("SaslServerFactory.PLAIN", PlainSaslServerFactory.class.getName());
    }

    public static void initialize() {
        Security.addProvider(new PlainSaslServerProvider());
    }
}

The PlainSaslServerFactory provides PlainSaslServer as the SaslServer. 

PlainSaslServer evaluates the username/password in the evaluateResponse method. 

Kafka Security - Part 2

In the previous blog post Kafka Security - Part 1, I describe in length on how to setup SASL Plaintext as the security mechanism.

In this blog post, I would explain on how to setup security differently for Broker to Broker communication and Client (Producer/Consumer) to Broker communication.

In Kafka, you can set up multiple listeners

1. Setup two Listeners in servers.properties:
   
   listeners=CLIENT_LISTENER://:9091, BROKER_LISTENER://:9092
   
2. Setup security map for the custom listeners:
   
   listener.security.protocol.map= CLIENT_LISTENER:SASL_PLAINTEXT, BROKER_LISTENER: SASL_PLAINTEXT
   
3. Setup broker to broker communication to use BROKER_LISTENER
   
   security.inter.broker.protocol=BROKER_LISTENER
   
4. Setup broker to broker communication to use PLAIN, for this scenario
   
   sasl.mechanism.inter.broker.protocol=PLAIN
   
5. Setup Client (Producer/Consumer) to Broker communication to use PLAIN for this case.
   
   sasl.enabled.mechanisms=PLAIN

For this scenario, we expect clients to communicate with broker on port 9091 and brokers to communicate with another broker on port 9092.

If you have Kafka sitting in a different VLAN than your client, then in that case, you can have port 9092 (for Broker-Broker communication) not exposed out side the VLAN

Kafka Security - Part 1

Custom Kafka Security - Part 1

In this series, my hope is to explain on how to configure Kafka Security and how to setup custom Kafka Security. The current documentation about configuring Kafka Security is basically limited to few pages. I hoping to fill in the gaps and provide more information about my findings. As a disclaimer, I do not pretend to be a security expert or Kafka expert.

Kafka supports various types of authentication methods as described in the Apache Kafka documentation

In this first case, we would like to secure Broker to Broker and Client to Broker using SASL/Plain and without TLS.

Authentication using SASL/PLAIN without TLS

This article provides the configuration for authentication of connections to brokers from clients (producers and consumers), using SASL without TLS. 

SASL/PLAIN is a simple username/password authentication mechanism to implement secure authentication. Kafka supports a default implementation for SASL/PLAIN. Apache Kafka documentation provides detail on how it can be extended for production here.The username is used as the authenticated Principal for configuration of ACLs etc.

The first step is to configure the broker to authenticate the in-coming call using SASL with PlainText. This call can come from other brokers or clients such as consumer or producer.

The second step is to configure the broker to call other brokers using SASL with PlainText. 

The third steps is to configure the clients (producer/consumer) to use SASL with PlainText.

"Configuring Kafka Brokers" section takes care of the first and second steps.

Configuring Kafka Brokers: 

1. Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="admin-secret"
   user_admin="admin-secret"
   user_alice="alice-secret";
};

Let's try to understand this: 

The properties username and password in the KafkaServer section are used by the broker to initiate connections to other brokers. In this example, admin is the user for inter-broker communication. 

The set of properties user_userName defines the passwords for all users that connect to the broker. The broker validates all client connections including those from other brokers using these properties. This configuration defines two users (admin and alice). 

2. Pass the JAAS config file location as JVM parameter to each Kafka broker:

-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf

3. Configure SASL port and SASL mechanisms in server.properties. These are properties that would be configured:

listeners=OUR_LISTENER://:9092
advertised.listeners= OUR_LISTENER://:9092
security.inter.broker.protocol= OUR_LISTENER
listener.security.protocol.map= OUR_LISTENER:SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

Let's try to understand this:

listeners=OUR_LISTENER://:9092

advertised.listeners= OUR_LISTENER://:9092

# You are defining a listener with the name OUR_LISTENER.  You can give any name to the listener. 

security.inter.broker.protocol= OUR_LISTENER
#The listener to communicate with for Broker to Broker communication

listener.security.protocol.map= OUR_LISTENER:SASL_PLAINTEXT

#The actual mapping of security protocol to listener name.
#Here we are specifying that we are using SASL with PLAINTEXT (not SSL)

sasl.mechanism.inter.broker.protocol=PLAIN

#The inter broker SASL mechanism uses PLAIN text (not SSL)

sasl.enabled.mechanisms=PLAIN

#The client broker SASL mechanism uses PLAIN text (not SSL)

SASL with PlainTexy Configuration for Kafka clients

1. Add a suitably modified JAAS config file to one below to the Client directory

KafkaClient { 
     org.apache.kafka.common.security.plain.PlainLoginModule required
     username="alice"
     password="alice-secret";
};

Let's try to understand this :
The username used to call the broker is specified as "alice" with password "alice-secret". 

2. Pass the JAAS config file location as JVM parameter to each client JVM. For example:

-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf

3. Instead of 1 and 2, you can provided the JAAS configuration in the Client properties

props.put("sasl.jaas.config", 
        "org.apache.kafka.common.security.plain.PlainLoginModule required\n" +
        "username=\"alice\"\n" +
        "password=\"alice-secret\";");

Here, we have configured to secure the Kafka Broker to use SASL with PlainText.
In the next blog kafka-security-part-2, I would explain on how to setup Kafka configuration for Broker to Broker communication and Client to Broker communication