.. title:: Encrypt with TLS .. meta:: :description:: Use TLS to encrypt communication between components in your Confluent Cloud clusters. .. _kafka_ssl_encryption: Encrypt with TLS ---------------- TLS overview ~~~~~~~~~~~~ .. include:: includes/intro_ssl.rst .. include:: ../includes/cp-demo-tip.rst Create TLS keys and certificates ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Refer to the Security Tutorial, which describes how to :ref:`create TLS keys and certificates `. .. _encryption-ssl-brokers: Brokers ~~~~~~~ Configure all brokers in the |ak| cluster to accept secure connections from clients. Any configuration changes made to the broker will require a :ref:`rolling restart `. Enable security for |ak| brokers as described in the section below. Additionally, if you are using |c3|, Auto Data Balancer, or Self-Balancing, configure your brokers for: * :ref:`Confluent Metrics Reporter ` * :ref:`Self-Balancing ` 1. Configure the password, truststore, and keystore in the ``server.properties`` file of every broker. Since this stores passwords directly in the broker configuration file, it is important to restrict access to these files using file system permissions. .. codewithvars:: 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. If you want to enable TLS for interbroker communication, add the following to the broker properties file (it defaults to ``PLAINTEXT``): .. codewithvars:: bash security.inter.broker.protocol=SSL 3. Tell the |ak-tm| brokers on which ports to listen for client and interbroker ``SSL`` connections. You must configure ``listeners``, and optionally ``advertised.listeners`` if the value is different from ``listeners``. .. codewithvars:: 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 clients connecting to the cluster do not use TLS .. codewithvars:: 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: ``ssl.cipher.suites`` A cipher suite is a named combination 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``), |ak| clients and brokers prefer TLSv1.3 if both support it, and falls back to TLSv1.2 otherwise (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 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 `_ must be obtained and installed in the JDK/JRE. See the `JCA Providers Documentation `_ for more information. .. _encryption-ssl-clients: Clients ~~~~~~~ .. include:: includes/intro_clients.rst If client authentication is not required by the broker, the following is a minimal configuration example that you can store in a client properties file ``client-ssl.properties``. Since this stores passwords directly in the client configuration file, it is important to restrict access to these files via file system permissions. .. codewithvars:: 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 using TLS is required, the client must provide the keystore as well. You can read about the additional configurations required in :ref:`TLS Authentication `. Examples using ``kafka-console-producer`` and ``kafka-console-consumer``, passing in the ``client-ssl.properties`` file with the properties defined above: .. codewithvars:: bash kafka-console-producer --broker-list 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 .. _clients-optional-settings: Optional settings ^^^^^^^^^^^^^^^^^ Here are some optional settings: ``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`` A cipher suite is a named combination of authentication, encryption, MAC address, and key exchange algorithm used to negotiate the security settings for a network connection using the TLS network protocol. * 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``), |ak| clients and brokers prefer TLSv1.3 if both support it, and falls back to TLSv1.2 otherwise (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 .. _encryption-ssl-zookeeper: |zk| ~~~~ Starting in |cp| version 5.5.0, the version of |zk| that is bundled with |ak| supports TLS. For details, refer to :ref:`kafka_adding_security`. .. _encryption-ssl-connect: |kconnect-long| ~~~~~~~~~~~~~~~ .. include:: includes/intro_connect.rst * :ref:`Confluent Monitoring Interceptors ` * :ref:`Confluent Metrics Reporter ` .. _encryption-ssl-connect-workers: 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, configs and offsets). .. codewithvars:: 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/consumer configuration that the worker uses. Depending on whether the connector is a source or sink connector: * For source connectors: configure the same properties adding the ``producer`` prefix. .. codewithvars:: 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 adding the ``consumer`` prefix. .. codewithvars:: 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 .. _encryption-ssl-rest: * For TLS 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 |ak| 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 :ref:`kafkarest_config`. Here is an example that sets the ``ssl.*`` properties to use TLS connections to the broker, and since ``listeners`` includes ``https`` these same settings are used to configure Connect's TLS endpoint: .. codewithvars:: 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 Connect's TLS endpoint differently than the TLS connections to the 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: .. codewithvars:: 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 .. _encryption-ssl-replicator: |crep-full| ~~~~~~~~~~~ .. include:: includes/intro_replicator.rst * :ref:`Kafka Connect ` To add TLS to the |crep-full| embedded consumer, modify the Replicator JSON properties file. Here is an example subset of configuration properties to add for TLS encryption: .. codewithvars:: 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" .... } } } .. seealso:: To see an example |crep-full| configuration, see the :devx-examples:`TLS source demo script|/replicator-security/scripts/submit_replicator_source_ssl_encryption.sh`. For demos of common security configurations see: :devx-examples:`Replicator security demos|replicator-security` To configure |crep-full| for a destination cluster with TLS encryption, modify the |crep| JSON configuration to include the following: .. codewithvars:: 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: .. codewithvars:: 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 more information see the :ref:`general security configuration for Connect workers `. .. seealso:: To see an example |crep-full| configuration, see the :devx-examples:`TLS destination demo script|replicator-security/scripts/submit_replicator_dest_ssl_encryption.sh`. For demos of common security configurations see: :devx-examples:`Replicator security demos|replicator-security` .. _encryption-ssl-c3: |c3| ~~~~ .. include:: includes/intro_c3.rst * :ref:`Confluent Metrics Reporter `: required on the production cluster being monitored * :ref:`Confluent Monitoring Interceptors `: optional if you are using Control Center streams monitoring Enable TLS for Control Center in the ``etc/confluent-control-center/control-center.properties`` file. .. codewithvars:: 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 .. _encryption-ssl-metrics-reporter: |cmetric-full| ~~~~~~~~~~~~~~ This section describes how to enable TLS encryption for |cmetric-full|, which is used for |c3| and Auto Data Balancer. To add TLS for the |cmetric-full|, add the following to the ``server.properties`` file on the brokers in the |ak| cluster being monitored. .. codewithvars:: 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 .. _encryption-ssl-interceptors: Confluent Monitoring Interceptor ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. include:: includes/intro_interceptors.rst Interceptors for General Clients ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ For |c3| stream monitoring to work with |ak| clients, you must configure TLS encryption for the Confluent Monitoring Interceptors in each client. 1. Verify that the client has configured interceptors. * Producer: .. codewithvars:: bash interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor * Consumer: .. codewithvars:: bash interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor 2. Configure TLS encryption for the interceptor. .. codewithvars:: bash confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093 confluent.monitoring.interceptor.security.protocol=SSL confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks confluent.monitoring.interceptor.ssl.truststore.password=test1234 Interceptors for |kconnect-long| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ For |c3| stream monitoring to work with |kconnect-long|, you must configure TLS for the Confluent Monitoring Interceptors in |kconnect-long|. 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 for TLS encryption with the ``producer`` prefix. .. codewithvars:: bash producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor producer.confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093 producer.confluent.monitoring.interceptor.security.protocol=SSL producer.confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks producer.confluent.monitoring.interceptor.ssl.truststore.password=test1234 * Sink connector: configure the Confluent Monitoring Interceptors for TLS encryption with the ``consumer`` prefix. .. codewithvars:: bash consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor consumer.confluent.monitoring.interceptor.bootstrap.servers=kafka1:9093 consumer.confluent.monitoring.interceptor.security.protocol=SSL consumer.confluent.monitoring.interceptor.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks consumer.confluent.monitoring.interceptor.ssl.truststore.password=test1234 Interceptors for Replicator ^^^^^^^^^^^^^^^^^^^^^^^^^^^ For |c3| stream monitoring to work with Replicator, you must configure TLS for the Confluent Monitoring Interceptors in the Replicator JSON configuration file. Here is an example subset of configuration properties to add for TLS encryption. .. codewithvars:: bash { "name":"replicator", "config":{ .... "src.consumer.group.id": "replicator", "src.consumer.interceptor.classes": "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor", "src.consumer.confluent.monitoring.interceptor.bootstrap.servers": "kafka1:9093", "src.consumer.confluent.monitoring.interceptor.security.protocol": "SSL", "src.consumer.confluent.monitoring.interceptor.ssl.truststore.location": "/var/ssl/private/kafka.client.truststore.jks", "src.consumer.confluent.monitoring.interceptor.ssl.truststore.password": "test1234", .... } } } .. _encryption-ssl-self-balancing: Enable TLS in a Self-Balancing cluster ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ To enable TLS encryption in a Self-Balancing cluster, add the following to the ``server.properties`` file on the brokers in the |ak| cluster. .. codewithvars:: 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 .. _encryption-ssl-schema-registry: |sr| ~~~~ .. include:: includes/intro_sr.rst Here is an example subset of ``schema-registry.properties`` configuration parameters to add for TLS encryption: .. codewithvars:: 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 .. _encryption-ssl-rest-proxy: REST Proxy ~~~~~~~~~~ Securing Confluent REST Proxy with TLS encryption requires that you configure security between: #. REST clients and the REST Proxy (HTTPS) #. REST proxy and the |ak| cluster You may also refer to the complete list of :ref:`REST Proxy configuration options`. First, configure HTTPS between REST clients and the REST Proxy. Here is an example subset of ``kafka-rest.properties`` configuration parameters to configure HTTPS: .. codewithvars:: bash ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks ssl.truststore.password=test1234 Then, configure TLS encryption between REST proxy and the |ak| cluster. Here is an example subset of ``kafka-rest.properties`` configuration parameters to add for TLS encryption: .. codewithvars:: 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 .. include:: includes/ssl_logging.rst