Important

You are viewing documentation for an older version of Confluent Platform. For the latest, click here.

Configuring PLAIN¶

Table of Contents

SASL/PLAIN Overview
Brokers
Clients
ZooKeeper
Kafka Connect
Confluent Replicator
Confluent Control Center
Confluent Metrics Reporter
Confluent Monitoring Interceptors
Schema Registry
REST Proxy

SASL/PLAIN Overview ¶

PLAIN, or SASL/PLAIN, is a simple username/password authentication mechanism that is typically used with TLS for encryption to implement secure authentication. Kafka supports a default implementation for SASL/PLAIN which can be extended for production use.

The username is used as the authenticated principal, which is used in authorization (e.g. ACLs).

Note

`PLAIN` versus `PLAINTEXT` Do not confuse the SASL mechanism PLAIN with no SSL encryption being called PLAINTEXT. Configuration parameters such as sasl.enabled.mechanisms or sasl.mechanism.inter.broker.protocol may be configured to use SASL mechanism PLAIN, whereas security.inter.broker.protocol or listeners may be configured for no SSL encryption SASL_PLAINTEXT.

SASL/PLAIN should only be used with SSL as transport layer to ensure that clear passwords are not transmitted on the wire without encryption.

The default implementation of SASL/PLAIN in Kafka specifies usernames and passwords in the JAAS configuration file. To avoid storing passwords on disk, you can plugin your own implementation of javax.security.auth.spi.LoginModule that provides usernames and passwords from an external source. The login module implementation should provide username as the public credential and password as the private credential of the Subject. The default implementation org.apache.kafka.common.security.plain.PlainLoginModule can be used as an example.

Important: If you are using a custom login module, you must also provide a custom implementation of org.apache.kafka.common.security.plain.PlainSaslServer and org.apache.kafka.common.security.plain.PlainSaslServerProvider.

In production systems, external authentication servers may implement password authentication. Kafka brokers can be integrated with these servers by adding your own implementation of javax.security.sasl.SaslServer. The default implementation included in Kafka in the package org.apache.kafka.common.security.plain can be used as an example to get started.

New providers must be installed and registered in the JVM. Providers can be installed by adding provider classes to the normal CLASSPATH or bundled as a jar file and added to {JAVA_HOME}/lib/ext.
Providers can be registered statically by adding a provider to the security properties file {JAVA_HOME}/lib/security/java.security where providerClassName is the fully qualified name of the new provider and n is the preference order with lower numbers indicating higher preference:
```
security.provider.n=providerClassName
```
Alternatively, you can register providers dynamically at runtime by invoking Security.addProvider at the beginning of the client application or in a static initializer in the login module. For example:
```
Security.addProvider(new PlainSaslServerProvider());
```
For more details, see JCA Reference.

The remainder of this page will show you how to configure SASL/PLAIN for each component in the Confluent Platform.

Brokers ¶

Configure all brokers in the Kafka cluster to accept secure connections from clients. Any configuration changes made to the broker will require a rolling restart.

Enable security for Kafka brokers as described in the section below. Additionally, if you are using Confluent Control Center or Auto Data Balancer, configure your brokers for:

Confluent Metrics Reporter

JAAS¶

First create the broker’s JAAS configuration file in each Kafka broker’s config directory, let’s call it kafka_server_jaas.conf for this example. This is the only way to configure JAAS for brokers; there is no broker configuration property sasl.jaas.config.
In each broker’s JAAS file, configure a KafkaServer section. This configuration defines two users (admin and kafkabroker1). The properties username and password 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, and the broker validates all client connections including those from other brokers using these properties.

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

Configuration¶

Enable SASL/PLAIN mechanism in the server.properties file of every broker.

# List of enabled mechanisms, can be more than one
sasl.enabled.mechanisms=PLAIN

# Specify one of of the SASL mechanisms
sasl.mechanism.inter.broker.protocol=PLAIN

If you want to enable SASL for inter-broker communication, add the following to the broker properties file (it defaults to PLAINTEXT). Set the protocol to:
- SASL_SSL: if SSL encryption is enabled (SSL encryption should always be used if SASL mechanism is PLAIN)
- SASL_PLAINTEXT: if SSL encryption is not enabled
```
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
security.inter.broker.protocol=SASL_SSL
```
Tell the Kafka brokers on which ports to listen for client and inter-broker SASL connections. You must configure listeners, and optionally advertised.listeners if the value is different from listeners. Set the listener to:
- SASL_SSL: if SSL encryption is enabled (SSL encryption should always be used if SASL mechanism is PLAIN)
- SASL_PLAINTEXT: if SSL encryption is not enabled
```
# With SSL encryption
listeners=SASL_SSL://kafka1:9093
advertised.listeners=SASL_SSL://0.0.0.0:9093

# Without SSL encryption
listeners=SASL_PLAINTEXT://kafka1:9093
advertised.listeners=SASL_PLAINTEXT://0.0.0.0:9093
```

Configure both SASL_SSL and PLAINTEXT ports if:

SASL is not enabled for inter-broker communication
Some clients connecting to the cluster do not use SASL

Example SASL listeners with SSL encryption, mixed with PLAINTEXT listeners

# With SSL encryption
listeners=PLAINTEXT://kafka1:9092,SASL_SSL://kafka1:9093
advertised.listeners=PLAINTEXT://0.0.0.0:9092,SASL_SSL://0.0.0.0:9093

# Without SSL encryption
listeners=PLAINTEXT://kafka1:9092,SASL_PLAINTEXT://kafka1:9093
advertised.listeners=PLAINTEXT://0.0.0.0:9092,SASL_PLAINTEXT://0.0.0.0:9093

Run¶

When you start each Kafka broker, pass the name of the JAAS file as a JVM parameter:

$ export KAFKA_OPTS=-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
$ bin/kafka-server-start etc/kafka/server.properties

Here are some optional settings that you can pass in as a JVM parameter when you start each broker from the command line.

-Dzookeeper.sasl.client For the SASL authentication to ZooKeeper, if you want to change the section name, set this system property to the appropriate name

Default: Client
Usage example: pass the parameter as a JVM parameter when you start the broker, e.g. -Dzookeeper.sasl.client=ZkClient

-Dzookeeper.sasl.client.username For the SASL authentication to ZooKeeper, if you want to change the service name, set the system property the appropriate name

Default: zookeeper
Usage example: pass the parameter as a JVM parameter when you start the broker, e.g. -Dzookeeper.sasl.client.username=zk

Clients ¶

The new Producer and Consumer clients support security for Kafka versions 0.9.0 and higher.

If you are using the Kafka Streams API, you can read on how to configure equivalent SSL and SASL parameters.

Configure the following properties in a client properties file client.properties.

sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
security.protocol=SASL_SSL

Configure the JAAS configuration property to describe how the clients like producer and consumer can connect to the Kafka Brokers. The properties username and password are used by clients to configure the user for client connections. In this example, clients connect to the broker as user kafkaclient1.

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="kafkaclient1" \
  password="kafkaclient1-secret";

ZooKeeper ¶

ZooKeeper does not support SASL/PLAIN authentication, but it does support another mechanism SASL/DIGEST-MD5.

For further details on ZooKeeper SASL authentication:

Client-Server mutual authentication : between the Kafka Broker (client) and ZooKeeper (server)
Server-Server mutual authentication: between the ZooKeeper nodes within an ensemble

Kafka Connect ¶

This section describes how to enable security for Kafka Connect. Securing Kafka Connect requires that you configure security for:

Kafka Connect workers: part of the Kafka Connect API, a worker is really just an advanced client, underneath the covers
Kafka Connect connectors: connectors may have embedded producers or consumers, so you must override the default configurations for Connect producers used with source connectors and Connect consumers used with sink connectors

Configure security for Kafka Connect as described in the section below. Additionally, if you are using Confluent Control Center streams monitoring for Kafka Connect, configure security for:

Confluent Monitoring Interceptors
Confluent Metrics Reporter

Configure all the following properties in connect-distributed.properties.

Configure the Connect workers to use SASL/PLAIN.

sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
security.protocol=SASL_SSL

Configure the JAAS configuration property to describe how Connect’s producers and consumers can connect to the Kafka Brokers. The properties username and password are used by Connect to configure the user for connections. In this example, Connect workers connect to the broker as user connect.

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

For the connectors to leverage security, you also have to override the default producer/consumer configuration that the worker uses. Depending on whether the connector is a source or sink connector:

Source connector: configure the same properties adding the producer prefix.

producer.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
producer.security.protocol=SASL_SSL
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

Sink connector: configure the same properties adding the consumer prefix.

consumer.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
consumer.security.protocol=SASL_SSL
consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";

Confluent Replicator ¶

Confluent Replicator is a type of Kafka source connector that replicates data from a source to destination Kafka cluster. An embedded consumer inside Replicator consumes data from the source cluster, and an embedded producer inside the Kafka Connect worker produces data to the destination cluster.

Replicator version 4.0 and earlier requires a connection to ZooKeeper in the origin and destination Kafka clusters. If ZooKeeper is configured for authentication, the client configures the ZooKeeper security credentials via the global JAAS configuration setting -Djava.security.auth.login.config on the Connect workers, and the ZooKeeper security credentials in the origin and destination clusters must be the same.

To configure Confluent Replicator security, you must configure the Replicator connector as shown below and additionally you must configure:

Kafka Connect

Configure Confluent Replicator to use SASL/PLAIN by adding these properties in the Replicator’s JSON configuration file. The JAAS configuration property defines username and password used by Replicator to configure the user for connections. In this example, Replicator connects to the broker as user replicator.

{
  "name":"replicator",
    "config":{
      ....
      "src.kafka.security.protocol" : "SASL_SSL",
      "src.kafka.sasl.mechanism" : "PLAIN",
      "src.kafka.sasl.jaas.config" : "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"replicator\" password=\"replicator-secret\";",
      ....
    }
  }
}

Confluent Control Center ¶

Confluent Control Center uses Kafka Streams as a state store, so if all the Kafka brokers in the cluster backing Control Center are secured, then the Control Center application also needs to be secured.

Enable security for the Control Center application as described in the section below. Additionally, configure security for the following components:

Confluent Metrics Reporter: required on the production cluster being monitored
Confluent Monitoring Interceptors: optional if you are using Control Center streams monitoring

Enable SASL/PLAIN and the security protocol for Control Center in the etc/confluent-control-center/control-center.properties file.

confluent.controlcenter.streams.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
confluent.controlcenter.streams.security.protocol=SASL_SSL

Configure the JAAS configuration property to describe how Confluent Control Center can connect to the Kafka Brokers. The properties username and password are used by Control Center to configure connections.

confluent.controlcenter.streams.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="confluent" \
  password="confluent-secret";

Confluent Metrics Reporter ¶

This section describes how to enable SASL/PLAIN for Confluent Metrics Reporter, which is used for Confluent Control Center and Auto Data Balancer.

To configure the Confluent Metrics Reporter for SASL/PLAIN, make the following configuration changes in the server.properties file in every broker in the production cluster being monitored.

Verify that the Confluent Metrics Reporter is enabled.

metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
confluent.metrics.reporter.bootstrap.servers=kafka1:9093

Enable the SASL/PLAIN mechanism for Confluent Metrics Reporter.

confluent.metrics.reporter.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
confluent.metrics.reporter.security.protocol=SASL_SSL

Confluent Monitoring Interceptors ¶

Confluent Monitoring Interceptors are used for Confluent Control Center streams monitoring. This section describes how to enable security for Confluent Monitoring Interceptors in three places:

General clients
Kafka Connect
Confluent Replicator

Interceptors for General Clients¶

For Confluent Control Center stream monitoring to work with Kafka clients, you must configure SASL/PLAIN for the Confluent Monitoring Interceptors in each client.

Verify that the client has configured interceptors.

Producer:

interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor

Consumer:

interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor

Configure the SASL mechanism and security protocol for the interceptor.

confluent.monitoring.interceptor.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
confluent.monitoring.interceptor.security.protocol=SASL_SSL

Configure the JAAS configuration property with a unique username and password.

confluent.monitoring.interceptor.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="confluent" \
  password="confluent-secret";

Interceptors for Kafka Connect¶

For Confluent Control Center stream monitoring to work with Kafka Connect, you must configure SASL/PLAIN for the Confluent Monitoring Interceptors in Kafka Connect. Configure the Connect workers by adding these properties in connect-distributed.properties, depending on whether the connectors are sources or sinks.

Source connector: configure the Confluent Monitoring Interceptors SASL mechanism with the producer prefix.

producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
producer.confluent.monitoring.interceptor.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
producer.confluent.monitoring.interceptor.security.protocol=SASL_SSL

Sink connector: configure the Confluent Monitoring Interceptors SASL mechanism with the consumer prefix.

consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
consumer.confluent.monitoring.interceptor.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
consumer.confluent.monitoring.interceptor.security.protocol=SASL_SSL

Configure the JAAS configuration property with a username and password.

Source connector: configure the Confluent Monitoring Interceptors JAAS configuration with the producer prefix.

producer.confluent.monitoring.interceptor.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="confluent" \
  password="confluent-secret";

Sink connector: configure the Confluent Monitoring Interceptors JAAS configuration with the consumer prefix.

consumer.confluent.monitoring.interceptor.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="confluent" \
  password="confluent-secret";

Interceptors for Replicator¶

For Confluent Control Center stream monitoring to work with Replicator, you must configure SASL for the Confluent Monitoring Interceptors in the Replicator JSON configuration file. Here is an example subset of configuration properties to add.

{
  "name":"replicator",
    "config":{
      ....
      "src.consumer.group.id": "replicator",
      "src.consumer.interceptor.classes": "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor",
      "src.consumer.confluent.monitoring.interceptor.sasl.mechanism": "PLAIN",
      "src.consumer.confluent.monitoring.interceptor.security.protocol": "SASL_SSL",
      "src.consumer.confluent.monitoring.interceptor.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required \nusername=\"confluent\" \npassword=\"confluent-secret\";",
      ....
    }
  }
}

Schema Registry ¶

Schema Registry uses Kafka to persist schemas, and so it acts as a client to write data to the Kafka cluster. Therefore, if the Kafka brokers are configured for security, you should also configure Schema Registry to use security. You may also refer to the complete list of Schema Registry configuration options.

Here is an example subset of schema-registry.properties configuration parameters to add for SASL authentication:

kafkastore.bootstrap.servers=SASL_SSL://kafka1:9093
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
kafkastore.security.protocol=SASL_SSL
kafkastore.sasl.mechanism=PLAIN

Configure the JAAS configuration property to describe how Schema Registry can connect to the Kafka Brokers. The properties username and password are used by Schema Registry to configure the user for connections. In this example, Schema Registry connects to the broker as user schemaregistry.

kafkastore.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="schemaregistry" \
  password="schemaregistry-secret";

REST Proxy ¶

Securing Confluent REST Proxy for SASL requires that you configure security between the REST proxy and the Kafka cluster.

You may also refer to the complete list of REST Proxy SASL configuration options.

Here is an example subset of kafka-rest.properties configuration parameters to add for SASL/PLAIN authentication

client.bootstrap.servers=kafka1:9093
client.sasl.mechanism=PLAIN
# Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
client.security.protocol=SASL_SSL

Configure the JAAS configuration property to describe how the REST Proxy can connect to the Kafka Brokers. The properties username and password are used by the REST Proxy to configure the user for connections. In this example, the REST Proxy connects to the broker as user restproxy.

client.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="restproxy" \
  password="restproxy-secret";