In release 0.9.0.0, the Kafka community added a number of features that, used either separately or together, increases security in a Kafka cluster. The following security measures are currently supported:
<ol>
<li>Authenticating clients (Producers and consumers) connections to brokers, using either SSL or SASL (Kerberos)</li>
<li>Authorizing read / write operations by clients</li>
<li>Encryption of data sent between brokers and clients, or between brokers, using SSL (Note there is performance degradation in the clients when SSL is enabled. The magnitude of the degradation depends on the CPU type.)</li>
<li>Security is optional - non-secured clusters are supported, as well as a mix of authenticated, unauthenticated, encrypted and non-encrypted clients.</li>
<li>Authorization is pluggable and supports integration with external authorization services</li>
</ol>
The guides below explain how to configure and use the security features in both clients and brokers.
<h3><aid="security_ssl">7.2 Encryption and Authentication using SSL</a></h3>
<li><h4>Generate SSL key and certificate for each Kafka broker</h4>
The first step of deploying HTTPS is to generate the key and the certificate for each machine in the cluster. You can use Java’s keytool utility to accomplish this task.
We will generate the key into a temporary keystore initially so that we can export and sign it later with CA.
You need to specify two parameters in the above command:
<ol>
<li>keystore: the keystore file that stores the certificate. The keystore file contains the private key of the certificate; therefore, it needs to be kept safely.</li>
<li>validity: the valid time of the certificate in days.</li>
</ol>
Ensure that common name (CN) matches exactly with 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 the malicious one.</li>
<li><h4>Creating your own CA</h4>
After the first step, 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.<p>
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 likes a government that issues passports—the government stamps (signs) each passport so that the passport becomes difficult to forge. Other governments verify the stamps 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.
<b>Note:</b> If you configure Kafka brokers to require client authentication by setting ssl.client.auth to be "requested" or "required" on <ahref="#config_broker">Kafka broker config</a> then you must provide a truststore for Kafka broker as well and it should have all the CA certificates that clients keys signed by.
In contrast to the keystore in step 1 that stores each machine’s own identity, the truststore of a client stores all the certificates that the client should trust. Importing a certificate into one’s truststore also means that trusting all certificates that are signed by that certificate. As the analogy above, trusting the government (CA) also means that trusting all passports (certificates) that it has issued. This attribute is called the chains of trust, and it is particularly useful when deploying SSL on a large Kafka 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.</li>
<li><h4>Signing the certificate</h4>
The next step is to sign all certificates generated by step 1 with the CA generated in step 2. First, you need to export the certificate from the keystore:
The definitions of the parameters are the following:
<ol>
<li>keystore: the location of the keystore</li>
<li>ca-cert: the certificate of the CA</li>
<li>ca-key: the private key of the CA</li>
<li>ca-password: the passphrase of the CA</li>
<li>cert-file: the exported, unsigned certificate of the server</li>
<li>cert-signed: the signed certificate of the server</li>
</ol>
Here is an example of a bash script with all above steps. Note that one of the commands assumes a password of `test1234`, so either use that password or edit the command before running it.
<li>ssl.client.auth = none ("required" => client authentication is required, "requested" => client authentication is requested and client without certs can still connect when this option chosen")</li>
<li>ssl.cipher.suites = A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. (Default is an empty list)</li>
<li>ssl.enabled.protocols = TLSv1.2,TLSv1.1,TLSv1 (list out the SSL protocols that you are going to accept from clients. Do note SSL is deprecated and using that in production is not recommended)</li>
<li> ssl.keystore.type = JKS</li>
<li>ssl.truststore.type = JKS</li>
</ol>
If you want to enable SSL for inter-broker communication, add the following to the broker properties file (it defaults to PLAINTEXT)
<pre>security.inter.broker.protocol = SSL</pre>
If you want to enable any cipher suites other than the defaults that comes with JVM like the ones listed here:
<ahref="https://docs.oracle.com/javase/7/docs/technotes/guides/security/SunProviders.html">https://docs.oracle.com/javase/7/docs/technotes/guides/security/SunProviders.html</a> you will need to install <b><ahref="http://www.oracle.com/technetwork/java/javase/downloads/jce-7-download-432124.html">Unlimited Strength Policy files</a></b><br>
Once you start the broker you should be able to see in the server.log
If the certificate does not show up or if there are any other error messages than your keystore is not setup properly.</li>
<li><h4>Configuring Kafka Clients</h4>h4>
SSL is supported only for new Kafka Producer & Consumer, the older API is not supported. The configs for SSL will be same for both producer & consumer.<br>
If client authentication is not required in the broker, then the following is a minimal configuration example:
Other configuration settings that may also be needed depending on our requirements and the broker configuration:\
<ol>
<li>ssl.provider (Optional). The name of the security provider used for SSL connections. Default value is the default security provider of the JVM.</li>
<li>ssl.cipher.suites (Optional). A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol.</li>
<li>ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 **Should list at least one of the protocols configured on the broker side</li>
<li>ssl.truststore.type = "JKS"</li>
<li>ssl.keystore.type = "JKS"</li>
</ol>
<br>
Examples using console-producer and console-consumer:
<h3><aid="security_sasl">7.3 Authentication using SASL</a></h3>
<ol>
<li><h4>Prerequisites</h4><br>
<ol>
<li><b>Kerberos</b><br>
If your organization is already using a Kerberos server (for example, by using Active Directory), there is no need to install a new server just for Kafka. Otherwise you will need to install one, your Linux vendor likely has packages for Kerberos and a short guide on how to install and configure it (<ahref="https://help.ubuntu.com/community/Kerberos">Ubuntu</a>, <ahref="https://access.redhat.com/documentation/en-US/Red_Hat_Enterprise_Linux/6/html/Managing_Smart_Cards/installing-kerberos.html">Redhat</a>). Note that if you are using Oracle Java, you will need to download JCE policy files for your Java version and copy them to $JAVA_HOME/jre/lib/security.</li>
<li><b>Create Kerberos Principals</b><br>
If you are using the organization's Kerberos or Active Directory server, ask your Kerberos administrator for a principal for each Kafka broker in your cluster and for every Linux user that will access Kafka with Kerberos authentication.</br>
If you installed your own Kerberos, you will need to create these principals yourself:</br>
<li><b>Make sure all hosts can be reachable using hostnames</b> - It is important in case of kerberos all your hosts can be resolved with their FQDNs.</li>
<li>KafkaServer is a section name in JAAS file used by KafkaServer/Broker. This section tells Kafka Server which principal to use and which keytab this principal is stored. It allows Kafka Server to login using the keytab specified in this section.</li>
<li>Client section is used to authenticate a SASL connection with zookeeper. It also allows a broker to set SASL ACL on zookeeper nodes which locks these nodes down so that only kafka broker can modify. It is necessary to have the same principal name across all the brokers. If you want to use a section name other than Client, then you need to set the system property <tt>zookeeper.sasl.client</tt> to the appropriate name (<i>e.g.</i>, <tt>-Dzookeeper.sasl.client=ZkClient</tt>).</li>
<li>KafkaClient section here describes how the clients like producer and consumer can connect to the Kafka Broker. Here we specified "useTicketCache=true" not a keytab this allows user to do kinit and run a kafka-console-consumer or kafka-console-producer to connect to broker. For a long running process one should create KafkaClient section similar to KafkaServer.</li>
<li>In KafkaServer and KafkaClient sections we've "serviceName" this should match principal name with which kafka broker is running. In the above example principal="kafka/kafka1.hostname.com@DOMAIN.com" so we've "kafka" which is matching the principalName.</li>
</ol>
</li>
<li><b><aname="jaas_client">Creating Client Side JAAS Config</a></b><br>
Clients (producers, consumers, connect workers, etc) will authenticate to the cluster with their own principal (usually with the same name as the user used for running the client), so obtain or create these principals as needed. Then create a JAAS file as follows:
<li>Pass the name of the jaas file you created in <ahref="#jaas_config_file">Creating JAAS Config File"</a> as a JVM parameter to the kafka broker: <pre>-Djava.security.auth.login.config=/etc/kafka/kafka_jaas.conf</pre></li>
<li>Make sure the keytabs configured in the kafka_jaas.conf are readable by the linux user who is starting kafka broker.</li>
<li>Configure a SASL port in server.properties, by adding the following to the <i>listeners</i> parameter, which contains one or more comma-separated values:
If you are only configuring SASL port (or if you are very paranoid and want the Kafka brokers to authenticate each other using SASL) then make sure you set same SASL protocol for inter-broker communication:
Kafka ships with a pluggable Authorizer and an out-of-box authorizer implementation that uses zookeeper to store all the acls. Kafka acls are defined in the general format of "Principal P is [Allowed/Denied] Operation O From Host H On Resource R". You can read more about the acl structure on KIP-11. In order to add, remove or list acls you can use the Kafka authorizer CLI. By default, if a Resource R has no associated acls, no one other than super users is allowed to access R. If you want change that behavior, you can include the following in broker.properties.
Kafka Authorization management CLI can be found under bin directory with all the other CLIs. The CLI script is called <b>kafka-acls.sh</b>. Following lists all the options that the script supports:
<p></p>
<tableclass="data-table">
<tr>
<th>Option</th>
<th>Description</th>
<th>Default</th>
<th>Option type</th>
</tr>
<tr>
<td>--add</td>
<td>Indicates to the script that user is trying to add an acl.</td>
<td></td>
<td>Action</td>
</tr>
<tr>
<td>--remove</td>
<td>Indicates to the script that user is trying to remove an acl.</td>
<td></td>
<td>Action</td>
</tr>
<tr>
<td>--list</td>
<td>Indicates to the script that user is trying to list acls.</td>
<td></td>
<td>Action</td>
</tr>
<tr>
<td>--authorizer</td>
<td>Fully qualified class name of the authorizer.</td>
<td>kafka.security.auth.SimpleAclAuthorizer</td>
<td>Configuration</td>
</tr>
<tr>
<td>--authorizer-properties</td>
<td>comma separated key=val pairs that will be passed to authorizer for initialization.</td>
<td></td>
<td>Configuration</td>
</tr>
<tr>
<td>--cluster</td>
<td>Specifies cluster as resource.</td>
<td></td>
<td>Resource</td>
</tr>
<tr>
<td>--topic [topic-name]</td>
<td>Specifies the topic as resource.</td>
<td></td>
<td>Resource</td>
</tr>
<tr>
<td>--consumer-group [group-name]</td>
<td>Specifies the consumer-group as resource.</td>
<td></td>
<td>Resource</td>
</tr>
<tr>
<td>--allow-principal</td>
<td>Principal is in PrincipalType:name format that will be added to ACL with Allow permission. <br>You can specify multiple --allow-principal in a single command.</td>
<td></td>
<td>Principal</td>
</tr>
<tr>
<td>--deny-principal</td>
<td>Principal is in PrincipalType:name format that will be added to ACL with Deny permission. <br>You can specify multiple --deny-principal in a single command.</td>
<td></td>
<td>Principal</td>
</tr>
<tr>
<td>--allow-hosts</td>
<td>Comma separated list of hosts from which principals listed in --allow-principals will have access.</td>
<td> if --allow-principals is specified defaults to * which translates to "all hosts"</td>
<td>Host</td>
</tr>
<tr>
<td>--deny-hosts</td>
<td>Comma separated list of hosts from which principals listed in --deny-principals will be denied access.</td>
<td>if --deny-principals is specified defaults to * which translates to "all hosts"</td>
<td> Convenience option to add/remove acls for producer role. This will generate acls that allows WRITE,
DESCRIBE on topic and CREATE on cluster.</td>
<td></td>
<td>Convenience</td>
</tr>
<tr>
<td>--consumer</td>
<td> Convenience option to add/remove acls for consumer role. This will generate acls that allows READ,
DESCRIBE on topic and READ on consumer-group.</td>
<td>Convenience</td>
</tr>
</tbody></table>
<h4>Examples</h4>
<ul>
<li><b>Adding Acls</b><br>
Suppose you want to add an acl "Principals User:Bob and User:Alice are allowed to perform Operation Read and Write on Topic Test-Topic from Host1 and Host2". You can do that by executing the CLI with following options:
By default all principals that don't have an explicit acl that allows access for an operation to a resource are denied. In rare cases where an allow acl is defined that allows access to all but some principal we will have to use the --deny-principals and --deny-host option. For example, if we want to allow all users to Read from Test-topic but only deny User:BadBob from host bad-host we can do so using following commands:
Above examples add acls to a topic by specifying --topic [topic-name] as the resource option. Similarly user can add acls to cluster by specifying --cluster and to a consumer group by specifying --consumer-group [group-name].</li>
<li><b>Removing Acls</b><br>
Removing acls is pretty much the same. The only difference is instead of --add option users will have to specify --remove option. To remove the acls added by the first example above we can execute the CLI with following options:
We can list acls for any resource by specifying the --list option with the resource. To list all acls for Test-topic we can execute the CLI with following options:
<li><b>Adding or removing a principal as producer or consumer</b><br>
The most common use case for acl management are adding/removing a principal as producer or consumer so we added convenience options to handle these cases. In order to add User:Bob as a producer of Test-topic we can execute the following command:
<h4><aid="zk_authz_new">7.5.1 New clusters</a></h4>
To enable ZooKeeper authentication on brokers, there are two necessary steps:
<ol>
<li> Create a JAAS login file and set the appropriate system property to point to it as described above</li>
<li> Set the configuration property <tt>zookeeper.set.acl</tt> in each broker to true</li>
</ol>
The metadata stored in ZooKeeper is such that only brokers will be able to modify the corresponding znodes, but znodes are world readable. The rationale behind this decision is that the data stored in ZooKeeper is not sensitive, but inappropriate manipulation of znodes can cause cluster disruption.
If you are running a version of Kafka that does not support security of simply with security disabled, and you want to make the cluster secure, then you need to execute the following steps to enable ZooKeeper authentication with minimal disruption to your operations:
<ol>
<li>Perform a rolling restart setting the JAAS login file, which enables brokers to authenticate. At the end of the rolling restart, brokers are able to manipulate znodes with strict ACLs, but they will not create znodes with those ACLs</li>
<li>Perform a second rolling restart of brokers, this time setting the configuration parameter <tt>zookeeper.set.acl</tt> to true, which enables the use of secure ACLs when creating znodes</li>
<li>Execute the ZkSecurityMigrator tool. To execute the tool, there is this script: <tt>./bin/zookeeper-security-migration.sh</tt> with <tt>zookeeper.acl</tt> set to secure. This tool traverses the corresponding sub-trees changing the ACLs of the znodes</li>
</ol>
<p>It is also possible to turn off authentication in a secure cluster. To do it, follow these steps:</p>
<ol>
<li>Perform a rolling restart of brokers setting the JAAS login file, which enables brokers to authenticate, but setting <tt>zookeeper.set.acl</tt> 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</li>
<li>Execute the ZkSecurityMigrator tool. To execute the tool, run this script <tt>./bin/zookeeper-security-migration.sh</tt> with <tt>zookeeper.acl</tt> set to unsecure. This tool traverses the corresponding sub-trees changing the ACLs of the znodes</li>
<li>Perform a second rolling restart of brokers, this time omitting the system property that sets the JAAS login file</li>
</ol>
Here is an example of how to run the migration tool:
It is also necessary to enable authentication on the ZooKeeper ensemble. To do it, we need to perform a rolling restart of the server and set a few properties. Please refer to the ZooKeeper documentation for more detail:
