# Enable Security for a KRaft-Based Cluster in Confluent Platform This tutorial provides an example of how to enable security on a Kraft-based Confluent Platform. ## Overview This tutorial provides a step-by-step example to enable [TLS/SSL encryption](protect-data/encrypt-tls.md#kafka-ssl-encryption), [SASL authentication](authentication/overview.md#kafka-sasl-auth), and [authorization](authorization/acls/overview.md#kafka-authorization) on Confluent Platform with monitoring using Confluent Control Center. Follow the steps to walk through configuration settings for securing Apache Kafka® brokers, Kafka Connect, and Confluent Replicator, plus all the components required for monitoring, including the Confluent Metrics Reporter. When working through the tutorial, be aware of the following: * For simplicity, this tutorial uses [SASL/PLAIN (or PLAIN)](authentication/sasl/plain/overview.md#kafka-sasl-auth-plain), a simple username/password authentication mechanism typically used with TLS encryption to implement secure authentication. * For production deployments of Confluent Platform, [SASL/GSSAPI (Kerberos)](authentication/sasl/gssapi/overview.md#kafka-sasl-auth-gssapi) or [SASL/SCRAM](authentication/sasl/scram/overview.md#kafka-sasl-auth-scram) is recommended. * Confluent Cloud uses [SASL/PLAIN (or PLAIN)](authentication/sasl/plain/overview.md#kafka-sasl-auth-plain) over TLS v1.2 encryption for authentication because it offers broad client support while providing a good level of security. The usernames and passwords used in the SASL exchange are API keys and secrets that should be securely managed using a secrets store and rotated periodically. ### Prerequisites You should understand why it is critical to secure Confluent Platform and have a conceptual understanding of how encryption, authentication, and authorization work. Before proceeding with this tutorial: * Use the [Quick Start for Confluent Platform](../get-started/platform-quickstart.md#quickstart) to bring up Confluent Platform *without* security enabled. * Verify that you can run Confluent Platform *without* security enabled. * *Do not* proceed until you have verified you can run Confluent Platform *without* security enabled, proceed with the security configuration additions and validation in this tutorial. ## Creating TLS/SSL Keys and Certificates Each machine in the cluster has a public-private key pair, and a certificate to identify the machine. The certificate, however, is unsigned, which means that an attacker can create such a certificate to pretend to be any machine. Therefore, it is important to prevent forged certificates by signing them for each machine in the cluster. A certificate authority (CA) is responsible for signing certificates. CA works like a government that issues passports - the government validates the identity of the person applying for the passport and then provides a passport in a standard form that is difficult to forge. Other governments verify the form is valid to ensure the passport is authentic. Similarly, the CA signs the certificates, and the cryptography guarantees that a signed certificate is computationally difficult to forge. Thus, as long as the CA is a genuine and trusted authority, the clients have high assurance that they are connecting to the authentic machines. The keystore stores each machine’s own identity. The truststore stores all the certificates that the machine should trust. Importing a certificate into one’s truststore also means trusting all certificates that are signed by that certificate authority. As the analogy above, trusting the certificate authority (CA) also means trusting all passports (certificates) that it has issued. This attribute is called the chain of trust, and it is particularly useful when deploying SSL on a large Confluent Platform cluster. You can sign all certificates in the cluster with a single CA, and have all machines share the same truststore that trusts the CA. That way all machines can authenticate all other machines. #### IMPORTANT OpenTLS/SSL certificates may include an [Extended Key Usage](https://www.openssl.org/docs/manmaster/man5/x509v3_config.html#Extended-Key-Usage) extension (`extendedKeyUsage`) to control the purpose for which the certificate public key can be used. If this field is empty, there are no restrictions on usage, but if any usage is specified, valid TLS/SSL implementations must enforce the restrictions. Extension key usages are relevant for client and server authentication. Confluent Server brokers require both client and server authentication for intracluster communication because every broker is both the client and the server for the other brokers. Some corporate CAs may have a signing profile for web servers †hat is used for Confluent Platform as well and only include the `serverAuth` usage value, causing the TLS/SSL handshake to fail. To deploy SSL, the general steps are: 1. Generate the keys and certificates. 2. Create your own Certificate Authority (CA). 3. Sign the certificate. The steps to create keys and sign certificates are enumerated below. You may also adapt the `kafka-generate-ssl` script from [confluent-platform-security-tools.git](https://github.com/confluentinc/confluent-platform-security-tools/blob/master/kafka-generate-ssl.sh). The definitions of the parameters used in the steps are as follows: * keystore: the location of the keystore * ca-cert: the certificate of the CA * ca-key: the private key of the CA * ca-password: the passphrase of the CA * cert-file: the exported, unsigned certificate of the server * cert-signed: the signed certificate of the server #### NOTE After a connection has been established, Confluent Platform does not perform certificate renegotiations or revocations. In cases where a certificate is compromised, or you wish to revoke a certificate, use ACL blacklists (specifically, the `--deny-principal` or `--deny-host` options) to remove a specific client. For details about ACLs, see [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](authorization/acls/overview.md#kafka-authorization). ### Configure host name verification Host name verification of servers is enabled by default for client connections as well as interbroker connections to prevent man-in-the-middle attacks. Server host name verification may be disabled by setting `ssl.endpoint.identification.algorithm` to an empty string. For example, ```bash ssl.endpoint.identification.algorithm= ``` For dynamically configured Confluent Server broker listeners, hostname verification may be disabled using `kafka-configs`. For example, ```bash ./bin/kafka-configs --bootstrap-server localhost:9093 --entity-type brokers --entity-name 0 --alter \ --add-config "listener.name.internal.ssl.endpoint.identification.algorithm=" ``` ### Configure host name In certificates If host name verification is enabled, clients verify the server’s fully qualified domain name (FQDN) against one of the following two fields: * Common Name (CN) * Subject Alternative Name (SAN) Both fields are valid, however [RFC-2818](https://tools.ietf.org/html/rfc2818) recommends the use of SAN. SAN is also more flexible, allowing for multiple DNS entries to be declared. Another advantage is that the CN can be set to a more meaningful value for authorization purposes. To add a SAN field, append the argument `-ext SAN=DNS:{FQDN}` to the keytool command: ```bash keytool -keystore kafka.server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA -ext SAN=DNS:{FQDN} ``` The following command can be run afterwards to verify the contents of the generated certificate: ```bash keytool -list -v -keystore kafka.server.keystore.jks ``` ### Generate the keys and certificates You can use the Java `keytool` utility for this process. Consult the [Java documentation](https://docs.oracle.com/javase/8/docs/technotes/tools/unix/keytool.html) for more information on the commands and arguments. 1. Generate the key and the certificate for each Confluent Server broker in the cluster. Generate the key into a keystore called `kafka.server.keystore` so that you can export and sign it later with CA. The keystore file contains the private key of the certificate; therefore, it needs to be kept safely. ```bash # With user prompts keytool -keystore kafka.server.keystore.jks -alias localhost -keyalg RSA -genkey # Without user prompts, pass command line arguments keytool -keystore kafka.server.keystore.jks -alias localhost -keyalg RSA -validity {validity} -genkey -storepass {keystore-pass} -keypass {key-pass} -dname {distinguished-name} -ext SAN=DNS:{hostname} ``` Ensure that the common name (CN) exactly matches the fully qualified domain name (FQDN) of the server. The client compares the CN with the DNS domain name to ensure that it is indeed connecting to the desired server, not a malicious one. The hostname of the server can also be specified in the Subject Alternative Name (SAN). Since the distinguished name is used as the server principal when TLS/SSL is used as the interbroker security protocol, it is useful to have hostname as a SAN rather than the CN. ### Create your own Certificate Authority (CA) 1. Generate a CA that is simply a public-private key pair and certificate, and it is intended to sign other certificates. ```bash openssl req -new -x509 -keyout ca-key -out ca-cert -days {validity} ``` 2. Add the generated CA to the **clients’ truststore** so that the clients can trust this CA: ```bash keytool -keystore kafka.client.truststore.jks -alias CARoot -importcert -file ca-cert ``` 3. Add the generated CA to the **brokers’ truststore** so that the brokers can trust this CA. ```bash keytool -keystore kafka.server.truststore.jks -alias CARoot -importcert -file ca-cert ``` ### Sign the certificate To sign all certificates in the keystore with the CA that you generated: 1. Export the certificate from the keystore: ```bash keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file ``` 2. Sign it with the CA: ```bash openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days {validity} -CAcreateserial -passin pass:{ca-password} ``` 3. Import both the certificate of the CA and the signed certificate into the broker keystore: ```bash keytool -keystore kafka.server.keystore.jks -alias CARoot -importcert -file ca-cert keytool -keystore kafka.server.keystore.jks -alias localhost -importcert -file cert-signed ``` ### Summary Combining the steps described above, the script to create the CA and broker and client truststores and keystores is as follows: ```bash keytool -keystore kafka.server.keystore.jks -alias localhost -keyalg RSA -validity {validity} -genkey openssl req -new -x509 -keyout ca-key -out ca-cert -days {validity} keytool -keystore kafka.client.truststore.jks -alias CARoot -importcert -file ca-cert keytool -keystore kafka.server.truststore.jks -alias CARoot -importcert -file ca-cert keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days {validity} -CAcreateserial -passin pass:{ca-password} keytool -keystore kafka.server.keystore.jks -alias CARoot -importcert -file ca-cert keytool -keystore kafka.server.keystore.jks -alias localhost -importcert -file cert-signed ``` #### NOTE In this tutorial, the client does not need a keystore because client authentication is done using SASL/PLAIN instead of mutual TLS (mTLS). However, if you use mTLS authentication, you create a client keystore and sign all certificates with the CA that you generated, similarly as done for the brokers. ## Configure Confluent Server brokers Administrators can configure a mix of secure and unsecured clients. This tutorial ensures that all broker/client and interbroker network communication is encrypted in the following manner: * All broker/client communication use `SASL_SSL` security protocol, which ensures that the communication is encrypted and authenticated using SASL/PLAIN. * All interbroker communication use `SSL` security protocol, which ensures that the communication is encrypted and authenticated using TLS. * The unsecured `PLAINTEXT` port is not enabled. The steps are as follows: 1. Enable the desired security protocols and ports in each Confluent Server broker’s `server.properties`. Notice that both `SSL` and `SASL_SSL` are enabled. ```bash listeners=SSL://:9093,SASL_SSL://:9094 # Kraft-specific configurations for the broker role # process.roles should be 'broker' for a dedicated broker node, 'controller' for a dedicated controller node, # or 'broker,controller' for a combined node. process.roles=broker,controller node.id={unique_node_id} # Unique ID for this broker/controller node # The list of controller nodes in the Kraft quorum. # Format: @:,@:,... # For example: 1@localhost:9093,2@localhost:9093,3@localhost:9093 controller.quorum.voters={node_id_1}@{host_1}:{port_1},{node_id_2}@{host_2}:{port_2},{node_id_3}@{host_3}:{port_3} ``` 2. To enable the Confluent Server brokers to authenticate each other using mutual TLS (mTLS) authentication, you need to configure all the Confluent Server brokers for client authentication (in this case, the requesting broker is the “client”). We recommend setting `ssl.client.auth=required`. We discourage configuring it as `requested` because misconfigured brokers will still connect successfully and it provides a false sense of security. ```bash security.inter.broker.protocol=SSL ssl.client.auth=required ``` 3. Define the TLS/SSL truststore, keystore, and password in the `server.properties` file of every Confluent Server broker. Because this stores passwords directly in the Confluent Server broker configuration file, it is important to 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 ``` 4. Enable SASL/PLAIN mechanism in the `server.properties` file of every broker. ```bash sasl.enabled.mechanisms=PLAIN ``` 5. Create the broker’s JAAS configuration file in each Confluent Server broker’s `config` directory, let’s call it `kafka_server_jaas.conf` for this example. * Configure a `KafkaServer` section used when the broker validates client connections, including those from other brokers. The broker properties `username` and `password` are used to initiate connections to other brokers, and in this example, `kafkabroker` is the user for interbroker communication. The `user_{userName}` property set defines the passwords for all other clients that connect to the broker. In this example, there are two users `kafkabroker` and `client`. #### NOTE Note the two semicolons in each section. ```bash KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="kafkabroker" password="kafkabroker-secret" user_kafkabroker="kafkabroker-secret" user_kafka-broker-metric-reporter="kafkabroker-metric-reporter-secret" user_client="client-secret"; }; ``` 1. If you are using Confluent Control Center to monitor your deployment, and if the monitoring cluster backing Confluent Control Center is also configured with the same security protocols, you must configure the Confluent Metrics Reporter for security as well. Add these configurations to the `server.properties` file of each Confluent Server broker. ```bash metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter confluent.metrics.reporter.security.protocol=SASL_SSL confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks confluent.metrics.reporter.ssl.truststore.password=test1234 confluent.metrics.reporter.sasl.mechanism=PLAIN confluent.metrics.reporter.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="kafka-broker-metric-reporter" \ password="kafka-broker-metric-reporter-secret"; ``` 2. To enable ACLs, we need to configure an authorizer. Kafka provides a simple authorizer implementation, and to use it, you can add the following to `server.properties`: ```shell authorizer.class.name=kafka.security.authorizer.AclAuthorizer ``` 3. The default behavior is such that if a resource has no associated ACLs, then no one is allowed to access the resource, except super users. Setting Confluent Server broker principals as super users is a convenient way to give them the required access to perform interbroker operations. Because this tutorial configures the interbroker security protocol as SSL, set the super user name to be the `distinguished name` configured in the broker’s certificate. (See other [authorization configuration options](authorization/acls/overview.md#kafka-auth-superuser)). ```bash .. comment: START_KRAFT_ADDITIONS_SUPER_USERS super.users=User:;User:;User:;User:;User:kafka-broker-metric-reporter .. comment: END_KRAFT_ADDITIONS_SUPER_USERS ``` Combining the configuration steps described above, the Confluent Server broker’s `server.properties` file contains the following configuration settings: ```bash .. comment: START_KRAFT_ADDITIONS_COMBINED # Kraft-specific configurations for the broker role process.roles=broker,controller node.id={unique_node_id} controller.quorum.voters={node_id_1}@{host_1}:{port_1},{node_id_2}@{host_2}:{port_2},{node_id_3}@{host_3}:{port_3} .. comment: END_KRAFT_ADDITIONS_COMBINED # Enable TLS/SSL security protocol for interbroker communication # Enable SASL_SSL security protocol for broker-client communication listeners=SSL://:9093,SASL_SSL://:9094 security.inter.broker.protocol=SSL ssl.client.auth=required # Broker security settings 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 sasl.enabled.mechanisms=PLAIN # Confluent Metrics Reporter for monitoring with Confluent Control Center metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter confluent.metrics.reporter.security.protocol=SASL_SSL confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/kafka.server.truststore.jks confluent.metrics.reporter.ssl.truststore.password=test1234 confluent.metrics.reporter.sasl.mechanism=PLAIN confluent.metrics.reporter.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="kafka-broker-metric-reporter" \ password="kafkabroker-metric-reporter-secret"; # ACLs authorizer.class.name=kafka.security.authorizer.AclAuthorizer super.users=User:;User:;User:;User: ``` #### NOTE This is not the full Confluent Server broker configuration. This is just the additional configurations required to enable security on a known working Confluent Platform cluster of Confluent Server brokers that is already successfully monitored using Confluent Control Center . Start each Confluent Server broker. Pass the name of the JAAS file as a JVM parameter: ```shell export KAFKA_OPTS=-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf kafka-server-start etc/kafka/server.properties ``` Alternatively, you can modify the configuration file with the following: ```bash 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"; ``` The properties `username` and `password` are used by the broker to initiate connections to other brokers. The set of properties `user_` 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. ## Configure Kafka clients ### Common configuration Any component that interacts with secured Confluent Server brokers is a *client* and must be configured for security as well. These clients include Kafka Connect workers and certain connectors such as Replicator, ksqlDB clients, non-Java clients, Confluent Control Center , Confluent Schema Registry, REST Proxy, etc. All Kafka clients share a general set of security configuration parameters required to interact with a secured Confluent Platform cluster: 1. To encrypt data using TLS/SSL and authenticate using SASL, configure the security protocol to use `SASL_SSL`. (If you want TLS/SSL for both encryption and authentication without SASL, the security protocol would be `SSL`). ```bash security.protocol=SASL_SSL ``` 2. To configure TLS encryption truststore settings, set the truststore configuration parameters. In this tutorial, the Kafka client does not need the keystore because authentication is done using SASL/PLAIN instead of mutual TLS (mTLS). ```bash ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 ``` 3. To configure SASL authentication, set the SASL mechanism, which in this tutorial is `PLAIN`. Then configure the JAAS configuration property to describe to connect to the Confluent Server brokers. The properties `username` and `password` are used to configure the user for connections. ```bash sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` Combining the configuration steps above, the Kafka client’s general pattern for enabling TLS/SSL encryption and SASL/PLAIN authentication is to add the following to the Kafka client’s properties file. ```bash security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` What differs between Kafka clients is the specific [configuration prefix](../kafka/security_prefixes.md#security-prefixes) that precedes each configuration parameter, as described in the sections below. ### Configure Console Producer and Consumer The command line tools for console producer and consumer are convenient ways to send and receive a small amount of data to the cluster. They are clients and thus need security configurations as well. 1. Create a `client_security.properties` file with the security configuration parameters described above, with no additional configuration prefix. ```bash security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` 2. Pass in the properties file when using the command line tools. ```bash kafka-console-producer --bootstrap-server kafka1:9094 --topic test-topic --producer.config client_security.properties kafka-console-consumer --bootstrap-server kafka1:9094 --topic test-topic --consumer.config client_security.properties ``` ### Configure ksqlDB and stream processing clients Enabling ksqlDB and stream processing clients for security is simply a matter of passing the security configurations to the relevant client constructor. Take the basic Kafka client security configuration: ```bash security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` And configure the application for the following: * Top-level, with no additional configuration prefix Combining these configurations, ksqlDB configuration for TLS/SSL encryption and SASL/PLAIN authentication is the following. You may configure them by either loading the properties from a file, as shown below, or by setting the properties programmatically. ```bash # Top level security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="client-secret"; ``` #### NOTE This is not the full configuration. This is just the additional configurations required to enable security on a known working ksqlDB or Java application that is monitored using Confluent Control Center . ## Configure Kafka Connect From the perspective of the Confluent Server brokers, Kafka Connect is another Kafka client, and this tutorial configures Kafka Connect for TLS/SSL encryption and SASL/PLAIN authentication. Enabling Connect for security is simply a matter of passing the security configurations to the Connect workers, the producers used by source connectors, and the consumers used by sink connectors. Take the basic client security configuration: ```bash security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` And configure Kafka Connect for the following: * Top-level for Connect workers, with no additional configuration prefix * Embedded producer for source connectors, with an additional configuration prefix `producer.` * Embedded consumers for sink connectors, with an additional configuration prefix `consumer.` Combining these configurations, a Kafka Connect worker configuration for TLS/SSL encryption and SASL/PLAIN authentication is the following. You may configure these settings in the `connect-distributed.properties` file. ```bash # Connect worker security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="connect" \ password="connect-secret"; # Embedded producer for source connectors producer.security.protocol=SASL_SSL producer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks producer.ssl.truststore.password=test1234 producer.sasl.mechanism=PLAIN producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="connect" \ password="connect-secret"; # Embedded consumer for sink connectors consumer.security.protocol=SASL_SSL consumer.ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks consumer.ssl.truststore.password=test1234 consumer.sasl.mechanism=PLAIN consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="connect" \ password="connect-secret"; ``` #### NOTE This is not the full Connect worker configuration. This is just the additional configurations required to enable security on a known working Kafka Connect cluster that is already successfully monitored using Confluent Control Center . Pass in the properties file when starting each Connect worker. ```shell connect-distributed etc/kafka/connect-distributed.properties ``` ## Replicator Confluent Replicator is a type of Confluent Platform source connector that replicates data from a source to destination Confluent Platform 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. Take the basic client security configuration: ```bash security.protocol=SASL_SSL ssl.truststore.location=/var/ssl/private/kafka.client.truststore.jks ssl.truststore.password=test1234 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="client" \ password="client-secret"; ``` And configure Replicator for the following: * Top-level Replicator consumer from the origin cluster, with an additional configuration prefix `src.kafka.` Combining the configuration steps described above, the Replicator JSON properties file contains the following configuration settings: ```bash { "name":"replicator", "config":{ .... "src.kafka.security.protocol" : "SASL_SSL", "src.kafka.ssl.truststore.location" : "var/private/ssl/kafka.server.truststore.jks", "src.kafka.ssl.truststore.password" : "test1234", "src.kafka.sasl.mechanism" : "PLAIN", "src.kafka.sasl.jaas.config" : "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"replicator\" password=\"replicator-secret\";", .... } } ``` #### NOTE This is not the full Replicator configuration. Rather, it shows the additional configurations required to enable security on a known working Replicator connector that is already successfully monitored using Confluent Control Center . After Kafka Connect is started, you can add the Confluent Replicator: ```bash curl -X POST -H "Content-Type: application/json" --data @replicator_properties.json http://connect:8083/connectors ``` ## Confluent Metrics Reporter If you are using Confluent Control Center to monitor your deployment, the Confluent Metrics Reporter is a client as well. If the monitoring cluster backing Confluent Control Center is also configured with the same security protocols, then configure the Confluent Metrics Reporter for security in each broker’s `server.properties` file. The configuration prefix is `confluent.metrics.reporter.` and is described [above](#tutorial-metrics-reporter). ## Authorization and ACLs Use [Centralized ACLs](authorization/rbac/authorization-acl-with-mds.md#authorization-acl-with-mds) to add, remove, or list ACLs when using RBAC. The most common tasks for ACL management are adding or removing a principal as a producer or consumer. For example, to add a client called `client-1` as a producer and consumer of a topic called `test-topic`, execute the following: ```none confluent iam acl create --allow --principal User:client-1 --operation write --topic test-topic --kafka-cluster confluent iam acl create --allow --principal User:client-1 --operation read --topic test-topic --kafka-cluster ``` For additional Confluent CLI details, refer to the ACL subcommands in [confluent iam](https://docs.confluent.io/confluent-cli/current/command-reference/iam/index.html). If you are not running RBAC, then refer to [Use Access Control Lists (ACLs) for Authorization in Confluent Platform](authorization/acls/overview.md#kafka-authorization). ## Troubleshoot In cases where the configuration does not work on the first attempt, debugging output is a helpful way to diagnose the cause of the problem: 1. Validate the keys and certificates in the keystores and truststores in the Confluent Server brokers and Kafka clients. ```bash keytool -list -v -keystore /var/ssl/private/kafka.server.keystore.jks ``` 2. Enable Kafka authorization logging by modifying the `etc/kafka/log4j.properties` file. Change the log level to DEBUG, and then restart the Confluent Server brokers. ```bash log4j.logger.kafka.authorizer.logger=DEBUG, authorizerAppender ``` 3. Enable TLS/SSL debug output by using the `javax.net.debug` system property, which requires a restart of the JVM. ```bash export KAFKA_OPTS=-Djavax.net.debug=all ``` 4. Enable SASL debug output using the `sun.security.krb5.debug` system property, which requires a restart of the JVM. ```bash export KAFKA_OPTS=-Dsun.security.krb5.debug=true ``` ## Next steps To see a fully secured multi-node cluster, check out the Docker-based [Confluent Platform demo](../tutorials/cp-demo/index.md#cp-demo). It shows entire configurations, including security-related and non security-related configuration parameters, on all components in Confluent Platform, and the demo’s playbook has a security section for further learning. Read the [documentation](overview.md#security) for more details about security design and configuration on all components in Confluent Platform. While this tutorial uses the PLAIN mechanism for the SASL examples, Confluent additionally supports [GSSAPI (Kerberos)](authentication/sasl/gssapi/overview.md#kafka-sasl-auth-gssapi) and [SCRAM](authentication/sasl/scram/overview.md#kafka-sasl-auth-scram), which are more suitable for production. We welcome feedback in the [Confluent community](https://launchpass.com/confluentcommunity) security channel in Slack!