<a id="kafka-sasl-auth-plain"></a>

# Use SASL/PLAIN Authentication in Confluent Platform

## SASL/PLAIN overview

PLAIN, or SASL/PLAIN, is a simple username and password authentication mechanism that
is typically used with TLS for encryption to implement secure authentication.
Apache Kafka® supports a [default implementation for SASL/PLAIN, which can be
extended for production use](../../../security_tutorial.md#security-tutorial).

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

The username is the authenticated `principal` used in authorization
(such as ACLs).

#### NOTE
**PLAIN compared to PLAINTEXT**: Do not confuse the SASL mechanism PLAIN with the no
TLS encryption option, which is called PLAINTEXT. Configuration parameters such as
`sasl.enabled.mechanisms` or `sasl.mechanism.inter.broker.protocol` can be
configured to use the SASL mechanism PLAIN, whereas
`security.inter.broker.protocol` or `listeners` can be configured to use
the no TLS encryption option, SASL_PLAINTEXT.

SASL/PLAIN should only be used with TLS as transport layer to ensure that
cleartext passwords are not transmitted without encryption.

The default implementation of SASL/PLAIN in Confluent Platform specifies usernames and passwords
in the JAAS configuration file. You can avoid storing cleartext passwords on disk by
configuring your own callback handlers that obtain username and password
from an external source using the configuration options `sasl.server.callback.handler.class`
and `sasl.client.callback.handler.class`.

In production systems, external authentication servers might implement password
authentication. You can plug in your own callback handlers that use external
authentication servers for password verification by configuring `sasl.server.callback.handler.class`.

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

<a id="sasl-plain-broker"></a>

## Configure Confluent Server 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](../../../../kafka/post-deployment.md#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](#sasl-plain-metrics-reporter)

<a id="auth-sasl-plain-broker-config"></a>

### Confluent Server broker SASL/PLAIN configuration

1. Enable SASL/PLAIN mechanism in the `server.properties` file of every broker.
   ```bash
   # 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
   ```

1. If you want to enable SASL for interbroker communication, add the following
   to the broker properties file (it defaults to `PLAINTEXT`).
   Set the protocol to:
   * `SASL_SSL`: if TLS/SSL encryption is enabled (TLS/SSL encryption should always be used if SASL mechanism is PLAIN)
   * `SASL_PLAINTEXT`: if TLS/SSL encryption is not enabled

   ```bash
   # Configure SASL_SSL if TLS/SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
   security.inter.broker.protocol=SASL_SSL
   ```
2. Tell the Kafka brokers on which ports to listen for client and interbroker
   `SASL` connections. You must configure `listeners`, and optionally
   `advertised.listeners` if the value is different from `listeners`.
   Set the listener to:
   * `SASL_SSL`: if TLS/SSL encryption is enabled (TLS/SSL encryption should always be used if SASL mechanism is PLAIN)
   * `SASL_PLAINTEXT`: if TLS/SSL encryption is not enabled

   ```none
   # With TLS/SSL encryption
   listeners=SASL_SSL://kafka1:9093
   advertised.listeners=SASL_SSL://localhost:9093

   # Without TLS/SSL encryption
   listeners=SASL_PLAINTEXT://kafka1:9093
   advertised.listeners=SASL_PLAINTEXT://localhost:9093
   ```
3. Configure both `SASL_SSL` and `PLAINTEXT` ports if:
   * SASL is not enabled for interbroker communication
   * Some clients connecting to the cluster do not use SASL

   Example SASL listeners with TLS/SSL encryption, mixed with PLAINTEXT listeners
   ```none
   # With TLS/SSL encryption
   listeners=PLAINTEXT://kafka1:9092,SASL_SSL://kafka1:9093
   advertised.listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093

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

1. If you are not using a separate JAAS configuration file to configure JAAS,
   then configure JAAS for the Kafka broker listener as follows:
   ```none
   # With TLS/SSL encryption
   listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="admin" \
      password="admin-secret" \
      user_admin="admin-secret" \
      user_kafkabroker1="kafkabroker1-secret";

   # Without TLS/SSL encryption
   listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
      username="admin" \
      password="admin-secret" \
      user_admin="admin-secret" \
      user_kafkabroker1="kafkabroker1-secret";
   ```

   The properties `username` and `password`  are used by the broker to initiate
   connections to other brokers.  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.

<a id="sasl-plain-clients"></a>

## Configure Kafka 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](/platform/current/clients/javadocs/javadoc/org/apache/kafka/common/config/SslConfigs.html) and
[SASL](/platform/current/clients/javadocs/javadoc/org/apache/kafka/common/config/SaslConfigs.html) parameters.

#### IMPORTANT
If you are configuring this for Schema Registry or REST Proxy, you must prefix each parameter with
`confluent.license`. For example, `sasl.mechanism` becomes
`confluent.license.sasl.mechanism`. For additional information, see
[Configure license clients to authenticate to Kafka](../../../../installation/license.md#kafka-rest-and-sasl-ssl-configs).

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

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

1. 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`.

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

<a id="sasl-plain-connect-workers"></a>

## Configure Kafka Connect

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

1. Kafka Connect workers: part of the Kafka Connect API, a worker is really just an advanced client, underneath the covers
2. 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
3. Kafka Connect REST: Kafka Connect exposes a REST API that can be configured to use TLS/SSL using [additional properties](../../../protect-data/encrypt-tls.md#encryption-ssl-rest)

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 Metrics Reporter](#sasl-plain-metrics-reporter)

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

1. Configure the Connect workers to use SASL/PLAIN.

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

1. 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`.

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

1. 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.

```bash
producer.sasl.mechanism=PLAIN
# Configure SASL_SSL if TLS/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.

```bash
consumer.sasl.mechanism=PLAIN
# Configure SASL_SSL if TLS/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";
```

<a id="sasl-plain-replicator"></a>

## Configure 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](#sasl-plain-connect-workers)

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`.

```bash
{
  "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\";",
      ....
    }
  }
}
```

#### SEE ALSO
To see an example Confluent Replicator configuration, see the [SASL source authentication demo script](https://github.com/confluentinc/examples/tree/latest//replicator-security/scripts/submit_replicator_source_sasl_plain_auth.sh). For demos of common security configurations see: [Replicator security demos](https://github.com/confluentinc/examples/tree/latest//replicator-security)

To configure Confluent Replicator for a destination cluster with SASL/PLAIN authentication, modify the Replicator JSON configuration to include the following:

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

Additionally the following properties are required in the Connect worker:

```bash
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="replicator" password="replicator-secret";
sasl.kerberos.service.name=kafka
producer.sasl.mechanism=GSSAPI
producer.security.protocol=SASL_SSL
producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="replicator" password="replicator-secret";
```

For more information see the general security configuration for Connect workers
[here](../../../../connect/security.md#connect-security).

#### SEE ALSO
To see an example Confluent Replicator configuration, see the [SASL destination authentication demo script](https://github.com/confluentinc/examples/tree/latest//replicator-security/scripts/submit_replicator_dest_sasl_plain_auth.sh). For demos of common security configurations see: [Replicator security demos](https://github.com/confluentinc/examples/tree/latest//replicator-security)

## Configure 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.

#### NOTE
When [RBAC](/control-center/current/security/c3-rbac.html) is enabled, Control Center
cannot be used in conjunction with Kerberos because Control Center cannot
support any SASL mechanism other than OAUTHBEARER.

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

* [Confluent Metrics Reporter](#sasl-plain-metrics-reporter): required on the production cluster being monitored

1. Enable SASL/PLAIN and the security protocol for Control Center in the
   `etc/confluent-control-center/control-center.properties` file.
   ```bash
   confluent.controlcenter.streams.sasl.mechanism=PLAIN
   # Configure SASL_SSL if TLS/SSL encryption is enabled; otherwise configure SASL_PLAINTEXT
   confluent.controlcenter.streams.security.protocol=SASL_SSL
   ```

1. Configure the JAAS configuration property to describe how Control Center can
   connect to the Kafka Brokers. The properties `username` and `password` are
   used by Control Center to configure connections.
   ```bash
   confluent.controlcenter.streams.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="confluent" \
   password="confluent-secret";
   ```

<a id="sasl-plain-metrics-reporter"></a>

## Configure Confluent Metrics Reporter

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.

1. Verify that the Confluent Metrics Reporter is enabled.

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

1. Enable the SASL/PLAIN mechanism for Confluent Metrics Reporter.

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

<a id="auth-sasl-plain-schema-registry"></a>

## Configure 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](../../../../schema-registry/installation/config.md#schemaregistry-config).

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

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

1. 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`.

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

## Configure REST Proxy

To secure Confluent REST Proxy for SASL you must configure security
between the REST proxy and the Confluent Platform cluster.

For a complete list of all configuration options, refer to [SASL Authentication](../../../../kafka-rest/production-deployment/rest-proxy/security.md#kafka-rest-security-kafka-auth-sasl).

1. Following is an example subset of `kafka-rest.properties` configuration parameters to add for SASL/PLAIN authentication:

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

1. 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`.

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