# Protect Data in Motion with TLS Encryption in Confluent Platform ## TLS encryption overview By default, Confluent Platform clusters communicate in `PLAINTEXT`, meaning that all data is sent in plain text (unencrypted). To encrypt data in motion (or data in transit) between services and components in your Confluent Platform cluster, you should configure all Confluent Platform services and components to use TLS encryption. TLS encryption is supported for all Confluent Platform services and components, including Confluent Server brokers, Connect workers, Kafka clients, and REST Proxy. For details on configuring TLS encryption for Confluent Server brokers and Confluent Platform services, see the following sections: * [Confluent Server brokers](#encryption-ssl-brokers) * [Clients](#encryption-ssl-clients) * [Kafka Connect](#encryption-ssl-connect) Confluent Platform supports [Transport Layer Security (TLS)](https://en.wikipedia.org/wiki/Transport_Layer_Security) encryption based on [OpenSSL](https://www.openssl.org/), an open source cryptography toolkit that provides an implementation of the Transport Layer Security (TLS) and Secure Socket Layer (SSL) protocols. Secure Sockets Layer (SSL) was the predecessor of Transport Layer Security (TLS), and has been deprecated since June 2015. For `TLS` configurations, properties often refer to `SSL`. Enabling TLS encryption might have a performance impact due to overhead of encrypting and decrypting data. TLS uses private-key/certificate pairs, which are used during the TLS handshake process. * Each Confluent Server broker needs its own private-key/certificate pair, and the Kafka client uses the certificate to authenticate to the Confluent Server broker. * Each logical client needs a private-key/certificate pair if client authentication is enabled, and the Confluent Server broker uses the certificate to authenticate the Kafka client. You can configure each Confluent Server broker and logical client with a truststore, which is used to determine which certificates (broker or logical client identities) to trust (authenticate). You can configure the truststore in many ways. Consider the following two examples: * The truststore contains one or many certificates: the Confluent Server broker or logical Kafka client trusts any certificate listed in the truststore. * The truststore contains a Certificate Authority (CA): the Confluent Server broker or logical Kafka client trusts any certificate that was signed by the CA in the truststore. Using the CA method is more convenient because adding a new Confluent Server broker or Kafka client doesn’t require changes to the truststore. For an example of the CA method, see [Single Trust Store Configuration](https://github.com/confluentinc/confluent-platform-security-tools/blob/master/single-trust-store-diagram.pdf). However, with the CA method, Confluent Platform clusters do not conveniently support blocking authentication for individual Confluent Server brokers or Kafka clients that were previously trusted using this mechanism (certificate revocation is typically done using Certificate Revocation Lists or the Online Certificate Status Protocol), so you would have to rely on authorization to block access. In contrast, if you use one or many certificates, blocking authentication is achieved by removing the Confluent Server broker certificate or Kafka client certificate from the truststore. ### Mutual (mTLS) authentication If you configure TLS encryption, you can optionally configure mutual (mTLS) authentication. You can configure just TLS encryption (by default, TLS encryption includes certificate authentication of the server) and use a separate mechanism for client authentication (for example, [mTLS](../authentication/mutual-tls/overview.md#kafka-ssl-authentication) or [SASL](../authentication/overview.md#kafka-sasl-auth)). By default, TLS encryption enables one-way authentication in which the client authenticates the server certificate. For bidirectional authentication, where the broker also authenticates the client certificate, you can use [mTLS](../authentication/mutual-tls/overview.md#kafka-ssl-authentication). When you use [mutual TLS (mTLS) authentication](../authentication/mutual-tls/overview.md#kafka-ssl-authentication), the Confluent Server broker authenticates the Kafka client and the Kafka client also authenticates the Confluent Server broker. This bidirectional, or mutual, authentication provides an additional layer of security for your Confluent Platform cluster. Note that mTLS is not supported for Kafka clients authenticating to ksqlDB. ## Create TLS keys and certificates Refer to the [Security Tutorial](../security_tutorial.md#security-tutorial), which describes how to [create TLS keys and certificates](../security_tutorial.md#generating-keys-certs). ## Configure TLS encryption for Confluent Server brokers Configure all Confluent Server brokers in your Confluent Platform cluster to accept secure connections from clients. Any configuration changes made to brokers require a [rolling restart](../../kafka/post-deployment.md#rolling-restart). Enable security for Confluent Server brokers as described in the section below. Additionally, if you are using Confluent Control Center , Auto Data Balancer, or Self-Balancing Clusters, configure your brokers for: * [Confluent Metrics Reporter](#encryption-ssl-metrics-reporter) * [Self-Balancing Clusters](#encryption-ssl-self-balancing) 1. Configure the password, truststore, and keystore in the `server.properties` file of every Confluent Server broker. Because this configuration stores passwords directly in the broker configuration file, restrict access to these files using file system permissions. ```bash ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks ssl.truststore.password=test1234 ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks ssl.keystore.password=test1234 ssl.key.password=test1234 ``` 2. To enable TLS for interbroker communication, add the following to the Confluent Server broker properties file (default is `PLAINTEXT`): ```bash security.inter.broker.protocol=SSL ``` 3. Configure the Confluent Server brokers with the ports to listen for client and interbroker TLS (`SSL`) connections. If the value is different from `listeners`, you must configure `listeners`, and optionally `advertised.listeners`. ```bash listeners=SSL://kafka1:9093 advertised.listeners=SSL://:9093 ``` 4. Configure both `SSL` ports and `PLAINTEXT` ports if: * TLS is not enabled for interbroker communication * Some Kafka clients connecting to the cluster do not use TLS ```bash listeners=PLAINTEXT://kafka1:9092,SSL://kafka1:9093 advertised.listeners=PLAINTEXT://:9092,SSL://:9093 ``` Note that `advertised.host.name` and `advertised.port` configure a single `PLAINTEXT` port and are incompatible with secure protocols. Use `advertised.listeners` instead. ### Optional settings Here are some optional settings for Confluent Server brokers: `ssl.cipher.suites` : The named list of supported cipher suites, which are named combinations of authentication, encryption, MAC, and key exchange algorithms used to negotiate the security settings for a network connection using TLS.
* Type: list * Default: `null` (by default, all supported cipher suites are enabled) * Importance: medium `ssl.enabled.protocols` : The comma-separated list of protocols enabled for TLS connections. Kafka clients and Confluent Server brokers use TLSv1.3 if both support it, but fall back to TLSv1.2 assuming both support TLSv1.2.
* Type: list * Default: `TLSv1.2,TLSv1.3` * Importance: medium `ssl.truststore.type` : The file format of the truststore file.
* Type: string * Default: `JKS` * Importance: medium Due to import regulations in some countries, the Oracle implementation limits the strength of cryptographic algorithms available by default. If stronger algorithms are needed (for example, AES with 256-bit keys), the [JCE Unlimited Strength Jurisdiction Policy Files](http://www.oracle.com/technetwork/java/javase/downloads/index.html) must be obtained and installed in the JDK or JRE. See [JCA Providers Documentation](https://docs.oracle.com/javase/8/docs/technotes/guides/security/SunProviders.html) for more information. ## Configure TLS encryption for 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. If client authentication is not required by the Confluent Server broker, the following is a minimal configuration example that you can store in a client properties file `client-ssl.properties`. Because this configuration stores passwords directly in the Kafka client configuration file, it is important to restrict access to these files via file system permissions. ```bash bootstrap.servers=kafka1:9093 security.protocol=SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 ``` If client authentication requires TLS, the client must provide the keystore as well. You can read about the additional configurations required in [mTLS authentication](../authentication/mutual-tls/overview.md#kafka-ssl-authentication). Here are examples using Kafka tools, `kafka-console-producer` and `kafka-console-consumer`, to pass in the `client-ssl.properties` file with the properties specified above: ```bash kafka-console-producer --bootstrap-server kafka1:9093 \ --topic test \ --producer.config client-ssl.properties kafka-console-consumer --bootstrap-server kafka1:9093 \ --topic test \ --consumer.config client-ssl.properties \ --from-beginning ``` ### Optional settings Here are some commonly used optional settings for Kafka clients: `ssl.provider` : The name of the security provider used for TLS connections. Default value is the default security provider of the JVM.
* Type: string * Default: `null` * Importance: medium `ssl.cipher.suites` : The named list of supported cipher suites, which are named combinations of authentication, encryption, MAC, and key exchange algorithms used to negotiate the security settings for a network connection using TLS.
* Type: list * Default: `null` (by default, all supported cipher suites are enabled) * Importance: medium `ssl.enabled.protocols` : The comma-separated list of protocols enabled for TLS connections. The default value is `TLSv1.2,TLSv1.3` when running with Java 11 or later, `TLSv1.2` otherwise. With the default value for Java 11 (`TLSv1.2,TLSv1.3`), Kafka clients and brokers prefer TLSv1.3 if both support it, but fall back to TLSv1.2 (assuming both support at least TLSv1.2).
* Type: list * Default: `TLSv1.2,TLSv1.3` * Importance: medium `ssl.truststore.type` : The file format of the truststore file.
* Type: string * Default: `JKS` * Importance: medium ## Configure TLS encryption for Connect workers 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](#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: Configure the top-level settings in the Connect workers to use TLS by adding these properties in `connect-distributed.properties`. These top-level settings are used by the Connect worker for group coordination and to read and write to the internal topics which are used to track the cluster’s state (for example, configurations and offsets). ```bash bootstrap.servers=kafka1:9093 security.protocol=SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 ``` Connect workers manage the producers used by source connectors and the consumers used by sink connectors. So, for the connectors to leverage security, you also have to override the default producer or consumer configuration that the worker uses. Depending on whether the connector is a source or sink connector: * For source connectors, configure the same properties, but add the `producer` prefix. ```bash producer.bootstrap.servers=kafka1:9093 producer.security.protocol=SSL producer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks producer.ssl.truststore.password=test1234 ``` * For sink connectors, configure the same properties, but add the `consumer` prefix. ```bash consumer.bootstrap.servers=kafka1:9093 consumer.security.protocol=SSL consumer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks consumer.ssl.truststore.password=test1234 ``` ## Configure TLS encryption for the REST API For TLS encryption with the REST API, configure the following additional properties: | Property | Note | |----------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `listeners` | List of REST listeners in the format


`protocol://host:port,protocol2://host2:port`,


where the protocol is either `http` or `https`.

| | `rest.advertised.listener` | Configures the listener used for communication between workers.


Valid values are either `http` or `https`.


If the `listeners` property is not defined or if it contains an `http`


listener, the default value for this field is `http`. When the `listeners`


property is defined and contains only `https` listeners, the default value is


`https`.

| | `ssl.client.auth` | Valid values are `none`, `requested`, and `required`. It controls whether:


1. the client is required to do TLS client authentication (`required`)


2. it can decide to skip the TLS client authentication (`requested`)


3. the TLS client authentication will be disabled (`none`)

| | `listeners.https.ssl.*` | You can use the `listeners.https.` prefix with an TLS configuration parameter


to override the default TLS configuration that is shared with the connections


to the Confluent Server broker. If at least one parameter with this prefix exists, the


implementation uses only the TLS parameters with this prefix and ignores all TLS


parameters without this prefix. If no parameter with prefix `listeners.https.`


exists, the parameters without a prefix are used.

| Note that if the `listeners.https.ssl.*` properties are not defined then the `ssl.*` properties will be used. For a list of all REST API `ssl.*` properties, see [Standalone REST Proxy Configuration Options for Confluent Platform](../../kafka-rest/production-deployment/rest-proxy/config.md#kafkarest-config). Here is an example that sets the `ssl.*` properties to use TLS connections to the Confluent Server broker, and since `listeners` includes `https` these same settings are used to configure the Connect TLS endpoint: ```bash listeners=https://myhost:8443 rest.advertised.listener=https rest.advertised.host.name= rest.advertised.host.port=8083 ssl.client.auth=requested ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks ssl.truststore.password=test1234 ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks ssl.keystore.password=test1234 ssl.key.password=test1234 ``` To configure the Connect TLS endpoint differently than the TLS connections to the Confluent Server broker, simply define the `listeners.https.ssl.*` properties with the correct settings. Note that as soon as any `listeners.https.ssl.*` properties are specified, none of the top level `ssl.*` properties will apply, so be sure to define all of the necessary `listeners.https.ssl.*` properties: ```bash listeners=https://myhost:8443 rest.advertised.listener=https rest.advertised.host.name= rest.advertised.host.port=8083 listeners.https.ssl.client.authentication=requested listeners.https.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks listeners.https.ssl.truststore.password=test1234 listeners.https.ssl.keystore.location=/var/ssl/private/kafka.server.keystore.jks listeners.https.ssl.keystore.password=test1234 listeners.https.ssl.key.password=test1234 listeners.https.ssl.endpoint.identification.algorithm=HTTPS ``` ## Configure TLS encryption for 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](#encryption-ssl-connect) To add TLS encryption to the Confluent Replicator embedded consumer, modify the Replicator JSON properties file. Here is an example subset of configuration properties to add for TLS encryption: ```bash { "name":"replicator", "config":{ .... "src.kafka.ssl.truststore.location":"/etc/kafka/secrets/kafka.connect.truststore.jks", "src.kafka.ssl.truststore.password":"confluent", "src.kafka.security.protocol":"SSL" .... } } } ``` #### SEE ALSO To see an example Confluent Replicator configuration, see the [TLS source demo script](https://github.com/confluentinc/examples/tree/latest//replicator-security/scripts/submit_replicator_source_ssl_encryption.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 TLS encryption, modify the Replicator JSON configuration to include the following: ```bash { "name":"replicator", "config":{ .... "dest.kafka.ssl.truststore.location":"/etc/kafka/secrets/kafka.connect.truststore.jks", "dest.kafka.ssl.truststore.password":"confluent", "dest.kafka.security.protocol":"SSL" .... } } } ``` Additionally the following properties are required in the Connect worker: ```bash security.protocol=SSL ssl.truststore.location=/etc/kafka/secrets/kafka.connect.truststore.jks ssl.truststore.password=confluent producer.security.protocol=SSL producer.ssl.truststore.location=/etc/kafka/secrets/kafka.connect.truststore.jks producer.ssl.truststore.password=confluent ``` For details, see the [general security configuration for Connect workers](../../connect/security.md#connect-security). #### SEE ALSO For an example Confluent Replicator configuration, see [TLS destination demo script](https://github.com/confluentinc/examples/tree/latest/replicator-security/scripts/submit_replicator_dest_ssl_encryption.sh). For demos of common security configurations, see: [Replicator security demos](https://github.com/confluentinc/examples/tree/latest/replicator-security). ## Configure TLS encryption for 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](#encryption-ssl-metrics-reporter): required on the production cluster being monitored Enable TLS encryption for Confluent Control Center in the `etc/confluent-control-center/control-center.properties` file. ```bash confluent.controlcenter.streams.security.protocol=SSL confluent.controlcenter.streams.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks confluent.controlcenter.streams.ssl.truststore.password=test1234 ``` ## Configure TLS encryption for Metrics Reporter This section describes how to enable TLS encryption for Confluent Metrics Reporter, which is used for Confluent Control Center and Auto Data Balancer. To add TLS encryption for the Confluent Metrics Reporter, add the following to the `server.properties` file on the Confluent Server brokers in the Confluent Platform cluster being monitored. ```bash confluent.metrics.reporter.bootstrap.servers=kafka1:9093 confluent.metrics.reporter.security.protocol=SSL confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks confluent.metrics.reporter.ssl.truststore.password=test1234 ``` ## Enable TLS encryption in a Self-Balancing cluster To enable TLS encryption in a Self-Balancing cluster, add the following to the `server.properties` file on the Confluent Server brokers in the Confluent Platform cluster. ```bash confluent.rebalancer.metrics.security.protocol=SSL confluent.rebalancer.metrics.ssl.truststore.location=/etc/kafka/secrets/kafka.client.truststore.jks confluent.rebalancer.metrics.ssl.truststore.password=confluent confluent.rebalancer.metrics.ssl.keystore.location=/etc/kafka/secrets/kafka.client.keystore.jks confluent.rebalancer.metrics.ssl.keystore.password=confluent confluent.rebalancer.metrics.ssl.key.password=confluent ``` ## 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). Here is an example subset of `schema-registry.properties` configuration parameters to add for TLS encryption: ```bash kafkastore.bootstrap.servers=SSL://kafka1:9093 kafkastore.security.protocol=SSL kafkastore.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks kafkastore.ssl.truststore.password=test1234 ``` ## Configure TLS encryption for REST Proxy To use TLS encryption with Confluent REST Proxy, you need to configure security between: * REST clients and the REST Proxy (HTTPS) * REST Proxy and the Confluent Platform cluster For the complete list of options, see [REST Proxy configuration options](../../kafka-rest/production-deployment/rest-proxy/security.md#kafkarest-security). First, configure HTTPS between REST clients and the REST Proxy. Here is an example subset of `kafka-rest.properties` configuration parameters to configure HTTPS: ```bash ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks ssl.truststore.password=test1234 ``` Then, configure TLS encryption between REST Proxy and the Confluent Platform cluster. Here is an example subset of `kafka-rest.properties` configuration parameters to add for TLS encryption: ```bash client.bootstrap.servers=kafka1:9093 client.security.protocol=SSL client.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks client.ssl.truststore.password=test1234 ``` ## Enable TLS logging Enable TLS debug logging at the JVM level by starting the Confluent Server broker and clients with the `javax.net.debug` system property. For example: ```bash export KAFKA_OPTS=-Djavax.net.debug=all kafka-server-start etc/kafka/server.properties ``` Once you start the Confluent Server broker, you should see the following in the `server.log`: ```bash with addresses: PLAINTEXT -> EndPoint(192.168.64.1,9092,PLAINTEXT),SSL -> EndPoint(192.168.64.1,9093,SSL) ``` To verify that the server keystore and truststore are configured correctly, run the following `openssl s_client` command: ```bash openssl s_client -debug -connect localhost:9093 -tls1 ``` Note: TLSv1 should be listed under `ssl.enabled.protocols`. In the output of this command you should see the server certificate: ```bash -----BEGIN CERTIFICATE----- {variable sized random bytes} -----END CERTIFICATE----- subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Joe Smith issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com ``` For details, see [Debugging SSL/TLS Connections [Java Documentation]](http://docs.oracle.com/javase/8/docs/technotes/guides/security/jsse/ReadDebug.html). If the certificate is not returned using the `openssl` command, or if there are other error messages, then your keys or certificates are not set up correctly. Review your configurations.