<commit_before><!--
 Licensed to the Apache Software Foundation (ASF) under one or more
 contributor license agreements.  See the NOTICE file distributed with
 this work for additional information regarding copyright ownership.
 The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-->

<h3><a id="security_overview">7.1 Security Overview</a></h3>
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</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><a id="security_ssl">7.2 Encryption and Authentication using SSL</a></h3>

<h3><a id="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 (<a href="https://help.ubuntu.com/community/Kerberos">Ubuntu</a>, <a href="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>
            <code>sudo /usr/sbin/kadmin.local -q 'addprinc -randkey kafka/hostname@domainname'<br>
                sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/kafka.keytab kafka/hostname@domainname"</code></li>
        <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><b><a name="jaas_config_file">Creating JAAS Config File</a></b><br>
            Each node in the cluster should have a JAAS file similar to the example below. Add this file to kafka/config dir:
        <pre>
            KafkaServer {
                com.sun.security.auth.module.Krb5LoginModule required
                useKeyTab=true
                storeKey=true
                serviceName="kafka"
                keyTab="/etc/security/keytabs/kafka1.keytab"
                principal="kafka/kafka1.hostname.com@DOMAIN.COM";
            };

            Client {
               com.sun.security.auth.module.Krb5LoginModule required
               useKeyTab=true
               storeKey=true
               serviceName="zookeeper"
               keyTab="/etc/security/keytabs/kafka1.keytab"
               principal="kafka/kafka1.hostname.com@DOMAIN.COM";
            };

            KafkaClient {
               com.sun.security.auth.module.Krb5LoginModule required
               useTicketCache=true
               serviceName="kafka";
            };
        </pre>
            <u>Important notes:</u>
            <ol>
                <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 important to have the same principal name across all the brokers.</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><a name="jaas_client">Creating Client Side JAAS Config</a></b><br>
        Clients (producers, consumers, copycat 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:
            <pre>
                KafkaClient {
                    com.sun.security.auth.module.Krb5LoginModule required
                    useKeyTab=true
                    storeKey=true
                    serviceName="kafka"
                    keyTab="/etc/security/keytabs/kafka1.keytab"
                    principal="kafkaproducer/hostname@DOMAIN.COM";
                };
            </pre>
        </li>
    </ol></li>
    <li><h4>Configuring Kafka Brokers</h4>
    <ol>
        <li>Pass the name of the jaas file you created in <a href="#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:
            <pre>listeners=SASL_PLAINTEXT://host.name:port</pre>
        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:
        <pre>security.inter.broker.protocol=SASL_PLAINTEXT</pre></li>

    </ol>
    </li>
    <li><h4>Configuring Kafka Clients</h4>
        SASL authentication is only supported for new kafka producer and consumer, the older API is not supported.>br>
        To configure SASL authentication on the clients:
        <ol>
            <li>pass the name of the jaas file you created in <a href="#jaas_client">Creating Client Side JAAS Config"</a> as a JVM parameter to the client JVM:
        <pre>-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf</pre></li>
            <li>Make sure the keytabs configured in the kafka_client_jaas.conf are readable by the linux user who is starting kafka client.</li>
            <li>Configure the following property in producer.properties or consumer.properties:
                <pre>security.protocol=SASL_PLAINTEXT</pre></li>
        </ol></li>
</ol>

<h3><a id="security_authz">7.4 Authorization and ACLs</a></h3><commit_msg>KAFKA-2724: ZK Auth documentation.<commit_after><!--
 Licensed to the Apache Software Foundation (ASF) under one or more
 contributor license agreements.  See the NOTICE file distributed with
 this work for additional information regarding copyright ownership.
 The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-->

<h3><a id="security_overview">7.1 Security Overview</a></h3>
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</li>
    <li>Authenticate brokers connecting to ZooKeeper</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><a id="security_ssl">7.2 Encryption and Authentication using SSL</a></h3>

<h3><a id="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 (<a href="https://help.ubuntu.com/community/Kerberos">Ubuntu</a>, <a href="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>
            <code>sudo /usr/sbin/kadmin.local -q 'addprinc -randkey kafka/hostname@domainname'<br>
                sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/kafka.keytab kafka/hostname@domainname"</code></li>
        <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><b><a name="jaas_config_file">Creating JAAS Config File</a></b><br>
            Each node in the cluster should have a JAAS file similar to the example below. Add this file to kafka/config dir:
        <pre>
            KafkaServer {
                com.sun.security.auth.module.Krb5LoginModule required
                useKeyTab=true
                storeKey=true
                serviceName="kafka"
                keyTab="/etc/security/keytabs/kafka1.keytab"
                principal="kafka/kafka1.hostname.com@DOMAIN.COM";
            };

            Client {
               com.sun.security.auth.module.Krb5LoginModule required
               useKeyTab=true
               storeKey=true
               serviceName="zookeeper"
               keyTab="/etc/security/keytabs/kafka1.keytab"
               principal="kafka/kafka1.hostname.com@DOMAIN.COM";
            };

            KafkaClient {
               com.sun.security.auth.module.Krb5LoginModule required
               useTicketCache=true
               serviceName="kafka";
            };
        </pre>
            <u>Important notes:</u>
            <ol>
                <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><a name="jaas_client">Creating Client Side JAAS Config</a></b><br>
        Clients (producers, consumers, copycat 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:
            <pre>
                KafkaClient {
                    com.sun.security.auth.module.Krb5LoginModule required
                    useKeyTab=true
                    storeKey=true
                    serviceName="kafka"
                    keyTab="/etc/security/keytabs/kafka1.keytab"
                    principal="kafkaproducer/hostname@DOMAIN.COM";
                };
            </pre>
        </li>
    </ol></li>
    <li><h4>Configuring Kafka Brokers</h4>
    <ol>
        <li>Pass the name of the jaas file you created in <a href="#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:
            <pre>listeners=SASL_PLAINTEXT://host.name:port</pre>
        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:
        <pre>security.inter.broker.protocol=SASL_PLAINTEXT</pre></li>

    </ol>
    </li>
    <li><h4>Configuring Kafka Clients</h4>
        SASL authentication is only supported for new kafka producer and consumer, the older API is not supported.>br>
        To configure SASL authentication on the clients:
        <ol>
            <li>pass the name of the jaas file you created in <a href="#jaas_client">Creating Client Side JAAS Config"</a> as a JVM parameter to the client JVM:
        <pre>-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf</pre></li>
            <li>Make sure the keytabs configured in the kafka_client_jaas.conf are readable by the linux user who is starting kafka client.</li>
            <li>Configure the following property in producer.properties or consumer.properties:
                <pre>security.protocol=SASL_PLAINTEXT</pre></li>
        </ol></li>
</ol>

<h3><a id="security_authz">7.4 Authorization and ACLs</a></h3>
<h3><a id="zk_authz">7.5 ZooKeeper Authentication</a></h3>
<h4><a id="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.

<h4><a id="zk_authz_migration">7.5.2 Migrating clusters</a></h4>
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:
<pre>
./bin/zookeeper-security-migration --zookeeper.acl=secure --zookeeper.connection=localhost:2181
</pre>
<p>Run this to see the full list of parameters:</p>
<pre>
./bin/zookeeper-security-migration --help
</pre>
<h4><a id="zk_authz_new">7.5.3 Migrating the ZooKeeper ensemble</a></h4>
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:
<ol>
	<li><a href="http://zookeeper.apache.org/doc/r3.4.6/zookeeperProgrammers.html#sc_ZooKeeperAccessControl">Apache ZooKeeper documentation</a></li>
	<li><a href="https://cwiki.apache.org/confluence/display/ZOOKEEPER/Zookeeper+and+SASL">Apache ZooKeeper wiki</a></li>
	<li><a href="http://www.cloudera.com/content/www/en-us/documentation/cdh/5-1-x/CDH5-Security-Guide/cdh5sg_zookeeper_security.html">Cloudera ZooKeeper security configuration</a></li>
</ol>
{'fraction_non_alphanumeric': 0.08502121913580248, 'fraction_numerical': 0.0055941358024691355, 'mean_word_length': 3.2765518663641986, 'pattern_counts': {'":': 0, '<': 347, '<?xml version=': 0, '>': 351, 'https://': 5, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1829856', 'n_tokens_mistral': 5924, 'n_tokens_neox': 5673, 'n_words': 2277}