22/security.html [18:895]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
<script id="security-template" type="text/x-handlebars-template">
    <h3><a id="security_overview" href="#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>Authentication of connections to brokers from clients (producers and consumers), other brokers and tools, using either SSL or SASL. Kafka supports the following SASL mechanisms:
        <ul>
            <li>SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0</li>
            <li>SASL/PLAIN - starting at version 0.10.0.0</li>
            <li>SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512 - starting at version 0.10.2.0</li>
            <li>SASL/OAUTHBEARER - starting at version 2.0</li>
        </ul></li>
        <li>Authentication of connections from brokers to ZooKeeper</li>
        <li>Encryption of data transferred between brokers and clients, between brokers, or between brokers and tools using SSL (Note that there is a performance degradation when SSL is enabled, the magnitude of which depends on the CPU type and the JVM implementation.)</li>
        <li>Authorization of read / write operations by clients</li>
        <li>Authorization is pluggable and integration with external authorization services is supported</li>
    </ol>

    It's worth noting that security is optional - non-secured clusters are supported, as well as a mix of authenticated, unauthenticated, encrypted and non-encrypted clients.

    The guides below explain how to configure and use the security features in both clients and brokers.

    <h3><a id="security_ssl" href="#security_ssl">7.2 Encryption and Authentication using SSL</a></h3>
    Apache Kafka allows clients to connect over SSL. By default, SSL is disabled but can be turned on as needed.

    <ol>
        <li><h4><a id="security_ssl_key" href="#security_ssl_key">Generate SSL key and certificate for each Kafka broker</a></h4>
            The first step of deploying one or more brokers with the SSL support 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.
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA</code></pre>

            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>
            <br>

        <h5><a id="security_confighostname" href="#security_confighostname">Configuring Host Name Verification</a></h5>
        From Kafka version 2.0.0 onwards, host name verification of servers is enabled by default for client connections
        as well as inter-broker connections to prevent man-in-the-middle attacks. Server host name verification may be disabled
        by setting <code>ssl.endpoint.identification.algorithm</code> to an empty string. For example,
        <pre class="line-numbers"><code class="language-text">	ssl.endpoint.identification.algorithm=</code></pre>
        For dynamically configured broker listeners, hostname verification may be disabled using <code>kafka-configs.sh</code>.
        For example,
        <pre class="line-numbers"><code class="language-text">        bin/kafka-configs.sh --bootstrap-server localhost:9093 --entity-type brokers --entity-name 0 --alter --add-config "listener.name.internal.ssl.endpoint.identification.algorithm="
        </code></pre>

        For older versions of Kafka, <code>ssl.endpoint.identification.algorithm</code> is not defined by default, so host name
        verification is not performed. The property should be set to <code>HTTPS</code> to enable host name verification.

        <pre class="line-numbers"><code class="language-text">	ssl.endpoint.identification.algorithm=HTTPS </code></pre>

        Host name verification must be enabled to prevent man-in-the-middle attacks if server endpoints are not validated
        externally.

        <h5><a id="security_configcerthostname" href="#security_configcerthstname">Configuring Host Name In Certificates</a></h5>
        If host name verification is enabled, clients will verify the server's fully qualified domain name (FQDN) against one of
        the following two fields:
        <ol>
            <li>Common Name (CN)
            <li>Subject Alternative Name (SAN)
        </ol>
        <br>
        Both fields are valid, RFC-2818 recommends the use of SAN however. 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 following argument <code> -ext SAN=DNS:{FQDN} </code> to the keytool command:
        <pre class="line-numbers"><code class="language-bash">        keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA -ext SAN=DNS:{FQDN}
        </code></pre>
        The following command can be run afterwards to verify the contents of the generated certificate:
        <pre class="line-numbers"><code class="language-bash">        keytool -list -v -keystore server.keystore.jks
        </code></pre>
        </li>
        <li><h4><a id="security_ssl_ca" href="#security_ssl_ca">Creating your own CA</a></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.
            <pre class="line-numbers"><code class="language-bash">            openssl req -new -x509 -keyout ca-key -out ca-cert -days 365</code></pre>

            The generated CA is simply a public-private key pair and certificate, and it is intended to sign other certificates.<br>

            The next step is to add the generated CA to the **clients' truststore** so that the clients can trust this CA:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert</code></pre>

            <b>Note:</b> If you configure the Kafka brokers to require client authentication by setting ssl.client.auth to be "requested" or "required" on the <a href="#config_broker">Kafka brokers config</a> then you must provide a truststore for the Kafka brokers as well and it should have all the CA certificates that clients' keys were signed by.
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert</code></pre>

            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 trusting all certificates that are signed by that certificate. As the analogy above, trusting the government (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 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><a id="security_ssl_signing" href="#security_ssl_signing">Signing the certificate</a></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:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file</code></pre>

            Then sign it with the CA:
            <pre class="line-numbers"><code class="language-bash">            openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days {validity} -CAcreateserial -passin pass:{ca-password}</code></pre>

            Finally, you need to import both the certificate of the CA and the signed certificate into the keystore:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
            keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed</code></pre>

            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.
                <pre>
            #!/bin/bash
            #Step 1
            keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
            #Step 2
            openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
            keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
            keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
            #Step 3
            keytool -keystore 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 365 -CAcreateserial -passin pass:test1234
            keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
            keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed</code></pre></li>
        <li><h4><a id="security_configbroker" href="#security_configbroker">Configuring Kafka Brokers</a></h4>
            Kafka Brokers support listening for connections on multiple ports.
            We need to configure the following property in server.properties, which must have one or more comma-separated values:
            <pre>listeners</code></pre>

            If SSL is not enabled for inter-broker communication (see below for how to enable it), both PLAINTEXT and SSL ports will be necessary.
            <pre class="line-numbers"><code class="language-text">            listeners=PLAINTEXT://host.name:port,SSL://host.name:port</code></pre>

            Following SSL configs are needed on the broker side
            <pre class="line-numbers"><code class="language-text">            ssl.keystore.location=/var/private/ssl/server.keystore.jks
            ssl.keystore.password=test1234
            ssl.key.password=test1234
            ssl.truststore.location=/var/private/ssl/server.truststore.jks
            ssl.truststore.password=test1234</code></pre>

            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.

            Optional settings that are worth considering:
            <ol>
                <li>ssl.client.auth=none ("required" => client authentication is required, "requested" => client authentication is requested and client without certs can still connect. The usage of "requested" is discouraged as it provides a false sense of security and misconfigured clients will still connect successfully.)</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. (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 that SSL is deprecated in favor of TLS and using SSL in production is not recommended)</li>
                <li>ssl.keystore.type=JKS</li>
                <li>ssl.truststore.type=JKS</li>
                <li>ssl.secure.random.implementation=SHA1PRNG</li>
            </ol>
            If you want to enable SSL for inter-broker communication, add the following to the server.properties file (it defaults to PLAINTEXT)
            <pre>
            security.inter.broker.protocol=SSL</code></pre>

            <p>
            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 <a href="http://www.oracle.com/technetwork/java/javase/downloads/index.html">JCE Unlimited Strength Jurisdiction Policy Files</a> must be obtained and installed in the JDK/JRE. See the
            <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/SunProviders.html">JCA Providers Documentation</a> for more information.
            </p>

            <p>
            The JRE/JDK will have a default pseudo-random number generator (PRNG) that is used for cryptography operations, so it is not required to configure the
            implementation used with the <pre>ssl.secure.random.implementation</code></pre>. However, there are performance issues with some implementations (notably, the
            default chosen on Linux systems, <pre>NativePRNG</code></pre>, utilizes a global lock). In cases where performance of SSL connections becomes an issue,
            consider explicitly setting the implementation to be used. The <pre>SHA1PRNG</code></pre> implementation is non-blocking, and has shown very good performance
            characteristics under heavy load (50 MB/sec of produced messages, plus replication traffic, per-broker).
            </p>

            Once you start the broker you should be able to see in the server.log
            <pre>
            with addresses: PLAINTEXT -> EndPoint(192.168.64.1,9092,PLAINTEXT),SSL -> EndPoint(192.168.64.1,9093,SSL)</code></pre>

            To check quickly if  the server keystore and truststore are setup properly you can run the following command
            <pre>openssl s_client -debug -connect localhost:9093 -tls1</code></pre> (Note: TLSv1 should be listed under ssl.enabled.protocols)<br>
            In the output of this command you should see server's certificate:
            <pre>
            -----BEGIN CERTIFICATE-----
            {variable sized random bytes}
            -----END CERTIFICATE-----
            subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Sriharsha Chintalapani
            issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com</code></pre>
            If the certificate does not show up or if there are any other error messages then your keystore is not setup properly.</li>

        <li><h4><a id="security_configclients" href="#security_configclients">Configuring Kafka Clients</a></h4>
            SSL is supported only for the new Kafka Producer and Consumer, the older API is not supported. The configs for SSL will be the same for both producer and consumer.<br>
            If client authentication is not required in the broker, then the following is a minimal configuration example:
            <pre class="line-numbers"><code class="language-text">            security.protocol=SSL
            ssl.truststore.location=/var/private/ssl/client.truststore.jks
            ssl.truststore.password=test1234</code></pre>

            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.

            If client authentication is required, then a keystore must be created like in step 1 and the following must also be configured:
            <pre class="line-numbers"><code class="language-text">            ssl.keystore.location=/var/private/ssl/client.keystore.jks
            ssl.keystore.password=test1234
            ssl.key.password=test1234</code></pre>

            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. It 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:
            <pre class="line-numbers"><code class="language-bash">            kafka-console-producer.sh --broker-list localhost:9093 --topic test --producer.config client-ssl.properties
            kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config client-ssl.properties</code></pre>
        </li>
    </ol>
    <h3><a id="security_sasl" href="#security_sasl">7.3 Authentication using SASL</a></h3>

    <ol>
    <li><h4><a id="security_sasl_jaasconfig" href="#security_sasl_jaasconfig">JAAS configuration</a></h4>
    <p>Kafka uses the Java Authentication and Authorization Service
    (<a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jaas/JAASRefGuide.html">JAAS</a>)
    for SASL configuration.</p>
        <ol>
        <li><h5><a id="security_jaas_broker"
            href="#security_jaas_broker">JAAS configuration for Kafka brokers</a></h5>

            <p><tt>KafkaServer</tt> is the section name in the JAAS file used by each
            KafkaServer/Broker. This section provides SASL configuration options
            for the broker including any SASL client connections made by the broker
            for inter-broker communication. If multiple listeners are configured to use
            SASL, the section name may be prefixed with the listener name in lower-case
            followed by a period, e.g. <tt>sasl_ssl.KafkaServer</tt>.</p>

            <p><tt>Client</tt> section is used to authenticate a SASL connection with
            zookeeper. It also allows the brokers to set SASL ACL on zookeeper
            nodes which locks these nodes down so that only the brokers can
            modify it. It is necessary to have the same principal name across all
            brokers. If you want to use a section name other than Client, set the
            system property <tt>zookeeper.sasl.clientconfig</tt> to the appropriate
            name (<i>e.g.</i>, <tt>-Dzookeeper.sasl.clientconfig=ZkClient</tt>).</p>

            <p>ZooKeeper uses "zookeeper" as the service name by default. If you
            want to change this, set the system property
            <tt>zookeeper.sasl.client.username</tt> to the appropriate name
            (<i>e.g.</i>, <tt>-Dzookeeper.sasl.client.username=zk</tt>).</p></li>

            <p>Brokers may also configure JAAS using the broker configuration property <code>sasl.jaas.config</code>.
            The property name must be prefixed with the listener prefix including the SASL mechanism,
            i.e. <code>listener.name.{listenerName}.{saslMechanism}.sasl.jaas.config</code>. Only one
            login module may be specified in the config value. If multiple mechanisms are configured on a
            listener, configs must be provided for each mechanism using the listener and mechanism prefix.
            For example,
                    <pre class="line-numbers"><code class="language-text">        listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
            username="admin" \
            password="admin-secret";
        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_alice="alice-secret";</code></pre>

            If JAAS configuration is defined at different levels, the order of precedence used is:
            <ul>
              <li>Broker configuration property <code>listener.name.{listenerName}.{saslMechanism}.sasl.jaas.config</code></li>
              <li><code>{listenerName}.KafkaServer</code> section of static JAAS configuration</code></li>
              <li><code>KafkaServer</code> section of static JAAS configuration</code></li>
            </ul>
            Note that ZooKeeper JAAS config may only be configured using static JAAS configuration.

            <p>See <a href="#security_sasl_kerberos_brokerconfig">GSSAPI (Kerberos)</a>,
            <a href="#security_sasl_plain_brokerconfig">PLAIN</a>,
            <a href="#security_sasl_scram_brokerconfig">SCRAM</a> or
            <a href="#security_sasl_oauthbearer_brokerconfig">OAUTHBEARER</a> for example broker configurations.</p></li>


        <li><h5><a id="security_jaas_client"
            href="#security_jaas_client">JAAS configuration for Kafka clients</a></h5>

            <p>Clients may configure JAAS using the client configuration property
            <a href="#security_client_dynamicjaas">sasl.jaas.config</a>
            or using the <a href="#security_client_staticjaas">static JAAS config file</a>
            similar to brokers.</p>

            <ol>
            <li><h6><a id="security_client_dynamicjaas"
                href="#security_client_dynamicjaas">JAAS configuration using client configuration property</a></h6>
                <p>Clients may specify JAAS configuration as a producer or consumer property without
                creating a physical configuration file. This mode also enables different producers
                and consumers within the same JVM to use different credentials by specifying
                different properties for each client. If both static JAAS configuration system property
                <code>java.security.auth.login.config</code> and client property <code>sasl.jaas.config</code>
                are specified, the client property will be used.</p>

                <p>See <a href="#security_sasl_kerberos_clientconfig">GSSAPI (Kerberos)</a>,
                <a href="#security_sasl_plain_clientconfig">PLAIN</a>,
                <a href="#security_sasl_scram_clientconfig">SCRAM</a> or
                <a href="#security_sasl_oauthbearer_clientconfig">OAUTHBEARER</a> for example configurations.</p></li>

                <li><h6><a id="security_client_staticjaas" href="#security_client_staticjaas">JAAS configuration using static config file</a></h6>
                To configure SASL authentication on the clients using static JAAS config file:
                <ol>
                <li>Add a JAAS config file with a client login section named <tt>KafkaClient</tt>. Configure
                    a login module in <tt>KafkaClient</tt> for the selected mechanism as described in the examples
                    for setting up <a href="#security_sasl_kerberos_clientconfig">GSSAPI (Kerberos)</a>,
                    <a href="#security_sasl_plain_clientconfig">PLAIN</a>,
                    <a href="#security_sasl_scram_clientconfig">SCRAM</a> or
                    <a href="#security_sasl_oauthbearer_clientconfig">OAUTHBEARER</a>.
                    For example, <a href="#security_sasl_gssapi_clientconfig">GSSAPI</a>
                    credentials may be configured as:
                    <pre class="line-numbers"><code class="language-text">        KafkaClient {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        keyTab="/etc/security/keytabs/kafka_client.keytab"
        principal="kafka-client-1@EXAMPLE.COM";
    };</code></pre>
                </li>
                <li>Pass the JAAS config file location as JVM parameter to each client JVM. For example:
                    <pre class="line-numbers"><code class="language-bash">    -Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf</code></pre></li>
                </ol>
                </li>
            </ol>
            </li>
        </ol>
    </li>
    <li><h4><a id="security_sasl_config"
        href="#security_sasl_config">SASL configuration</a></h4>

        <p>SASL may be used with PLAINTEXT or SSL as the transport layer using the
        security protocol SASL_PLAINTEXT or SASL_SSL respectively. If SASL_SSL is
        used, then <a href="#security_ssl">SSL must also be configured</a>.</p>

        <ol>
        <li><h5><a id="security_sasl_mechanism"
            href="#security_sasl_mechanism">SASL mechanisms</a></h5>
            Kafka supports the following SASL mechanisms:
            <ul>
                <li><a href="#security_sasl_kerberos">GSSAPI</a> (Kerberos)</li>
                <li><a href="#security_sasl_plain">PLAIN</a></li>
                <li><a href="#security_sasl_scram">SCRAM-SHA-256</a></li>
                <li><a href="#security_sasl_scram">SCRAM-SHA-512</a></li>
                <li><a href="#security_sasl_oauthbearer">OAUTHBEARER</a></li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_brokerconfig"
            href="#security_sasl_brokerconfig">SASL configuration for Kafka brokers</a></h5>
            <ol>
            <li>Configure a SASL port in server.properties, by adding at least one of
                SASL_PLAINTEXT or SASL_SSL to the <i>listeners</i> parameter, which
                contains one or more comma-separated values:
                <pre>    listeners=SASL_PLAINTEXT://host.name:port</code></pre>
                If you are only configuring a SASL port (or if you want
                the Kafka brokers to authenticate each other using SASL) then make sure
                you set the same SASL protocol for inter-broker communication:
                <pre>    security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL)</code></pre></li>
            <li>Select one or more  <a href="#security_sasl_mechanism">supported mechanisms</a>
                to enable in the broker and follow the steps to configure SASL for the mechanism.
                To enable multiple mechanisms in the broker, follow the steps
                <a href="#security_sasl_multimechanism">here</a>.</li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_clientconfig"
            href="#security_sasl_clientconfig">SASL configuration for Kafka clients</a></h5>
            <p>SASL authentication is only supported for the new Java Kafka producer and
            consumer, the older API is not supported.</p>

            <p>To configure SASL authentication on the clients, select a SASL
            <a href="#security_sasl_mechanism">mechanism</a> that is enabled in
            the broker for client authentication and follow the steps to configure SASL
            for the selected mechanism.</p></li>
        </ol>
    </li>
    <li><h4><a id="security_sasl_kerberos" href="#security_sasl_kerberos">Authentication using SASL/Kerberos</a></h4>
        <ol>
        <li><h5><a id="security_sasl_kerberos_prereq" href="#security_sasl_kerberos_prereq">Prerequisites</a></h5>
        <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 operating system user that will access Kafka with Kerberos authentication (via clients and tools).</br>
            If you have installed your own Kerberos, you will need to create these principals yourself using the following commands:
                <pre class="line-numbers"><code class="language-bash">        sudo /usr/sbin/kadmin.local -q 'addprinc -randkey kafka/{hostname}@{REALM}'
        sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/{keytabname}.keytab kafka/{hostname}@{REALM}"</code></pre></li>
            <li><b>Make sure all hosts can be reachable using hostnames</b> - it is a Kerberos requirement that all your hosts can be resolved with their FQDNs.</li>
        </ol>
        <li><h5><a id="security_sasl_kerberos_brokerconfig" href="#security_sasl_kerberos_brokerconfig">Configuring Kafka Brokers</a></h5>
        <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example (note that each broker should have its own keytab):
            <pre class="line-numbers"><code class="language-text">        KafkaServer {
            com.sun.security.auth.module.Krb5LoginModule required
            useKeyTab=true
            storeKey=true
            keyTab="/etc/security/keytabs/kafka_server.keytab"
            principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
        };

        // Zookeeper client authentication
        Client {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        keyTab="/etc/security/keytabs/kafka_server.keytab"
        principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
        };</code></pre>

            </li>
            <tt>KafkaServer</tt> section in the JAAS file tells the broker which principal to use and the location of the keytab where this principal is stored. It
            allows the broker to login using the keytab specified in this section. See <a href="#security_sasl_brokernotes">notes</a> for more details on Zookeeper SASL configuration.
            <li>Pass the JAAS and optionally the krb5 file locations as JVM parameters to each Kafka broker (see <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/KerberosReq.html">here</a> for more details):
                <pre>    -Djava.security.krb5.conf=/etc/kafka/krb5.conf
        -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre>
            </li>
            <li>Make sure the keytabs configured in the JAAS file are readable by the operating system user who is starting kafka broker.</li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>. For example:
            <pre>    listeners=SASL_PLAINTEXT://host.name:port
        security.inter.broker.protocol=SASL_PLAINTEXT
        sasl.mechanism.inter.broker.protocol=GSSAPI
        sasl.enabled.mechanisms=GSSAPI
            </code></pre>
            </li>We must also configure the service name in server.properties, which should match the principal name of the kafka brokers. In the above example, principal is "kafka/kafka1.hostname.com@EXAMPLE.com", so:
            <pre>    sasl.kerberos.service.name=kafka</code></pre>

        </ol></li>
        <li><h5><a id="security_sasl_kerberos_clientconfig" href="#security_sasl_kerberos_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
                <li>
                    Clients (producers, consumers, connect workers, etc) will authenticate to the cluster with their
                    own principal (usually with the same name as the user running the client), so obtain or create
                    these principals as needed. Then configure the JAAS configuration property for each client.
                    Different clients within a JVM may run as different users by specifying different principals.
                    The property <code>sasl.jaas.config</code> in producer.properties or consumer.properties describes
                    how clients like producer and consumer can connect to the Kafka Broker. The following is an example
                    configuration for a client using a keytab (recommended for long-running processes):
                <pre>
    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useKeyTab=true \
        storeKey=true  \
        keyTab="/etc/security/keytabs/kafka_client.keytab" \
        principal="kafka-client-1@EXAMPLE.COM";</code></pre>

                   For command-line utilities like kafka-console-consumer or kafka-console-producer, kinit can be used
                   along with "useTicketCache=true" as in:
                <pre>
    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useTicketCache=true;</code></pre>

                   JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                   as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                   <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</li>
                <li>Make sure the keytabs configured in the JAAS configuration are readable by the operating system user who is starting kafka client.</li>
                <li>Optionally pass the krb5 file locations as JVM parameters to each client JVM (see <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/KerberosReq.html">here</a> for more details):
                <pre>    -Djava.security.krb5.conf=/etc/kafka/krb5.conf</code></pre></li>
                <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_PLAINTEXT (or SASL_SSL)
    sasl.mechanism=GSSAPI
    sasl.kerberos.service.name=kafka</code></pre></li>
            </ol>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_plain" href="#security_sasl_plain">Authentication using SASL/PLAIN</a></h4>
        <p>SASL/PLAIN is a simple username/password authentication mechanism that is typically used with TLS for encryption to implement secure authentication.
        Kafka supports a default implementation for SASL/PLAIN which can be extended for production use as described <a href="#security_sasl_plain_production">here</a>.</p>
        The username is used as the authenticated <code>Principal</code> for configuration of ACLs etc.
        <ol>
        <li><h5><a id="security_sasl_plain_brokerconfig" href="#security_sasl_plain_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre class="line-numbers"><code class="language-text">        KafkaServer {
            org.apache.kafka.common.security.plain.PlainLoginModule required
            username="admin"
            password="admin-secret"
            user_admin="admin-secret"
            user_alice="alice-secret";
        };</code></pre>
                This configuration defines two users (<i>admin</i> and <i>alice</i>). The properties <tt>username</tt> and <tt>password</tt>
                in the <tt>KafkaServer</tt> section are used by the broker to initiate connections to other brokers. In this example,
                <i>admin</i> is the user for inter-broker communication. The set of properties <tt>user_<i>userName</i></tt> 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.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>. For example:
                <pre>    listeners=SASL_SSL://host.name:port
        security.inter.broker.protocol=SASL_SSL
        sasl.mechanism.inter.broker.protocol=PLAIN
        sasl.enabled.mechanisms=PLAIN</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_plain_clientconfig" href="#security_sasl_plain_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
            <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
                The following is an example configuration for a client for the PLAIN mechanism:
                <pre class="line-numbers"><code class="language-text">    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
        username="alice" \
        password="alice-secret";</code></pre>
                <p>The options <tt>username</tt> and <tt>password</tt> are used by clients to configure
                the user for client connections. In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different user names
                and passwords in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL
    sasl.mechanism=PLAIN</code></pre></li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_plain_production" href="#security_sasl_plain_production">Use of SASL/PLAIN in production</a></h5>
            <ul>
            <li>SASL/PLAIN should be used only with SSL as transport layer to ensure that clear passwords are not transmitted on the wire without encryption.</li>
            <li>The default implementation of SASL/PLAIN in Kafka specifies usernames and passwords in the JAAS configuration file as shown
                <a href="#security_sasl_plain_brokerconfig">here</a>. From Kafka version 2.0 onwards, you can avoid storing clear passwords on disk
                by configuring your own callback handlers that obtain username and password from an external source using the configuration options
                <code>sasl.server.callback.handler.class</code> and <code>sasl.client.callback.handler.class</code>.</li>
            <li>In production systems, external authentication servers may implement password authentication. From Kafka version 2.0 onwards,
                you can plug in your own callback handlers that use external authentication servers for password verification by configuring
                <code>sasl.server.callback.handler.class</code>.</li>
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_scram" href="#security_sasl_scram">Authentication using SASL/SCRAM</a></h4>
        <p>Salted Challenge Response Authentication Mechanism (SCRAM) is a family of SASL mechanisms that
        addresses the security concerns with traditional mechanisms that perform username/password authentication
        like PLAIN and DIGEST-MD5. The mechanism is defined in <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>.
        Kafka supports <a href="https://tools.ietf.org/html/rfc7677">SCRAM-SHA-256</a> and SCRAM-SHA-512 which
        can be used with TLS to perform secure authentication. The username is used as the authenticated
        <code>Principal</code> for configuration of ACLs etc. The default SCRAM implementation in Kafka
        stores SCRAM credentials in Zookeeper and is suitable for use in Kafka installations where Zookeeper
        is on a private network. Refer to <a href="#security_sasl_scram_security">Security Considerations</a>
        for more details.</p>
        <ol>
        <li><h5><a id="security_sasl_scram_credentials" href="#security_sasl_scram_credentials">Creating SCRAM Credentials</a></h5>
        <p>The SCRAM implementation in Kafka uses Zookeeper as credential store. Credentials can be created in
        Zookeeper using <tt>kafka-configs.sh</tt>. For each SCRAM mechanism enabled, credentials must be created
        by adding a config with the mechanism name. Credentials for inter-broker communication must be created
        before Kafka brokers are started. Client credentials may be created and updated dynamically and updated
        credentials will be used to authenticate new connections.</p>
        <p>Create SCRAM credentials for user <i>alice</i> with password <i>alice-secret</i>:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]' --entity-type users --entity-name alice
        </code></pre>
        <p>The default iteration count of 4096 is used if iterations are not specified. A random salt is created
        and the SCRAM identity consisting of salt, iterations, StoredKey and ServerKey are stored in Zookeeper.
        See <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a> for details on SCRAM identity and the individual fields.
        <p>The following examples also require a user <i>admin</i> for inter-broker communication which can be created using:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin
        </code></pre>
        <p>Existing credentials may be listed using the <i>--describe</i> option:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name alice
        </code></pre>
        <p>Credentials may be deleted for one or more SCRAM mechanisms using the <i>--delete</i> option:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name alice
        </code></pre>
        </li>
        <li><h5><a id="security_sasl_scram_brokerconfig" href="#security_sasl_scram_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre>
    KafkaServer {
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="admin"
        password="admin-secret";
    };</code></pre>
                The properties <tt>username</tt> and <tt>password</tt> in the <tt>KafkaServer</tt> section are used by
                the broker to initiate connections to other brokers. In this example, <i>admin</i> is the user for
                inter-broker communication.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>.</code></pre> For example:
                <pre>
    listeners=SASL_SSL://host.name:port
    security.inter.broker.protocol=SASL_SSL
    sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 (or SCRAM-SHA-512)
    sasl.enabled.mechanisms=SCRAM-SHA-256 (or SCRAM-SHA-512)</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_scram_clientconfig" href="#security_sasl_scram_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
            <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
                The following is an example configuration for a client for the SCRAM mechanisms:
                <pre class="line-numbers"><code class="language-text">   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
        username="alice" \
        password="alice-secret";</code></pre>

                <p>The options <tt>username</tt> and <tt>password</tt> are used by clients to configure
                the user for client connections. In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different user names
                and passwords in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL
    sasl.mechanism=SCRAM-SHA-256 (or SCRAM-SHA-512)</code></pre></li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_scram_security" href="#security_sasl_scram_security">Security Considerations for SASL/SCRAM</a></h5>
            <ul>
            <li>The default implementation of SASL/SCRAM in Kafka stores SCRAM credentials in Zookeeper. This
            is suitable for production use in installations where Zookeeper is secure and on a private network.</li>
            <li>Kafka supports only the strong hash functions SHA-256 and SHA-512 with a minimum iteration count
            of 4096. Strong hash functions combined with strong passwords and high iteration counts protect
            against brute force attacks if Zookeeper security is compromised.</li>
            <li>SCRAM should be used only with TLS-encryption to prevent interception of SCRAM exchanges. This
            protects against dictionary or brute force attacks and against impersonation if Zookeeper is compromised.</li>
            <li>From Kafka version 2.0 onwards, the default SASL/SCRAM credential store may be overridden using custom callback handlers
            by configuring <code>sasl.server.callback.handler.class</code> in installations where Zookeeper is not secure.</li>
            <li>For more details on security considerations, refer to
            <a href="https://tools.ietf.org/html/rfc5802#section-9">RFC 5802</a>.
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_oauthbearer" href="#security_sasl_oauthbearer">Authentication using SASL/OAUTHBEARER</a></h4>
        <p>The <a href="https://tools.ietf.org/html/rfc6749">OAuth 2 Authorization Framework</a> "enables a third-party application to obtain limited access to an HTTP service,
        either on behalf of a resource owner by orchestrating an approval interaction between the resource owner and the HTTP
        service, or by allowing the third-party application to obtain access on its own behalf."  The SASL OAUTHBEARER mechanism
        enables the use of the framework in a SASL (i.e. a non-HTTP) context; it is defined in <a href="https://tools.ietf.org/html/rfc7628">RFC 7628</a>.
        The default OAUTHBEARER implementation in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>
        and is only suitable for use in non-production Kafka installations. Refer to <a href="#security_sasl_oauthbearer_security">Security Considerations</a>
        for more details.</p>
        <ol>
        <li><h5><a id="security_sasl_oauthbearer_brokerconfig" href="#security_sasl_oauthbearer_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre>
    KafkaServer {
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
        unsecuredLoginStringClaim_sub="admin";
    };</code></pre>
                The property <tt>unsecuredLoginStringClaim_sub</tt> in the <tt>KafkaServer</tt> section is used by
                the broker when it initiates connections to other brokers. In this example, <i>admin</i> will appear in the
                subject (<tt>sub</tt>) claim and will be the user for inter-broker communication.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>.</code></pre> For example:
                <pre>
    listeners=SASL_SSL://host.name:port (or SASL_PLAINTEXT if non-production)
    security.inter.broker.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
    sasl.mechanism.inter.broker.protocol=OAUTHBEARER
    sasl.enabled.mechanisms=OAUTHBEARER</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_oauthbearer_clientconfig" href="#security_sasl_oauthbearer_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
	    <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
	        The following is an example configuration for a client for the OAUTHBEARER mechanisms:
                <pre class="line-numbers"><code class="language-text">   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        unsecuredLoginStringClaim_sub="alice";</code></pre>

                <p>The option <tt>unsecuredLoginStringClaim_sub</tt> is used by clients to configure
                the subject (<tt>sub</tt>) claim, which determines the user for client connections.
                In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different subject (<tt>sub</tt>)
                claims in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
    sasl.mechanism=OAUTHBEARER</code></pre></li>
             <li>The default implementation of SASL/OAUTHBEARER depends on the jackson-databind library.
                 Since it's an optional dependency, users have to configure it as a dependency via their build tool.</li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_unsecured_retrieval" href="#security_sasl_oauthbearer_unsecured_retrieval">Unsecured Token Creation Options for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>The default implementation of SASL/OAUTHBEARER in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>.
            While suitable only for non-production use, it does provide the flexibility to create arbitrary tokens in a DEV or TEST environment.</li>
            <li>Here are the various supported JAAS module options on the client side (and on the broker side if OAUTHBEARER is the inter-broker protocol):
                 <table>
                 <tr>
                 <th>JAAS Module Option for Unsecured Token Creation</th>
                 <th>Documentation</th>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginStringClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>String</tt> claim with the given name and value. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginNumberClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>Number</tt> claim with the given name and value. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginListClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>String List</tt> claim with the given name and values parsed
                 from the given value where the first character is taken as the delimiter. For
                 example: <tt>unsecuredLoginListClaim_fubar="|value1|value2"</tt>. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginExtension_&lt;extensionname&gt;="value"</tt></td>
                 <td>Creates a <tt>String</tt> extension with the given name and value.
                 For example: <tt>unsecuredLoginExtension_traceId="123"</tt>. A valid extension name
                 is any sequence of lowercase or uppercase alphabet characters. In addition, the "auth" extension name is reserved.
                 A valid extension value is any combination of characters with ASCII codes 1-127.
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginPrincipalClaimName</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt>
                 claim holding the principal name to be something other than '<tt>sub</tt>'.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginLifetimeSeconds</tt></td>
                 <td>Set to an integer value if the token expiration is to be set to something
                 other than the default value of 3600 seconds (which is 1 hour). The
                 '<tt>exp</tt>' claim will be set to reflect the expiration time.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginScopeClaimName</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt> or
                 <tt>String List</tt> claim holding any token scope to be something other than
                 '<tt>scope</tt>'.</td>
                 </tr>
                 </table>
            </li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_unsecured_validation" href="#security_sasl_oauthbearer_unsecured_validation">Unsecured Token Validation Options for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>Here are the various supported JAAS module options on the broker side for <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Token</a> validation:
                 <table>
                 <tr>
                 <th>JAAS Module Option for Unsecured Token Validation</th>
                 <th>Documentation</th>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorPrincipalClaimName="value"</tt></td>
                 <td>Set to a non-empty value if you wish a particular <tt>String</tt> claim
                 holding a principal name to be checked for existence; the default is to check
                 for the existence of the '<tt>sub</tt>' claim.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorScopeClaimName="value"</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt> or
                 <tt>String List</tt> claim holding any token scope to be something other than
                 '<tt>scope</tt>'.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorRequiredScope="value"</tt></td>
                 <td>Set to a space-delimited list of scope values if you wish the
                 <tt>String/String List</tt> claim holding the token scope to be checked to
                 make sure it contains certain values.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorAllowableClockSkewMs="value"</tt></td>
                 <td>Set to a positive integer value if you wish to allow up to some number of
                 positive milliseconds of clock skew (the default is 0).</td>
                 </tr>
                 </table>
            </li>
            <li>The default unsecured SASL/OAUTHBEARER implementation may be overridden (and must be overridden in production environments)
            using custom login and SASL Server callback handlers.</li>
            <li>For more details on security considerations, refer to <a href="https://tools.ietf.org/html/rfc6749#section-10">RFC 6749, Section 10</a>.</li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_refresh" href="#security_sasl_oauthbearer_refresh">Token Refresh for SASL/OAUTHBEARER</a></h5>
        Kafka periodically refreshes any token before it expires so that the client can continue to make
        connections to brokers. The parameters that impact how the refresh algorithm
        operates are specified as part of the producer/consumer/broker configuration
        and are as follows. See the documentation for these properties elsewhere for
        details.  The default values are usually reasonable, in which case these
        configuration parameters would not need to be explicitly set.
        <table>
        <tr>
        <th>Producer/Consumer/Broker Configuration Property</th>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.window.factor</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.window.jitter</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.min.period.seconds</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.min.buffer.seconds</tt></td>
        </tr>
        </table>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_prod" href="#security_sasl_oauthbearer_prod">Secure/Production Use of SASL/OAUTHBEARER</a></h5>
        Production use cases will require writing an implementation of
        <tt>org.apache.kafka.common.security.auth.AuthenticateCallbackHandler</tt> that can handle an instance of
        <tt>org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback</tt> and declaring it via either the
        <tt>sasl.login.callback.handler.class</tt> configuration option for a
        non-broker client or via the
        <tt>listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class</tt>
        configuration option for brokers (when SASL/OAUTHBEARER is the inter-broker
        protocol).
        <p>
        Production use cases will also require writing an implementation of
        <tt>org.apache.kafka.common.security.auth.AuthenticateCallbackHandler</tt> that can handle an instance of
        <tt>org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallback</tt> and declaring it via the
        <tt>listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class</tt>
        broker configuration option.
        </li>
        <li><h5><a id="security_sasl_oauthbearer_security" href="#security_sasl_oauthbearer_security">Security Considerations for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>The default implementation of SASL/OAUTHBEARER in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>.
            This is suitable only for non-production use.</li>
            <li>OAUTHBEARER should be used in production enviromnments only with TLS-encryption to prevent interception of tokens.</li>
            <li>The default unsecured SASL/OAUTHBEARER implementation may be overridden (and must be overridden in production environments)
            using custom login and SASL Server callback handlers as described above.</li>
            <li>For more details on OAuth 2 security considerations in general, refer to <a href="https://tools.ietf.org/html/rfc6749#section-10">RFC 6749, Section 10</a>.</li>
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_multimechanism" href="#security_sasl_multimechanism">Enabling multiple SASL mechanisms in a broker</a></h4>
        <ol>
        <li>Specify configuration for the login modules of all enabled mechanisms in the <tt>KafkaServer</tt> section of the JAAS config file. For example:
            <pre>
        KafkaServer {
            com.sun.security.auth.module.Krb5LoginModule required
            useKeyTab=true
            storeKey=true
            keyTab="/etc/security/keytabs/kafka_server.keytab"
            principal="kafka/kafka1.hostname.com@EXAMPLE.COM";

            org.apache.kafka.common.security.plain.PlainLoginModule required
            username="admin"
            password="admin-secret"
            user_admin="admin-secret"
            user_alice="alice-secret";
        };</code></pre></li>
        <li>Enable the SASL mechanisms in server.properties: <pre>    sasl.enabled.mechanisms=GSSAPI,PLAIN,SCRAM-SHA-256,SCRAM-SHA-512,OAUTHBEARER</code></pre></li>
        <li>Specify the SASL security protocol and mechanism for inter-broker communication in server.properties if required:
            <pre>
    security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL)
    sasl.mechanism.inter.broker.protocol=GSSAPI (or one of the other enabled mechanisms)</code></pre></li>
        <li>Follow the mechanism-specific steps in <a href="#security_sasl_kerberos_brokerconfig">GSSAPI (Kerberos)</a>,
            <a href="#security_sasl_plain_brokerconfig">PLAIN</a>,
            <a href="#security_sasl_scram_brokerconfig">SCRAM</a> and <a href="#security_sasl_oauthbearer_brokerconfig">OAUTHBEARER</a>
            to configure SASL for the enabled mechanisms.</li>
        </ol>
    </li>
    <li><h4><a id="saslmechanism_rolling_upgrade" href="#saslmechanism_rolling_upgrade">Modifying SASL mechanism in a Running Cluster</a></h4>
        <p>SASL mechanism can be modified in a running cluster using the following sequence:</p>
        <ol>
        <li>Enable new SASL mechanism by adding the mechanism to <tt>sasl.enabled.mechanisms</tt> in server.properties for each broker. Update JAAS config file to include both
            mechanisms as described <a href="#security_sasl_multimechanism">here</a>. Incrementally bounce the cluster nodes.</li>
        <li>Restart clients using the new mechanism.</li>
        <li>To change the mechanism of inter-broker communication (if this is required), set <tt>sasl.mechanism.inter.broker.protocol</tt> in server.properties to the new mechanism and
            incrementally bounce the cluster again.</li>
        <li>To remove old mechanism (if this is required), remove the old mechanism from <tt>sasl.enabled.mechanisms</tt> in server.properties and remove the entries for the
            old mechanism from JAAS config file. Incrementally bounce the cluster again.</li>
        </ol>
    </li>

    <li><h4><a id="security_delegation_token" href="#security_delegation_token">Authentication using Delegation Tokens</a></h4>
        <p>Delegation token based authentication is a lightweight authentication mechanism to complement existing SASL/SSL
            methods. Delegation tokens are shared secrets between kafka brokers and clients. Delegation tokens will help processing
            frameworks to distribute the workload to available workers in a secure environment without the added cost of distributing
            Kerberos TGT/keytabs or keystores when 2-way SSL is used. See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-48+Delegation+token+support+for+Kafka">KIP-48</a>
            for more details.</p>

        <p>Typical steps for delegation token usage are:</p>
        <ol>
        <li>User authenticates with the Kafka cluster via SASL or SSL, and obtains a delegation token. This can be done
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



24/security.html [18:895]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
<script id="security-template" type="text/x-handlebars-template">
    <h3><a id="security_overview" href="#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>Authentication of connections to brokers from clients (producers and consumers), other brokers and tools, using either SSL or SASL. Kafka supports the following SASL mechanisms:
        <ul>
            <li>SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0</li>
            <li>SASL/PLAIN - starting at version 0.10.0.0</li>
            <li>SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512 - starting at version 0.10.2.0</li>
            <li>SASL/OAUTHBEARER - starting at version 2.0</li>
        </ul></li>
        <li>Authentication of connections from brokers to ZooKeeper</li>
        <li>Encryption of data transferred between brokers and clients, between brokers, or between brokers and tools using SSL (Note that there is a performance degradation when SSL is enabled, the magnitude of which depends on the CPU type and the JVM implementation.)</li>
        <li>Authorization of read / write operations by clients</li>
        <li>Authorization is pluggable and integration with external authorization services is supported</li>
    </ol>

    It's worth noting that security is optional - non-secured clusters are supported, as well as a mix of authenticated, unauthenticated, encrypted and non-encrypted clients.

    The guides below explain how to configure and use the security features in both clients and brokers.

    <h3><a id="security_ssl" href="#security_ssl">7.2 Encryption and Authentication using SSL</a></h3>
    Apache Kafka allows clients to connect over SSL. By default, SSL is disabled but can be turned on as needed.

    <ol>
        <li><h4><a id="security_ssl_key" href="#security_ssl_key">Generate SSL key and certificate for each Kafka broker</a></h4>
            The first step of deploying one or more brokers with the SSL support 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.
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA</code></pre>

            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>
            <br>

        <h5><a id="security_confighostname" href="#security_confighostname">Configuring Host Name Verification</a></h5>
        From Kafka version 2.0.0 onwards, host name verification of servers is enabled by default for client connections
        as well as inter-broker connections to prevent man-in-the-middle attacks. Server host name verification may be disabled
        by setting <code>ssl.endpoint.identification.algorithm</code> to an empty string. For example,
        <pre class="line-numbers"><code class="language-text">	ssl.endpoint.identification.algorithm=</code></pre>
        For dynamically configured broker listeners, hostname verification may be disabled using <code>kafka-configs.sh</code>.
        For example,
        <pre class="line-numbers"><code class="language-text">        bin/kafka-configs.sh --bootstrap-server localhost:9093 --entity-type brokers --entity-name 0 --alter --add-config "listener.name.internal.ssl.endpoint.identification.algorithm="
        </code></pre>

        For older versions of Kafka, <code>ssl.endpoint.identification.algorithm</code> is not defined by default, so host name
        verification is not performed. The property should be set to <code>HTTPS</code> to enable host name verification.

        <pre class="line-numbers"><code class="language-text">	ssl.endpoint.identification.algorithm=HTTPS </code></pre>

        Host name verification must be enabled to prevent man-in-the-middle attacks if server endpoints are not validated
        externally.

        <h5><a id="security_configcerthostname" href="#security_configcerthstname">Configuring Host Name In Certificates</a></h5>
        If host name verification is enabled, clients will verify the server's fully qualified domain name (FQDN) against one of
        the following two fields:
        <ol>
            <li>Common Name (CN)
            <li>Subject Alternative Name (SAN)
        </ol>
        <br>
        Both fields are valid, RFC-2818 recommends the use of SAN however. 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 following argument <code> -ext SAN=DNS:{FQDN} </code> to the keytool command:
        <pre class="line-numbers"><code class="language-bash">        keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA -ext SAN=DNS:{FQDN}
        </code></pre>
        The following command can be run afterwards to verify the contents of the generated certificate:
        <pre class="line-numbers"><code class="language-bash">        keytool -list -v -keystore server.keystore.jks
        </code></pre>
        </li>
        <li><h4><a id="security_ssl_ca" href="#security_ssl_ca">Creating your own CA</a></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.
            <pre class="line-numbers"><code class="language-bash">            openssl req -new -x509 -keyout ca-key -out ca-cert -days 365</code></pre>

            The generated CA is simply a public-private key pair and certificate, and it is intended to sign other certificates.<br>

            The next step is to add the generated CA to the **clients' truststore** so that the clients can trust this CA:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert</code></pre>

            <b>Note:</b> If you configure the Kafka brokers to require client authentication by setting ssl.client.auth to be "requested" or "required" on the <a href="#config_broker">Kafka brokers config</a> then you must provide a truststore for the Kafka brokers as well and it should have all the CA certificates that clients' keys were signed by.
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert</code></pre>

            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 trusting all certificates that are signed by that certificate. As the analogy above, trusting the government (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 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><a id="security_ssl_signing" href="#security_ssl_signing">Signing the certificate</a></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:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file</code></pre>

            Then sign it with the CA:
            <pre class="line-numbers"><code class="language-bash">            openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days {validity} -CAcreateserial -passin pass:{ca-password}</code></pre>

            Finally, you need to import both the certificate of the CA and the signed certificate into the keystore:
            <pre class="line-numbers"><code class="language-bash">            keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
            keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed</code></pre>

            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.
                <pre>
            #!/bin/bash
            #Step 1
            keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
            #Step 2
            openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
            keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
            keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
            #Step 3
            keytool -keystore 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 365 -CAcreateserial -passin pass:test1234
            keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
            keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed</code></pre></li>
        <li><h4><a id="security_configbroker" href="#security_configbroker">Configuring Kafka Brokers</a></h4>
            Kafka Brokers support listening for connections on multiple ports.
            We need to configure the following property in server.properties, which must have one or more comma-separated values:
            <pre>listeners</code></pre>

            If SSL is not enabled for inter-broker communication (see below for how to enable it), both PLAINTEXT and SSL ports will be necessary.
            <pre class="line-numbers"><code class="language-text">            listeners=PLAINTEXT://host.name:port,SSL://host.name:port</code></pre>

            Following SSL configs are needed on the broker side
            <pre class="line-numbers"><code class="language-text">            ssl.keystore.location=/var/private/ssl/server.keystore.jks
            ssl.keystore.password=test1234
            ssl.key.password=test1234
            ssl.truststore.location=/var/private/ssl/server.truststore.jks
            ssl.truststore.password=test1234</code></pre>

            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.

            Optional settings that are worth considering:
            <ol>
                <li>ssl.client.auth=none ("required" => client authentication is required, "requested" => client authentication is requested and client without certs can still connect. The usage of "requested" is discouraged as it provides a false sense of security and misconfigured clients will still connect successfully.)</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. (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 that SSL is deprecated in favor of TLS and using SSL in production is not recommended)</li>
                <li>ssl.keystore.type=JKS</li>
                <li>ssl.truststore.type=JKS</li>
                <li>ssl.secure.random.implementation=SHA1PRNG</li>
            </ol>
            If you want to enable SSL for inter-broker communication, add the following to the server.properties file (it defaults to PLAINTEXT)
            <pre>
            security.inter.broker.protocol=SSL</code></pre>

            <p>
            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 <a href="http://www.oracle.com/technetwork/java/javase/downloads/index.html">JCE Unlimited Strength Jurisdiction Policy Files</a> must be obtained and installed in the JDK/JRE. See the
            <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/SunProviders.html">JCA Providers Documentation</a> for more information.
            </p>

            <p>
            The JRE/JDK will have a default pseudo-random number generator (PRNG) that is used for cryptography operations, so it is not required to configure the
            implementation used with the <pre>ssl.secure.random.implementation</code></pre>. However, there are performance issues with some implementations (notably, the
            default chosen on Linux systems, <pre>NativePRNG</code></pre>, utilizes a global lock). In cases where performance of SSL connections becomes an issue,
            consider explicitly setting the implementation to be used. The <pre>SHA1PRNG</code></pre> implementation is non-blocking, and has shown very good performance
            characteristics under heavy load (50 MB/sec of produced messages, plus replication traffic, per-broker).
            </p>

            Once you start the broker you should be able to see in the server.log
            <pre>
            with addresses: PLAINTEXT -> EndPoint(192.168.64.1,9092,PLAINTEXT),SSL -> EndPoint(192.168.64.1,9093,SSL)</code></pre>

            To check quickly if  the server keystore and truststore are setup properly you can run the following command
            <pre>openssl s_client -debug -connect localhost:9093 -tls1</code></pre> (Note: TLSv1 should be listed under ssl.enabled.protocols)<br>
            In the output of this command you should see server's certificate:
            <pre>
            -----BEGIN CERTIFICATE-----
            {variable sized random bytes}
            -----END CERTIFICATE-----
            subject=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=Sriharsha Chintalapani
            issuer=/C=US/ST=CA/L=Santa Clara/O=org/OU=org/CN=kafka/emailAddress=test@test.com</code></pre>
            If the certificate does not show up or if there are any other error messages then your keystore is not setup properly.</li>

        <li><h4><a id="security_configclients" href="#security_configclients">Configuring Kafka Clients</a></h4>
            SSL is supported only for the new Kafka Producer and Consumer, the older API is not supported. The configs for SSL will be the same for both producer and consumer.<br>
            If client authentication is not required in the broker, then the following is a minimal configuration example:
            <pre class="line-numbers"><code class="language-text">            security.protocol=SSL
            ssl.truststore.location=/var/private/ssl/client.truststore.jks
            ssl.truststore.password=test1234</code></pre>

            Note: ssl.truststore.password is technically optional but highly recommended. If a password is not set access to the truststore is still available, but integrity checking is disabled.

            If client authentication is required, then a keystore must be created like in step 1 and the following must also be configured:
            <pre class="line-numbers"><code class="language-text">            ssl.keystore.location=/var/private/ssl/client.keystore.jks
            ssl.keystore.password=test1234
            ssl.key.password=test1234</code></pre>

            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. It 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:
            <pre class="line-numbers"><code class="language-bash">            kafka-console-producer.sh --broker-list localhost:9093 --topic test --producer.config client-ssl.properties
            kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config client-ssl.properties</code></pre>
        </li>
    </ol>
    <h3><a id="security_sasl" href="#security_sasl">7.3 Authentication using SASL</a></h3>

    <ol>
    <li><h4><a id="security_sasl_jaasconfig" href="#security_sasl_jaasconfig">JAAS configuration</a></h4>
    <p>Kafka uses the Java Authentication and Authorization Service
    (<a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jaas/JAASRefGuide.html">JAAS</a>)
    for SASL configuration.</p>
        <ol>
        <li><h5><a id="security_jaas_broker"
            href="#security_jaas_broker">JAAS configuration for Kafka brokers</a></h5>

            <p><tt>KafkaServer</tt> is the section name in the JAAS file used by each
            KafkaServer/Broker. This section provides SASL configuration options
            for the broker including any SASL client connections made by the broker
            for inter-broker communication. If multiple listeners are configured to use
            SASL, the section name may be prefixed with the listener name in lower-case
            followed by a period, e.g. <tt>sasl_ssl.KafkaServer</tt>.</p>

            <p><tt>Client</tt> section is used to authenticate a SASL connection with
            zookeeper. It also allows the brokers to set SASL ACL on zookeeper
            nodes which locks these nodes down so that only the brokers can
            modify it. It is necessary to have the same principal name across all
            brokers. If you want to use a section name other than Client, set the
            system property <tt>zookeeper.sasl.clientconfig</tt> to the appropriate
            name (<i>e.g.</i>, <tt>-Dzookeeper.sasl.clientconfig=ZkClient</tt>).</p>

            <p>ZooKeeper uses "zookeeper" as the service name by default. If you
            want to change this, set the system property
            <tt>zookeeper.sasl.client.username</tt> to the appropriate name
            (<i>e.g.</i>, <tt>-Dzookeeper.sasl.client.username=zk</tt>).</p></li>

            <p>Brokers may also configure JAAS using the broker configuration property <code>sasl.jaas.config</code>.
            The property name must be prefixed with the listener prefix including the SASL mechanism,
            i.e. <code>listener.name.{listenerName}.{saslMechanism}.sasl.jaas.config</code>. Only one
            login module may be specified in the config value. If multiple mechanisms are configured on a
            listener, configs must be provided for each mechanism using the listener and mechanism prefix.
            For example,
                    <pre class="line-numbers"><code class="language-text">        listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
            username="admin" \
            password="admin-secret";
        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_alice="alice-secret";</code></pre>

            If JAAS configuration is defined at different levels, the order of precedence used is:
            <ul>
              <li>Broker configuration property <code>listener.name.{listenerName}.{saslMechanism}.sasl.jaas.config</code></li>
              <li><code>{listenerName}.KafkaServer</code> section of static JAAS configuration</code></li>
              <li><code>KafkaServer</code> section of static JAAS configuration</code></li>
            </ul>
            Note that ZooKeeper JAAS config may only be configured using static JAAS configuration.

            <p>See <a href="#security_sasl_kerberos_brokerconfig">GSSAPI (Kerberos)</a>,
            <a href="#security_sasl_plain_brokerconfig">PLAIN</a>,
            <a href="#security_sasl_scram_brokerconfig">SCRAM</a> or
            <a href="#security_sasl_oauthbearer_brokerconfig">OAUTHBEARER</a> for example broker configurations.</p></li>


        <li><h5><a id="security_jaas_client"
            href="#security_jaas_client">JAAS configuration for Kafka clients</a></h5>

            <p>Clients may configure JAAS using the client configuration property
            <a href="#security_client_dynamicjaas">sasl.jaas.config</a>
            or using the <a href="#security_client_staticjaas">static JAAS config file</a>
            similar to brokers.</p>

            <ol>
            <li><h6><a id="security_client_dynamicjaas"
                href="#security_client_dynamicjaas">JAAS configuration using client configuration property</a></h6>
                <p>Clients may specify JAAS configuration as a producer or consumer property without
                creating a physical configuration file. This mode also enables different producers
                and consumers within the same JVM to use different credentials by specifying
                different properties for each client. If both static JAAS configuration system property
                <code>java.security.auth.login.config</code> and client property <code>sasl.jaas.config</code>
                are specified, the client property will be used.</p>

                <p>See <a href="#security_sasl_kerberos_clientconfig">GSSAPI (Kerberos)</a>,
                <a href="#security_sasl_plain_clientconfig">PLAIN</a>,
                <a href="#security_sasl_scram_clientconfig">SCRAM</a> or
                <a href="#security_sasl_oauthbearer_clientconfig">OAUTHBEARER</a> for example configurations.</p></li>

                <li><h6><a id="security_client_staticjaas" href="#security_client_staticjaas">JAAS configuration using static config file</a></h6>
                To configure SASL authentication on the clients using static JAAS config file:
                <ol>
                <li>Add a JAAS config file with a client login section named <tt>KafkaClient</tt>. Configure
                    a login module in <tt>KafkaClient</tt> for the selected mechanism as described in the examples
                    for setting up <a href="#security_sasl_kerberos_clientconfig">GSSAPI (Kerberos)</a>,
                    <a href="#security_sasl_plain_clientconfig">PLAIN</a>,
                    <a href="#security_sasl_scram_clientconfig">SCRAM</a> or
                    <a href="#security_sasl_oauthbearer_clientconfig">OAUTHBEARER</a>.
                    For example, <a href="#security_sasl_gssapi_clientconfig">GSSAPI</a>
                    credentials may be configured as:
                    <pre class="line-numbers"><code class="language-text">        KafkaClient {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        keyTab="/etc/security/keytabs/kafka_client.keytab"
        principal="kafka-client-1@EXAMPLE.COM";
    };</code></pre>
                </li>
                <li>Pass the JAAS config file location as JVM parameter to each client JVM. For example:
                    <pre class="line-numbers"><code class="language-bash">    -Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf</code></pre></li>
                </ol>
                </li>
            </ol>
            </li>
        </ol>
    </li>
    <li><h4><a id="security_sasl_config"
        href="#security_sasl_config">SASL configuration</a></h4>

        <p>SASL may be used with PLAINTEXT or SSL as the transport layer using the
        security protocol SASL_PLAINTEXT or SASL_SSL respectively. If SASL_SSL is
        used, then <a href="#security_ssl">SSL must also be configured</a>.</p>

        <ol>
        <li><h5><a id="security_sasl_mechanism"
            href="#security_sasl_mechanism">SASL mechanisms</a></h5>
            Kafka supports the following SASL mechanisms:
            <ul>
                <li><a href="#security_sasl_kerberos">GSSAPI</a> (Kerberos)</li>
                <li><a href="#security_sasl_plain">PLAIN</a></li>
                <li><a href="#security_sasl_scram">SCRAM-SHA-256</a></li>
                <li><a href="#security_sasl_scram">SCRAM-SHA-512</a></li>
                <li><a href="#security_sasl_oauthbearer">OAUTHBEARER</a></li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_brokerconfig"
            href="#security_sasl_brokerconfig">SASL configuration for Kafka brokers</a></h5>
            <ol>
            <li>Configure a SASL port in server.properties, by adding at least one of
                SASL_PLAINTEXT or SASL_SSL to the <i>listeners</i> parameter, which
                contains one or more comma-separated values:
                <pre>    listeners=SASL_PLAINTEXT://host.name:port</code></pre>
                If you are only configuring a SASL port (or if you want
                the Kafka brokers to authenticate each other using SASL) then make sure
                you set the same SASL protocol for inter-broker communication:
                <pre>    security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL)</code></pre></li>
            <li>Select one or more  <a href="#security_sasl_mechanism">supported mechanisms</a>
                to enable in the broker and follow the steps to configure SASL for the mechanism.
                To enable multiple mechanisms in the broker, follow the steps
                <a href="#security_sasl_multimechanism">here</a>.</li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_clientconfig"
            href="#security_sasl_clientconfig">SASL configuration for Kafka clients</a></h5>
            <p>SASL authentication is only supported for the new Java Kafka producer and
            consumer, the older API is not supported.</p>

            <p>To configure SASL authentication on the clients, select a SASL
            <a href="#security_sasl_mechanism">mechanism</a> that is enabled in
            the broker for client authentication and follow the steps to configure SASL
            for the selected mechanism.</p></li>
        </ol>
    </li>
    <li><h4><a id="security_sasl_kerberos" href="#security_sasl_kerberos">Authentication using SASL/Kerberos</a></h4>
        <ol>
        <li><h5><a id="security_sasl_kerberos_prereq" href="#security_sasl_kerberos_prereq">Prerequisites</a></h5>
        <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 operating system user that will access Kafka with Kerberos authentication (via clients and tools).</br>
            If you have installed your own Kerberos, you will need to create these principals yourself using the following commands:
                <pre class="line-numbers"><code class="language-bash">        sudo /usr/sbin/kadmin.local -q 'addprinc -randkey kafka/{hostname}@{REALM}'
        sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/{keytabname}.keytab kafka/{hostname}@{REALM}"</code></pre></li>
            <li><b>Make sure all hosts can be reachable using hostnames</b> - it is a Kerberos requirement that all your hosts can be resolved with their FQDNs.</li>
        </ol>
        <li><h5><a id="security_sasl_kerberos_brokerconfig" href="#security_sasl_kerberos_brokerconfig">Configuring Kafka Brokers</a></h5>
        <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example (note that each broker should have its own keytab):
            <pre class="line-numbers"><code class="language-text">        KafkaServer {
            com.sun.security.auth.module.Krb5LoginModule required
            useKeyTab=true
            storeKey=true
            keyTab="/etc/security/keytabs/kafka_server.keytab"
            principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
        };

        // Zookeeper client authentication
        Client {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        keyTab="/etc/security/keytabs/kafka_server.keytab"
        principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
        };</code></pre>

            </li>
            <tt>KafkaServer</tt> section in the JAAS file tells the broker which principal to use and the location of the keytab where this principal is stored. It
            allows the broker to login using the keytab specified in this section. See <a href="#security_sasl_brokernotes">notes</a> for more details on Zookeeper SASL configuration.
            <li>Pass the JAAS and optionally the krb5 file locations as JVM parameters to each Kafka broker (see <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/KerberosReq.html">here</a> for more details):
                <pre>    -Djava.security.krb5.conf=/etc/kafka/krb5.conf
        -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre>
            </li>
            <li>Make sure the keytabs configured in the JAAS file are readable by the operating system user who is starting kafka broker.</li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>. For example:
            <pre>    listeners=SASL_PLAINTEXT://host.name:port
        security.inter.broker.protocol=SASL_PLAINTEXT
        sasl.mechanism.inter.broker.protocol=GSSAPI
        sasl.enabled.mechanisms=GSSAPI
            </code></pre>
            </li>We must also configure the service name in server.properties, which should match the principal name of the kafka brokers. In the above example, principal is "kafka/kafka1.hostname.com@EXAMPLE.com", so:
            <pre>    sasl.kerberos.service.name=kafka</code></pre>

        </ol></li>
        <li><h5><a id="security_sasl_kerberos_clientconfig" href="#security_sasl_kerberos_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
                <li>
                    Clients (producers, consumers, connect workers, etc) will authenticate to the cluster with their
                    own principal (usually with the same name as the user running the client), so obtain or create
                    these principals as needed. Then configure the JAAS configuration property for each client.
                    Different clients within a JVM may run as different users by specifying different principals.
                    The property <code>sasl.jaas.config</code> in producer.properties or consumer.properties describes
                    how clients like producer and consumer can connect to the Kafka Broker. The following is an example
                    configuration for a client using a keytab (recommended for long-running processes):
                <pre>
    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useKeyTab=true \
        storeKey=true  \
        keyTab="/etc/security/keytabs/kafka_client.keytab" \
        principal="kafka-client-1@EXAMPLE.COM";</code></pre>

                   For command-line utilities like kafka-console-consumer or kafka-console-producer, kinit can be used
                   along with "useTicketCache=true" as in:
                <pre>
    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useTicketCache=true;</code></pre>

                   JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                   as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                   <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</li>
                <li>Make sure the keytabs configured in the JAAS configuration are readable by the operating system user who is starting kafka client.</li>
                <li>Optionally pass the krb5 file locations as JVM parameters to each client JVM (see <a href="https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/KerberosReq.html">here</a> for more details):
                <pre>    -Djava.security.krb5.conf=/etc/kafka/krb5.conf</code></pre></li>
                <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_PLAINTEXT (or SASL_SSL)
    sasl.mechanism=GSSAPI
    sasl.kerberos.service.name=kafka</code></pre></li>
            </ol>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_plain" href="#security_sasl_plain">Authentication using SASL/PLAIN</a></h4>
        <p>SASL/PLAIN is a simple username/password authentication mechanism that is typically used with TLS for encryption to implement secure authentication.
        Kafka supports a default implementation for SASL/PLAIN which can be extended for production use as described <a href="#security_sasl_plain_production">here</a>.</p>
        The username is used as the authenticated <code>Principal</code> for configuration of ACLs etc.
        <ol>
        <li><h5><a id="security_sasl_plain_brokerconfig" href="#security_sasl_plain_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre class="line-numbers"><code class="language-text">        KafkaServer {
            org.apache.kafka.common.security.plain.PlainLoginModule required
            username="admin"
            password="admin-secret"
            user_admin="admin-secret"
            user_alice="alice-secret";
        };</code></pre>
                This configuration defines two users (<i>admin</i> and <i>alice</i>). The properties <tt>username</tt> and <tt>password</tt>
                in the <tt>KafkaServer</tt> section are used by the broker to initiate connections to other brokers. In this example,
                <i>admin</i> is the user for inter-broker communication. The set of properties <tt>user_<i>userName</i></tt> 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.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>. For example:
                <pre>    listeners=SASL_SSL://host.name:port
        security.inter.broker.protocol=SASL_SSL
        sasl.mechanism.inter.broker.protocol=PLAIN
        sasl.enabled.mechanisms=PLAIN</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_plain_clientconfig" href="#security_sasl_plain_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
            <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
                The following is an example configuration for a client for the PLAIN mechanism:
                <pre class="line-numbers"><code class="language-text">    sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
        username="alice" \
        password="alice-secret";</code></pre>
                <p>The options <tt>username</tt> and <tt>password</tt> are used by clients to configure
                the user for client connections. In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different user names
                and passwords in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL
    sasl.mechanism=PLAIN</code></pre></li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_plain_production" href="#security_sasl_plain_production">Use of SASL/PLAIN in production</a></h5>
            <ul>
            <li>SASL/PLAIN should be used only with SSL as transport layer to ensure that clear passwords are not transmitted on the wire without encryption.</li>
            <li>The default implementation of SASL/PLAIN in Kafka specifies usernames and passwords in the JAAS configuration file as shown
                <a href="#security_sasl_plain_brokerconfig">here</a>. From Kafka version 2.0 onwards, you can avoid storing clear passwords on disk
                by configuring your own callback handlers that obtain username and password from an external source using the configuration options
                <code>sasl.server.callback.handler.class</code> and <code>sasl.client.callback.handler.class</code>.</li>
            <li>In production systems, external authentication servers may implement password authentication. From Kafka version 2.0 onwards,
                you can plug in your own callback handlers that use external authentication servers for password verification by configuring
                <code>sasl.server.callback.handler.class</code>.</li>
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_scram" href="#security_sasl_scram">Authentication using SASL/SCRAM</a></h4>
        <p>Salted Challenge Response Authentication Mechanism (SCRAM) is a family of SASL mechanisms that
        addresses the security concerns with traditional mechanisms that perform username/password authentication
        like PLAIN and DIGEST-MD5. The mechanism is defined in <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a>.
        Kafka supports <a href="https://tools.ietf.org/html/rfc7677">SCRAM-SHA-256</a> and SCRAM-SHA-512 which
        can be used with TLS to perform secure authentication. The username is used as the authenticated
        <code>Principal</code> for configuration of ACLs etc. The default SCRAM implementation in Kafka
        stores SCRAM credentials in Zookeeper and is suitable for use in Kafka installations where Zookeeper
        is on a private network. Refer to <a href="#security_sasl_scram_security">Security Considerations</a>
        for more details.</p>
        <ol>
        <li><h5><a id="security_sasl_scram_credentials" href="#security_sasl_scram_credentials">Creating SCRAM Credentials</a></h5>
        <p>The SCRAM implementation in Kafka uses Zookeeper as credential store. Credentials can be created in
        Zookeeper using <tt>kafka-configs.sh</tt>. For each SCRAM mechanism enabled, credentials must be created
        by adding a config with the mechanism name. Credentials for inter-broker communication must be created
        before Kafka brokers are started. Client credentials may be created and updated dynamically and updated
        credentials will be used to authenticate new connections.</p>
        <p>Create SCRAM credentials for user <i>alice</i> with password <i>alice-secret</i>:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]' --entity-type users --entity-name alice
        </code></pre>
        <p>The default iteration count of 4096 is used if iterations are not specified. A random salt is created
        and the SCRAM identity consisting of salt, iterations, StoredKey and ServerKey are stored in Zookeeper.
        See <a href="https://tools.ietf.org/html/rfc5802">RFC 5802</a> for details on SCRAM identity and the individual fields.
        <p>The following examples also require a user <i>admin</i> for inter-broker communication which can be created using:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin
        </code></pre>
        <p>Existing credentials may be listed using the <i>--describe</i> option:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name alice
        </code></pre>
        <p>Credentials may be deleted for one or more SCRAM mechanisms using the <i>--delete</i> option:
        <pre class="line-numbers"><code class="language-bash">    > bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name alice
        </code></pre>
        </li>
        <li><h5><a id="security_sasl_scram_brokerconfig" href="#security_sasl_scram_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre>
    KafkaServer {
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="admin"
        password="admin-secret";
    };</code></pre>
                The properties <tt>username</tt> and <tt>password</tt> in the <tt>KafkaServer</tt> section are used by
                the broker to initiate connections to other brokers. In this example, <i>admin</i> is the user for
                inter-broker communication.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>.</code></pre> For example:
                <pre>
    listeners=SASL_SSL://host.name:port
    security.inter.broker.protocol=SASL_SSL
    sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 (or SCRAM-SHA-512)
    sasl.enabled.mechanisms=SCRAM-SHA-256 (or SCRAM-SHA-512)</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_scram_clientconfig" href="#security_sasl_scram_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
            <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
                The following is an example configuration for a client for the SCRAM mechanisms:
                <pre class="line-numbers"><code class="language-text">   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
        username="alice" \
        password="alice-secret";</code></pre>

                <p>The options <tt>username</tt> and <tt>password</tt> are used by clients to configure
                the user for client connections. In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different user names
                and passwords in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL
    sasl.mechanism=SCRAM-SHA-256 (or SCRAM-SHA-512)</code></pre></li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_scram_security" href="#security_sasl_scram_security">Security Considerations for SASL/SCRAM</a></h5>
            <ul>
            <li>The default implementation of SASL/SCRAM in Kafka stores SCRAM credentials in Zookeeper. This
            is suitable for production use in installations where Zookeeper is secure and on a private network.</li>
            <li>Kafka supports only the strong hash functions SHA-256 and SHA-512 with a minimum iteration count
            of 4096. Strong hash functions combined with strong passwords and high iteration counts protect
            against brute force attacks if Zookeeper security is compromised.</li>
            <li>SCRAM should be used only with TLS-encryption to prevent interception of SCRAM exchanges. This
            protects against dictionary or brute force attacks and against impersonation if Zookeeper is compromised.</li>
            <li>From Kafka version 2.0 onwards, the default SASL/SCRAM credential store may be overridden using custom callback handlers
            by configuring <code>sasl.server.callback.handler.class</code> in installations where Zookeeper is not secure.</li>
            <li>For more details on security considerations, refer to
            <a href="https://tools.ietf.org/html/rfc5802#section-9">RFC 5802</a>.
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_oauthbearer" href="#security_sasl_oauthbearer">Authentication using SASL/OAUTHBEARER</a></h4>
        <p>The <a href="https://tools.ietf.org/html/rfc6749">OAuth 2 Authorization Framework</a> "enables a third-party application to obtain limited access to an HTTP service,
        either on behalf of a resource owner by orchestrating an approval interaction between the resource owner and the HTTP
        service, or by allowing the third-party application to obtain access on its own behalf."  The SASL OAUTHBEARER mechanism
        enables the use of the framework in a SASL (i.e. a non-HTTP) context; it is defined in <a href="https://tools.ietf.org/html/rfc7628">RFC 7628</a>.
        The default OAUTHBEARER implementation in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>
        and is only suitable for use in non-production Kafka installations. Refer to <a href="#security_sasl_oauthbearer_security">Security Considerations</a>
        for more details.</p>
        <ol>
        <li><h5><a id="security_sasl_oauthbearer_brokerconfig" href="#security_sasl_oauthbearer_brokerconfig">Configuring Kafka Brokers</a></h5>
            <ol>
            <li>Add a suitably modified JAAS file similar to the one below to each Kafka broker's config directory, let's call it kafka_server_jaas.conf for this example:
                <pre>
    KafkaServer {
        org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
        unsecuredLoginStringClaim_sub="admin";
    };</code></pre>
                The property <tt>unsecuredLoginStringClaim_sub</tt> in the <tt>KafkaServer</tt> section is used by
                the broker when it initiates connections to other brokers. In this example, <i>admin</i> will appear in the
                subject (<tt>sub</tt>) claim and will be the user for inter-broker communication.</li>
            <li>Pass the JAAS config file location as JVM parameter to each Kafka broker:
                <pre>    -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf</code></pre></li>
            <li>Configure SASL port and SASL mechanisms in server.properties as described <a href="#security_sasl_brokerconfig">here</a>.</code></pre> For example:
                <pre>
    listeners=SASL_SSL://host.name:port (or SASL_PLAINTEXT if non-production)
    security.inter.broker.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
    sasl.mechanism.inter.broker.protocol=OAUTHBEARER
    sasl.enabled.mechanisms=OAUTHBEARER</code></pre></li>
            </ol>
        </li>

        <li><h5><a id="security_sasl_oauthbearer_clientconfig" href="#security_sasl_oauthbearer_clientconfig">Configuring Kafka Clients</a></h5>
            To configure SASL authentication on the clients:
            <ol>
	    <li>Configure the JAAS configuration property for each client in producer.properties or consumer.properties.
                The login module describes how the clients like producer and consumer can connect to the Kafka Broker.
	        The following is an example configuration for a client for the OAUTHBEARER mechanisms:
                <pre class="line-numbers"><code class="language-text">   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        unsecuredLoginStringClaim_sub="alice";</code></pre>

                <p>The option <tt>unsecuredLoginStringClaim_sub</tt> is used by clients to configure
                the subject (<tt>sub</tt>) claim, which determines the user for client connections.
                In this example, clients connect to the broker as user <i>alice</i>.
                Different clients within a JVM may connect as different users by specifying different subject (<tt>sub</tt>)
                claims in <code>sasl.jaas.config</code>.</p>

                <p>JAAS configuration for clients may alternatively be specified as a JVM parameter similar to brokers
                as described <a href="#security_client_staticjaas">here</a>. Clients use the login section named
                <tt>KafkaClient</tt>. This option allows only one user for all client connections from a JVM.</p></li>
            <li>Configure the following properties in producer.properties or consumer.properties:
                <pre>
    security.protocol=SASL_SSL (or SASL_PLAINTEXT if non-production)
    sasl.mechanism=OAUTHBEARER</code></pre></li>
             <li>The default implementation of SASL/OAUTHBEARER depends on the jackson-databind library.
                 Since it's an optional dependency, users have to configure it as a dependency via their build tool.</li>
            </ol>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_unsecured_retrieval" href="#security_sasl_oauthbearer_unsecured_retrieval">Unsecured Token Creation Options for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>The default implementation of SASL/OAUTHBEARER in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>.
            While suitable only for non-production use, it does provide the flexibility to create arbitrary tokens in a DEV or TEST environment.</li>
            <li>Here are the various supported JAAS module options on the client side (and on the broker side if OAUTHBEARER is the inter-broker protocol):
                 <table>
                 <tr>
                 <th>JAAS Module Option for Unsecured Token Creation</th>
                 <th>Documentation</th>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginStringClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>String</tt> claim with the given name and value. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginNumberClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>Number</tt> claim with the given name and value. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginListClaim_&lt;claimname&gt;="value"</tt></td>
                 <td>Creates a <tt>String List</tt> claim with the given name and values parsed
                 from the given value where the first character is taken as the delimiter. For
                 example: <tt>unsecuredLoginListClaim_fubar="|value1|value2"</tt>. Any valid
                 claim name can be specified except '<tt>iat</tt>' and '<tt>exp</tt>' (these are
                 automatically generated).</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginExtension_&lt;extensionname&gt;="value"</tt></td>
                 <td>Creates a <tt>String</tt> extension with the given name and value.
                 For example: <tt>unsecuredLoginExtension_traceId="123"</tt>. A valid extension name
                 is any sequence of lowercase or uppercase alphabet characters. In addition, the "auth" extension name is reserved.
                 A valid extension value is any combination of characters with ASCII codes 1-127.
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginPrincipalClaimName</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt>
                 claim holding the principal name to be something other than '<tt>sub</tt>'.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginLifetimeSeconds</tt></td>
                 <td>Set to an integer value if the token expiration is to be set to something
                 other than the default value of 3600 seconds (which is 1 hour). The
                 '<tt>exp</tt>' claim will be set to reflect the expiration time.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredLoginScopeClaimName</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt> or
                 <tt>String List</tt> claim holding any token scope to be something other than
                 '<tt>scope</tt>'.</td>
                 </tr>
                 </table>
            </li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_unsecured_validation" href="#security_sasl_oauthbearer_unsecured_validation">Unsecured Token Validation Options for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>Here are the various supported JAAS module options on the broker side for <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Token</a> validation:
                 <table>
                 <tr>
                 <th>JAAS Module Option for Unsecured Token Validation</th>
                 <th>Documentation</th>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorPrincipalClaimName="value"</tt></td>
                 <td>Set to a non-empty value if you wish a particular <tt>String</tt> claim
                 holding a principal name to be checked for existence; the default is to check
                 for the existence of the '<tt>sub</tt>' claim.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorScopeClaimName="value"</tt></td>
                 <td>Set to a custom claim name if you wish the name of the <tt>String</tt> or
                 <tt>String List</tt> claim holding any token scope to be something other than
                 '<tt>scope</tt>'.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorRequiredScope="value"</tt></td>
                 <td>Set to a space-delimited list of scope values if you wish the
                 <tt>String/String List</tt> claim holding the token scope to be checked to
                 make sure it contains certain values.</td>
                 </tr>
                 <tr>
                 <td><tt>unsecuredValidatorAllowableClockSkewMs="value"</tt></td>
                 <td>Set to a positive integer value if you wish to allow up to some number of
                 positive milliseconds of clock skew (the default is 0).</td>
                 </tr>
                 </table>
            </li>
            <li>The default unsecured SASL/OAUTHBEARER implementation may be overridden (and must be overridden in production environments)
            using custom login and SASL Server callback handlers.</li>
            <li>For more details on security considerations, refer to <a href="https://tools.ietf.org/html/rfc6749#section-10">RFC 6749, Section 10</a>.</li>
            </ul>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_refresh" href="#security_sasl_oauthbearer_refresh">Token Refresh for SASL/OAUTHBEARER</a></h5>
        Kafka periodically refreshes any token before it expires so that the client can continue to make
        connections to brokers. The parameters that impact how the refresh algorithm
        operates are specified as part of the producer/consumer/broker configuration
        and are as follows. See the documentation for these properties elsewhere for
        details.  The default values are usually reasonable, in which case these
        configuration parameters would not need to be explicitly set.
        <table>
        <tr>
        <th>Producer/Consumer/Broker Configuration Property</th>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.window.factor</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.window.jitter</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.min.period.seconds</tt></td>
        </tr>
        <tr>
        <td><tt>sasl.login.refresh.min.buffer.seconds</tt></td>
        </tr>
        </table>
        </li>
        <li><h5><a id="security_sasl_oauthbearer_prod" href="#security_sasl_oauthbearer_prod">Secure/Production Use of SASL/OAUTHBEARER</a></h5>
        Production use cases will require writing an implementation of
        <tt>org.apache.kafka.common.security.auth.AuthenticateCallbackHandler</tt> that can handle an instance of
        <tt>org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenCallback</tt> and declaring it via either the
        <tt>sasl.login.callback.handler.class</tt> configuration option for a
        non-broker client or via the
        <tt>listener.name.sasl_ssl.oauthbearer.sasl.login.callback.handler.class</tt>
        configuration option for brokers (when SASL/OAUTHBEARER is the inter-broker
        protocol).
        <p>
        Production use cases will also require writing an implementation of
        <tt>org.apache.kafka.common.security.auth.AuthenticateCallbackHandler</tt> that can handle an instance of
        <tt>org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallback</tt> and declaring it via the
        <tt>listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class</tt>
        broker configuration option.
        </li>
        <li><h5><a id="security_sasl_oauthbearer_security" href="#security_sasl_oauthbearer_security">Security Considerations for SASL/OAUTHBEARER</a></h5>
            <ul>
            <li>The default implementation of SASL/OAUTHBEARER in Kafka creates and validates <a href="https://tools.ietf.org/html/rfc7515#appendix-A.5">Unsecured JSON Web Tokens</a>.
            This is suitable only for non-production use.</li>
            <li>OAUTHBEARER should be used in production enviromnments only with TLS-encryption to prevent interception of tokens.</li>
            <li>The default unsecured SASL/OAUTHBEARER implementation may be overridden (and must be overridden in production environments)
            using custom login and SASL Server callback handlers as described above.</li>
            <li>For more details on OAuth 2 security considerations in general, refer to <a href="https://tools.ietf.org/html/rfc6749#section-10">RFC 6749, Section 10</a>.</li>
            </ul>
        </li>
        </ol>
    </li>

    <li><h4><a id="security_sasl_multimechanism" href="#security_sasl_multimechanism">Enabling multiple SASL mechanisms in a broker</a></h4>
        <ol>
        <li>Specify configuration for the login modules of all enabled mechanisms in the <tt>KafkaServer</tt> section of the JAAS config file. For example:
            <pre>
        KafkaServer {
            com.sun.security.auth.module.Krb5LoginModule required
            useKeyTab=true
            storeKey=true
            keyTab="/etc/security/keytabs/kafka_server.keytab"
            principal="kafka/kafka1.hostname.com@EXAMPLE.COM";

            org.apache.kafka.common.security.plain.PlainLoginModule required
            username="admin"
            password="admin-secret"
            user_admin="admin-secret"
            user_alice="alice-secret";
        };</code></pre></li>
        <li>Enable the SASL mechanisms in server.properties: <pre>    sasl.enabled.mechanisms=GSSAPI,PLAIN,SCRAM-SHA-256,SCRAM-SHA-512,OAUTHBEARER</code></pre></li>
        <li>Specify the SASL security protocol and mechanism for inter-broker communication in server.properties if required:
            <pre>
    security.inter.broker.protocol=SASL_PLAINTEXT (or SASL_SSL)
    sasl.mechanism.inter.broker.protocol=GSSAPI (or one of the other enabled mechanisms)</code></pre></li>
        <li>Follow the mechanism-specific steps in <a href="#security_sasl_kerberos_brokerconfig">GSSAPI (Kerberos)</a>,
            <a href="#security_sasl_plain_brokerconfig">PLAIN</a>,
            <a href="#security_sasl_scram_brokerconfig">SCRAM</a> and <a href="#security_sasl_oauthbearer_brokerconfig">OAUTHBEARER</a>
            to configure SASL for the enabled mechanisms.</li>
        </ol>
    </li>
    <li><h4><a id="saslmechanism_rolling_upgrade" href="#saslmechanism_rolling_upgrade">Modifying SASL mechanism in a Running Cluster</a></h4>
        <p>SASL mechanism can be modified in a running cluster using the following sequence:</p>
        <ol>
        <li>Enable new SASL mechanism by adding the mechanism to <tt>sasl.enabled.mechanisms</tt> in server.properties for each broker. Update JAAS config file to include both
            mechanisms as described <a href="#security_sasl_multimechanism">here</a>. Incrementally bounce the cluster nodes.</li>
        <li>Restart clients using the new mechanism.</li>
        <li>To change the mechanism of inter-broker communication (if this is required), set <tt>sasl.mechanism.inter.broker.protocol</tt> in server.properties to the new mechanism and
            incrementally bounce the cluster again.</li>
        <li>To remove old mechanism (if this is required), remove the old mechanism from <tt>sasl.enabled.mechanisms</tt> in server.properties and remove the entries for the
            old mechanism from JAAS config file. Incrementally bounce the cluster again.</li>
        </ol>
    </li>

    <li><h4><a id="security_delegation_token" href="#security_delegation_token">Authentication using Delegation Tokens</a></h4>
        <p>Delegation token based authentication is a lightweight authentication mechanism to complement existing SASL/SSL
            methods. Delegation tokens are shared secrets between kafka brokers and clients. Delegation tokens will help processing
            frameworks to distribute the workload to available workers in a secure environment without the added cost of distributing
            Kerberos TGT/keytabs or keystores when 2-way SSL is used. See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-48+Delegation+token+support+for+Kafka">KIP-48</a>
            for more details.</p>

        <p>Typical steps for delegation token usage are:</p>
        <ol>
        <li>User authenticates with the Kafka cluster via SASL or SSL, and obtains a delegation token. This can be done
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -