Add Security to Running Clusters in Confluent Platform¶

Important

As of Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent recommends KRaft mode for new deployments. For more information, see KRaft Overview for Confluent Platform.

This topic describes how to add security (TLS or SASL) to a running ZooKeeper-based cluster.

Adding security to brokers and clients running TLS or SASL authentication¶

You can secure a running Confluent Platform cluster using one or more of the supported protocols. This is done in phases:

Incrementally restart the cluster nodes to open additional secured port(s).
Restart Kafka clients using the secured rather than PLAINTEXT port (assuming you are securing the client-broker connection).
Incrementally restart the cluster again to enable broker-to-broker security (if this is required).
A final incremental restart to close the PLAINTEXT port.

The specific steps for configuring security protocols are described in the respective sections for TLS and SASL. Follow these steps to enable security for your desired protocol(s).

The security implementation lets you configure different protocols for both broker-client and broker-broker communication. These must be enabled in separate restarts. A PLAINTEXT port must be left open throughout so brokers and/or clients can continue to communicate.

When performing an incremental restart, take into consideration the recommendations for doing rolling restarts to avoid downtime for end users.

For example, if you want to encrypt both broker-client and broker-broker communication with TLS:

In the first incremental restart, open a TLS port on each node:
```
listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092
```
Note

In Confluent Platform clusters, you can update some Confluent Server broker configurations without restarting the broker by adding or removing listeners dynamically. When adding a new listener, provide the security configuration of the listener using the listener prefix listener.name.{listenerName}. If the new listener uses SASL, then provide the JAAS configuration property sasl.jaas.config with the listener and mechanism prefix. For more details, refer to JAAS.
Then restart the Kafka clients, changing their configuration to point at the newly-opened, secured port:
```
bootstrap.servers=[broker1:9092,...]
security.protocol=SSL
...etc
```
For more details, refer to Protect Data in Motion with TLS Encryption in Confluent Platform.
In the second incremental server restart, instruct Confluent Platform to use TLS as the broker-broker protocol (which will use the same TLS port):
```
listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092
security.inter.broker.protocol=SSL
```
In the final restart, secure the cluster by closing the PLAINTEXT port:
```
listeners=SSL://broker1:9092
security.inter.broker.protocol=SSL
```

Alternatively, you might choose to open multiple ports so that different protocols can be used for broker-broker and broker-client communication. If you want to use TLS encryption throughout (for example, for broker-broker and broker-client communication), but also want to add SASL authentication to the broker-client connection:

Open two additional ports during the first restart:

listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092,SASL_SSL://broker1:9093

Again, restart the Kafka clients, changing their configuration to point at the newly-opened, SASL and TLS secured port:
```
bootstrap.servers=[broker1:9093,...]
security.protocol=SASL_SSL
...etc
```
For more details, refer to SASL.
The second server restart would switch the cluster to use encrypted broker-broker communication using the TLS port you previously opened on port 9092:
```
listeners=PLAINTEXT://broker1:9091,SSL://broker1:9092,SASL_SSL://broker1:9093
security.inter.broker.protocol=SSL
```

The final restart secures the cluster by closing the PLAINTEXT port:

listeners=SSL://broker1:9092,SASL_SSL://broker1:9093
security.inter.broker.protocol=SSL

Adding security to a running ZooKeeper cluster¶

This section describes how to add SASL (with or without TLS) or mTLS security to a running ZooKeeper cluster.

Adding SASL security to ZooKeeper¶

If you are running a Confluent Platform cluster with security disabled and you want to make the cluster secure, then you must perform the following steps to enable ZooKeeper authentication with minimal disruption to your operations:

Note

Migrating ZooKeeper security when the Confluent Platform cluster is not running (no controller node in ZooKeeper) results in a controller node being created, but populated with a null value. In this scenario, leader election may not work properly in subsequent starts of the Confluent Platform cluster.

In zookeeper.properties, add the authentication provider to enable ZooKeeper security:
```
authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
```

Export the ZooKeeper JAAS file before restarting ZooKeeper:

KAFKA_OPTS=-Djava.security.auth.login.config=<path>/zookeeper_server_jaas.conf

The content of zookeeper_server_jaas.conf should look like the following:

Server {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   user_super="adminsecret"
   user_bob="bobsecret";
};

Perform ZooKeeper rolling restarts with the JAAS login file shown above; this enables brokers to authenticate.
Create the JAAS file for the brokers to authenticate with ZooKeeper. Add the Client information to the broker JAAS configuration and then export the JAAS file:
```
export KAFKA_OPTS=-Djava.security.auth.login.config=<path>/kafka_server_jaas.conf
```
The content of the broker JAAS file (kafka_server_jaas.conf) should look like the following:
```
Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="bob"
   password="bobsecret";
};
```
Perform a rolling restart of brokers, this time setting the configuration parameter zookeeper.set.acl to true, which enables the use of secure ACLs when creating znodes.
Execute the ZkSecurityMigrator tool using the script: zookeeper-security-migration with zookeeper.acl set to secure. This tool traverses the corresponding sub-trees, changing the ACLs of the znodes.

If you wish to validate that security has been enabled between the broker and ZooKeeper:

Export the broker JAAS configuration:

export KAFKA_OPTS=-Djava.security.auth.login.config=<path>/kafka_server_jaas.conf

Create a new topic named test using the kafka-topic command:

kafka-topics --bootstrap-server localhost:9092 --create --topic test --partitions 2 --replication-factor 2

Log in to the zookeeper-shell and check the ACL of the newly-created znode, which should have the ACL enabled:

zookeeper-shell <zk_host>:<zk_port>

[zk: localhost:12181(CONNECTED) 9] getAcl /config/topics/test
'world,'anyone
: r
'sasl,'bob
: cdrwa

If you want to turn off authentication in a secure cluster:

Perform a rolling restart of brokers setting the JAAS login file, which enables brokers to authenticate, but setting zookeeper.set.acl to false. At the end of the rolling restart, brokers stop creating znodes with secure ACLs, but are still able to authenticate and manipulate all znodes.
Run the ZkSecurityMigrator tool using the script zookeeper-security-migration with zookeeper.acl set to unsecure. This tool traverses the corresponding sub-trees, changing the ACLs of the znodes.
Perform a second rolling restart of brokers, this time omitting the system property that sets the JAAS login file.

Here is an example of how to run the migration tool:

zookeeper-security-migration --zookeeper.acl=secure --zookeeper.connect=localhost:2181

Run this command to see the full list of parameters:

zookeeper-security-migration --help

Configuring TLS encryption for SASL security¶

To configure TLS encryption for SASL security:

In zookeeper.properties, add the authentication provider to enable ZooKeeper security:
```
authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
```
Optionally, to enable TLS encryption, specify the following configurations (be sure to use camel case for keyStore and trustStore). You will have both a clientPort and a secureClientPort at this point:
```
authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
secureClientPort=2182
serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
ssl.keyStore.location=<path-to-zookeeper-keystore>
ssl.keyStore.password=<zookeeper-keystore-password>
ssl.trustStore.location=<path-to-zookeeper-truststore>
ssl.trustStore.password=<zookeeper-truststore-password>
ssl.clientAuth=none
```
Important

ZooKeeper does not support setting the key password in the ZooKeeper server keystore to a value different from the keystore password itself. Be sure to set the key password to be the same as the keystore password.

Export the ZooKeeper JAAS file before restarting ZooKeeper:

KAFKA_OPTS=-Djava.security.auth.login.config=<path>/zookeeper_server_jaas.conf

The content of zookeeper_server_jaas.conf should look like the following:

Server {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   user_super="adminsecret"
   user_bob="bobsecret";
};

Perform ZooKeeper rolling restarts with the JAAS login file shown above; this enables brokers to authenticate.
Create the JAAS file for the brokers to authenticate with ZooKeeper. Add the Client information to the broker JAAS configuration and then export the JAAS file:
```
export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-kafka_server_jaas.conf>
```
The content of the broker JAAS file (kafka_server_jaas.conf) should look like the following:
```
Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="bob"
   password="bobsecret";
};
```

Optionally, if enabling TLS encryption to ZooKeeper, add these broker configurations (do not use camel case in truststore):

# Connect to the ZooKeeper port configured for TLS
zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
# Required to use TLS to ZooKeeper (default is false)
zookeeper.ssl.client.enable=true
# Required to use TLS to ZooKeeper
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
# Define trust store to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

Perform a rolling restart of brokers, this time setting zookeeper.set.acl=true, which enables the use of secure ACLs when creating znodes.
Execute the ZkSecurityMigrator tool using the scriptv zookeeper-security-migration with zookeeper.acl set to secure. This tool traverses the corresponding sub-trees, changing the ACLs of the znodes.

Optionally, if enabling TLS encryption to ZooKeeper, connect to the TLS-encrypted port on ZooKeeper (for example, 2182) and specify --zk-tls-config-file <path-to-tls-configs> (note the double-dash) with this content for the file (do not use camel case in truststore):

zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

To validate that security has been enabled between the broker and ZooKeeper:

Export the broker JAAS configuration:

export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-kafka_server_jaas.conf>

Create a new topic named test:

kafka-topics --bootstrap-server <kafka-host>:<kafka-port> --create --topic test --partitions 2 --replication-factor 2

Log in to the zookeeper-shell and check the ACL of the newly-created znode, which should have the ACL enabled:

zookeeper-shell <zk_host>:<zk_port>

[zk: localhost:12181(CONNECTED) 9] getAcl /config/topics/test
'world,'anyone
: r
'sasl,'bob
: cdrwa

Optionally, if enabling TLS encryption to ZooKeeper you can connect to the TLS port with zookeeper-shell by including the command line flags -zk-tls-config-file <path-to-tls-configs> (use a single dash) with this content for the file (do not use camel case in truststore):

zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

If you have enabled TLS encryption, then you can now remove the clientPort configuration in ZooKeeper.

Adding mTLS security to ZooKeeper¶

To add mTLS security to a running cluster with minimal disruption to operations:

Enable SASL and/or mTLS authentication on ZooKeeper. If enabling mTLS, you would have both a non-TLS port and a TLS port, as shown here:

clientPort=2181
secureClientPort=2182
serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
ssl.keyStore.location=<path-to-zookeeper-keystore>
ssl.keyStore.password=<zookeeper-keystore-password>
ssl.trustStore.location=<path-to-zookeeper-truststore>
ssl.trustStore.password=<zookeeper-truststore-password>

Important

ZooKeeper does not support setting the key password in the ZooKeeper server keystore to a value different from the keystore password itself. Be sure to set the key password to be the same as the keystore password.

Perform a rolling restart of brokers.

Specify the JAAS login file and/or defining ZooKeeper mTLS configurations (including connections to the TLS-enabled ZooKeeper port) as required. This enables brokers to authenticate to ZooKeeper.

At the end of the rolling restart, brokers can manipulate znodes with strict ACLs, but will not create znodes with those ACLs. Note that unlike ZooKeeper, Confluent Platform does not use camel case names for TLS-related configurations (for example, Confluent Platform uses zookeeper.ssl.keystore.location, while ZooKeeper uses ssl.keyStore.location.

# Connect to the ZooKeeper port configured for TLS
zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
# Required to use TLS to ZooKeeper (default is false)
zookeeper.ssl.client.enable=true
# Required to use TLS to ZooKeeper
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
# Define key/trust stores to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
zookeeper.ssl.keystore.location=<path-to-kafka-keystore>
zookeeper.ssl.keystore.password=<kafka-keystore-password>
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>

If you enabled mTLS, disable the non-TLS port in ZooKeeper:

secureClientPort=2182
serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
ssl.keyStore.location=<path-to-zookeeper-keystore>
ssl.keyStore.password=<zookeeper-keystore-password>
ssl.trustStore.location=<path-to-zookeeper-truststore>
ssl.trustStore.password=<zookeeper-truststore-password>

Perform a second rolling restart of Confluent Server brokers, this time setting zookeeper.set.acl=true, which enables the use of secure ACLs when creating znodes.

# Connect to the ZooKeeper port configured for TLS
zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
# Required to use TLS to ZooKeeper (default is false)
zookeeper.ssl.client.enable=true
# Required to use TLS to ZooKeeper
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
# Define key/trust stores to use TLS to ZooKeeper; ignored unless zookeeper.ssl.client.enable=true
zookeeper.ssl.keystore.location=<path-to-kafka-keystore>
zookeeper.ssl.keystore.password=<kafka-keystore-password>
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>
# Tell broker to create ACLs on znodes
zookeeper.set.acl=true

Run the ZkSecurityMigrator tool using the script: zookeeper-security-migration with zookeeper.acl set to secure. This tool traverses the corresponding sub-trees, changing the ACLs of the znodes. Because you are enabling mTLS, specify the --zk-tls-config-file <file> option.

zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=localhost:2182 --zk-tls-config-file <path-to-tls-config-file.properties>

The TLS configuration should look like this:

zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.truststore.location=<path-to-kafka-truststore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>
zookeeper.ssl.keystore.password-<encrypted-keystore-password>

To turn off mTLS authentication in this secure cluster:

Perform a rolling restart of Confluent Server brokers setting the JAAS login file and/or defining ZooKeeper mTLS configurations, which enables brokers to authenticate, but setting zookeeper.set.acl to false. At the end of the rolling restart, brokers stop creating znodes with secure ACLs, but are still able to authenticate and manipulate all znodes.
Run the ZkSecurityMigrator tool using the script zookeeper-security-migration with zookeeper.acl set to unsecure. This tool traverses the corresponding sub-trees, changing the ACLs of the znodes. Specify --zk-tls-config-file <file> if you need to set TLS configuration.
```
--zookeeper.connect=localhost:2182
```

If you are disabling mTLS, enable the non-TLS port in ZooKeeper:

clientPort=2181
secureClientPort=2182
serverCnxnFactory=org.apache.zookeeper.server.NettyServerCnxnFactory
authProvider.x509=org.apache.zookeeper.server.auth.X509AuthenticationProvider
ssl.keyStore.location=<path-to-zookeeper-keystore>
ssl.keyStore.password=<zookeeper-keystore-password>
ssl.trustStore.location=<path-to-zookeeper-truststore>
ssl.trustStore.password=<zookeeper-truststore-password>

Perform a second rolling restart of Confluent Server brokers, this time omitting the system property that sets the JAAS login file and/or removes ZooKeeper mTLS configuration (including connection to the non-TLS-enabled ZooKeeper port) as required.
If you are disabling mTLS, disable the TLS port in Confluent Platform:
```
clientPort=2181
```

Migrating the ZooKeeper ensemble¶

You must also enable authentication on the ZooKeeper ensemble. This requires that you perform a rolling restart of the ensemble and set a few properties. Refer to the ZooKeeper documentation for more detail: