Chapter 6. Securing access to Kafka

Procedure

1. Generate TLS certificates for all Kafka brokers in your cluster. The certificates should have their advertised and bootstrap addresses in their Common Name or Subject Alternative Name.

2. Edit the Kafka configuration properties file on all cluster nodes for the following:

   • Change the listener.security.protocol.map field to specify the SSL protocol for the listener where you want to use TLS encryption.
   • Set the ssl.keystore.location option to the path to the JKS keystore with the broker certificate.

   • Set the ssl.keystore.password option to the password you used to protect the keystore.

     For example:

     listeners=UNENCRYPTED://:9092,ENCRYPTED://:9093,REPLICATION://:9094
     listener.security.protocol.map=UNENCRYPTED:PLAINTEXT,ENCRYPTED:SSL,REPLICATION:PLAINTEXT
     ssl.keystore.location=/path/to/keystore/server-1.jks
     ssl.keystore.password=123456

3. (Re)start the Kafka brokers

Procedure

1. Prepare a JKS (Java Keystore ) truststore containing the public key of the CA (Certification Authority) used to sign the user certificates.

2. Edit the Kafka configuration properties file on all cluster nodes as follows:

   • Specify the path to the JKS truststore using the ssl.truststore.location property.
   • If the truststore is password-protected, set the password using ssl.truststore.password property.

   • Set the ssl.client.auth property to required.

     TLS client authentication configuration

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

3. (Re)start the Kafka brokers.

Procedure

1. Edit or create the /opt/kafka/config/jaas.conf JAAS configuration file to enable the PlainLoginModule and specify the allowed usernames and passwords.

   Make sure this file is the same on all Kafka brokers.

   JAAS configuration

   KafkaServer {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       user_admin="123456"
       user_user1="123456"
       user_user2="123456";
   };

2. Edit the Kafka configuration properties file on all cluster nodes as follows:

   • Enable SASL PLAIN authentication on specific listeners using the listener.security.protocol.map property. Specify SASL_PLAINTEXT or SASL_SSL.

   • Set the sasl.enabled.mechanisms property to PLAIN.

     SASL plain configuration

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

3. (Re)start the Kafka brokers using the KAFKA_OPTS environment variable to pass the JAAS configuration to Kafka brokers:

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

Procedure

1. Edit or create the /opt/kafka/config/jaas.conf JAAS configuration file to enable the ScramLoginModule.

   Make sure this file is the same on all Kafka brokers.

   JAAS configuration

   KafkaServer {
       org.apache.kafka.common.security.scram.ScramLoginModule required;
   };

2. Edit the Kafka configuration properties file on all cluster nodes as follows:

   • Enable SASL SCRAM authentication on specific listeners using the listener.security.protocol.map property. Specify SASL_PLAINTEXT or SASL_SSL.

   • Set the sasl.enabled.mechanisms option to SCRAM-SHA-256 or SCRAM-SHA-512.

     For example:

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

3. (Re)start the Kafka brokers using the KAFKA_OPTS environment variable to pass the JAAS configuration to Kafka brokers.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

Procedure

1. Specify an inter-broker SASL mechanism in the Kafka configuration using the sasl.mechanism.inter.broker.protocol property.

   Inter-broker SASL mechanism

   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

2. (Optional) If you are using a SCRAM mechanism, register SCRAM credentials on the Kafka cluster by adding SCRAM users.

   This ensures that the credentials for authentication are updated for bootstrapping before the Kafka cluster is running.

3. Specify the username and password for inter-broker communication in the KafkaServer JAAS context using the username and password fields.

   Inter-broker JAAS context

   KafkaServer {
       org.apache.kafka.common.security.plain.ScramLoginModule required
       username="admin"
       password="123456"
       # ...
   };

1. Service principals
2. Kafka brokers to use the Kerberos login
3. ZooKeeper to use Kerberos login
4. Producer and consumer clients to access Kafka using Kerberos authentication

1. Create the service principals, and keytabs that store the principal keys, through the Kerberos KDC.

   Make sure the domain name in the Kerberos principal is in uppercase.

   For example:

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     The ZooKeeper service principal must have the same hostname as the zookeeper.connect configuration in the Kafka config/server.properties file:

     zookeeper.connect=node1.example.redhat.com:2181

     If the hostname is not the same, localhost is used and authentication will fail.

2. Create a directory on the host and add the keytab files:

   For example:

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. Ensure the kafka user can access the directory:

   chown kafka:kafka -R /opt/kafka/krb5

1. Create or modify the opt/kafka/config/jaas.conf file to support ZooKeeper client and server operations:

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 1
       storeKey=true 2
       useTicketCache=false 3
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 4
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 5
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   1

   Set to true to get the principal key from the keytab.

   2

   Set to true to store the principal key.

   3

   Set to true to obtain the Ticket Granting Ticket (TGT) from the ticket cache.

   4

   The keyTab property points to the location of the keytab file copied from the Kerberos KDC. The location and file must be readable by the kafka user.

   5

   The principal property is configured to match the fully-qualified principal name created on the KDC host, which follows the format SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME.

2. Edit opt/kafka/config/zookeeper.properties to use the updated JAAS configuration:

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 1
   kerberos.removeHostFromPrincipal=false 2
   kerberos.removeRealmFromPrincipal=false 3
   quorum.auth.enableSasl=true 4
   quorum.auth.learnerRequireSasl=true 5
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 6
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 7
   quorum.cnxn.threads.size=20

   1

   Controls the frequency for login renewal in milliseconds, which can be adjusted to suit ticket renewal intervals. Default is one hour.

   2

   Dictates whether the hostname is used as part of the login principal name. If using a single keytab for all nodes in the cluster, this is set to true. However, it is recommended to generate a separate keytab and fully-qualified principal for each broker host for troubleshooting.

   3

   Controls whether the realm name is stripped from the principal name for Kerberos negotiations. It is recommended that this setting is set as false.

   4

   Enables SASL authentication mechanisms for the ZooKeeper server and client.

   5

   The RequireSasl properties controls whether SASL authentication is required for quorum events, such as master elections.

   6

   The loginContext properties identify the name of the login context in the JAAS configuration used for authentication configuration of the specified component. The loginContext names correspond to the names of the relevant sections in the opt/kafka/config/jaas.conf file.

   7

   Controls the naming convention to be used to form the principal name used for identification. The placeholder _HOST is automatically resolved to the hostnames defined by the server.1 properties at runtime.

3. Start ZooKeeper with JVM parameters to specify the Kerberos login configuration:

   su - kafka
   export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   If you are not using the default service name (zookeeper), add the name using the -Dzookeeper.sasl.client.username=NAME parameter.

   Note

   If you are using the /etc/krb5.conf location, you do not need to specify -Djava.security.krb5.conf=/etc/krb5.conf when starting ZooKeeper, Kafka, or the Kafka producer and consumer.

1. Modify the opt/kafka/config/jaas.conf file with the following elements:

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. Configure each broker in the Kafka cluster by modifying the listener configuration in the config/server.properties file so the listeners use the SASL/GSSAPI login.

   Add the SASL protocol to the map of security protocols for the listener, and remove any unwanted protocols.

   For example:

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 1
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2
   # ..
   sasl.enabled.mechanisms=GSSAPI 3
   sasl.mechanism.inter.broker.protocol=GSSAPI 4
   sasl.kerberos.service.name=kafka 5
   ...

   1

   Two listeners are configured: a secure listener for general-purpose communications with clients (supporting TLS for communications), and a replication listener for inter-broker communications.

   2

   For TLS-enabled listeners, the protocol name is SASL_PLAINTEXT. For non-TLS-enabled connectors, the protocol name is SASL_PLAINTEXT. If SSL is not required, you can remove the ssl.* properties.

   3

   SASL mechanism for Kerberos authentication is GSSAPI.

   4

   Kerberos authentication for inter-broker communication.

   5

   The name of the service used for authentication requests is specified to distinguish it from other services that may also be using the same Kerberos configuration.

3. Start the Kafka broker, with JVM parameters to specify the Kerberos login configuration:

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   If the broker and ZooKeeper cluster were previously configured and working with a non-Kerberos-based authentication system, it is possible to start the ZooKeeper and broker cluster and check for configuration errors in the logs.

   After starting the broker and Zookeeper instances, the cluster is now configured for Kerberos authentication.

1. Add the Kerberos configuration to the producer or consumer configuration file.

   For example:

   /opt/kafka/config/producer.properties

   # ...
   sasl.mechanism=GSSAPI 1
   security.protocol=SASL_PLAINTEXT 2
   sasl.kerberos.service.name=kafka 3
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 4
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   1

   Configuration for Kerberos (GSSAPI) authentication.

   2

   Kerberos uses the SASL plaintext (username/password) security protocol.

   3

   The service principal (user) for Kafka that was configured in the Kerberos KDC.

   4

   Configuration for the JAAS using the same properties defined in jaas.conf.

   /opt/kafka/config/consumer.properties

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

2. Run the clients to verify that you can send and receive messages from the Kafka brokers.

   Producer client:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   Consumer client:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

Enabling SASL DIGEST-MD5 authentication

1. On all Kafka broker nodes, create or edit the /opt/kafka/config/jaas.conf JAAS configuration file and add the following context:

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="<Username>"
       password="<Password>";
   };

   The username and password should be the same as configured in ZooKeeper.

   Following example shows the Client context:

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="kafka"
       password="123456";
   };

2. Restart all Kafka broker nodes one by one. To pass the JAAS configuration to Kafka brokers, use the KAFKA_OPTS environment variable.

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   For information on restarting brokers in a multi-node cluster, see Section 4.3, “Performing a graceful rolling restart of Kafka brokers”.

Procedure

1. Edit the Kafka configuration properties file to set the zookeeper.set.acl field to true on all cluster nodes.

   zookeeper.set.acl=true

2. Start the Kafka brokers.

Enabling the ZooKeeper ACLs

1. Edit the Kafka configuration properties file to set the zookeeper.set.acl field to true on all cluster nodes.

   zookeeper.set.acl=true

2. Restart all Kafka brokers one by one.

   For information on restarting brokers in a multi-node cluster, see Section 4.3, “Performing a graceful rolling restart of Kafka brokers”.

3. Set the ACLs on all existing ZooKeeper znodes using the zookeeper-security-migration.sh tool.

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=<ZooKeeperURL>
   exit

   For example:

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=zoo1.my-domain.com:2181
   exit

1. Configure an OAuth 2.0 authorization server for the Streams for Apache Kafka cluster and Kafka clients
2. Deploy or update the Kafka cluster with Kafka broker listeners configured to use OAuth 2.0
3. Update your Java-based Kafka clients to use OAuth 2.0

Procedure

1. Access the Red Hat Single Sign-On Admin Console or use the Red Hat Single Sign-On Admin CLI to enable Authorization Services for the Kafka broker client you created when setting up OAuth 2.0 authentication.
2. Use Authorization Services to define resources, authorization scopes, policies, and permissions for the client.
3. Bind the permissions to users and clients by assigning them roles and groups.

4. Configure the Kafka brokers to use Red Hat Single Sign-On authorization.

   Add the following to the Kafka server.properties configuration file to install the authorizer in Kafka:

   authorizer.class.name=io.strimzi.kafka.oauth.server.authorizer.KeycloakAuthorizer
   principal.builder.class=io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder

5. Add configuration for the Kafka brokers to access the authorization server and Authorization Services.

   Here we show example configuration added as additional properties to server.properties, but you can also define them as environment variables using capitalized or upper-case naming conventions.

   strimzi.authorization.token.endpoint.uri="https://<auth_server_address>/auth/realms/REALM-NAME/protocol/openid-connect/token" 1
   strimzi.authorization.client.id="kafka" 2

   1

   The OAuth 2.0 token endpoint URL to Red Hat Single Sign-On. For production, always use https:// urls.

   2

   The client ID of the OAuth 2.0 client definition in Red Hat Single Sign-On that has Authorization Services enabled. Typically, kafka is used as the ID.

6. (Optional) Add configuration for specific Kafka clusters.

   For example:

   strimzi.authorization.kafka.cluster.name="kafka-cluster" 1

   1

   The name of a specific Kafka cluster. Names are used to target permissions, making it possible to manage multiple clusters within the same Red Hat Single Sign-On realm. The default value is kafka-cluster.

7. (Optional) Delegate to simple authorization:

   strimzi.authorization.delegate.to.kafka.acl="true" 1

   1

   Delegate authorization to Kafka AclAuthorizer if access is denied by Red Hat Single Sign-On Authorization Services policies. The default is false.

8. (Optional) Add configuration for TLS connection to the authorization server.

   For example:

   strimzi.authorization.ssl.truststore.location=<path_to_truststore> 1
   strimzi.authorization.ssl.truststore.password=<my_truststore_password> 2
   strimzi.authorization.ssl.truststore.type=JKS 3
   strimzi.authorization.ssl.secure.random.implementation=SHA1PRNG 4
   strimzi.authorization.ssl.endpoint.identification.algorithm=HTTPS 5

   1

   The path to the truststore that contain the certificates.

   2

   The password for the truststore.

   3

   The truststore type. If not set, the default Java keystore type is used.

   4

   Random number generator implementation. If not set, the Java platform SDK default is used.

   5

   Hostname verification. If set to an empty string, the hostname verification is turned off. If not set, the default value is HTTPS, which enforces hostname verification for server certificates.

9. (Optional) Configure the refresh of grants from the authorization server. The grants refresh job works by enumerating the active tokens and requesting the latest grants for each.

   For example:

   strimzi.authorization.grants.refresh.period.seconds="120" 1
   strimzi.authorization.grants.refresh.pool.size="10" 2
   strimzi.authorization.grants.max.idle.time.seconds="300" 3
   strimzi.authorization.grants.gc.period.seconds="300" 4
   strimzi.authorization.reuse.grants="false" 5

   1

   Specifies how often the list of grants from the authorization server is refreshed (once per minute by default). To turn grants refresh off for debugging purposes, set to "0".

   2

   Specifies the size of the thread pool (the degree of parallelism) used by the grants refresh job. The default value is "5".

   3

   The time, in seconds, after which an idle grant in the cache can be evicted. The default value is 300.

   4

   The time, in seconds, between consecutive runs of a job that cleans stale grants from the cache. The default value is 300.

   5

   Controls whether the latest grants are fetched for a new session. When disabled, grants are retrieved from Red Hat Single Sign-On and cached for the user. The default value is true.

10. (Optional) Configure network timeouts when communicating with the authorization server.

    For example:

    strimzi.authorization.connect.timeout.seconds="60" 1
    strimzi.authorization.read.timeout.seconds="60" 2
    strimzi.authorization.http.retries="2" 3

    1

    The connect timeout in seconds when connecting to the Red Hat Single Sign-On token endpoint. The default value is 60.

    2

    The read timeout in seconds when connecting to the Red Hat Single Sign-On token endpoint. The default value is 60.

    3

    The maximum number of times to retry (without pausing) a failed HTTP request to the authorization server. The default value is 0, meaning that no retries are performed. To use this option effectively, consider reducing the timeout times for the strimzi.authorization.connect.timeout.seconds and strimzi.authorization.read.timeout.seconds options. However, note that retries may prevent the current worker thread from being available to other requests, and if too many requests stall, it could make the Kafka broker unresponsive.

11. (Optional) Enable OAuth 2.0 metrics for token validation and authorization:

    oauth.enable.metrics="true" 1

    1

    Controls whether to enable or disable OAuth metrics. The default value is false.

12. (Optional) Remove the Accept header from requests:

    oauth.include.accept.header="false" 1

    1

    Set to false if including the header is causing issues when communicating with the authorization server. The default value is true.

13. Verify the configured permissions by accessing Kafka brokers as clients or users with specific roles, making sure they have the necessary access, or do not have the access they are not supposed to have.

Procedure

1. Write the OPA policies required for authorizing client requests to perform operations on the Kafka brokers.

   See Defining OPA policies.

   Now configure the Kafka brokers to use OPA.

2. Install the OPA authorizer plugin for Kafka.

   See Connecting to the OPA.

   Make sure that the plugin files are included in the Kafka classpath.

3. Add the following to the Kafka server.properties configuration file to enable the OPA plugin:

   authorizer.class.name: com.bisnode.kafka.authorization.OpaAuthorizer

4. Add further configuration to server.properties for the Kafka brokers to access the OPA policy engine and policies.

   For example:

   opa.authorizer.url=https://OPA-ADDRESS/allow 1
   opa.authorizer.allow.on.error=false 2
   opa.authorizer.cache.initial.capacity=50000 3
   opa.authorizer.cache.maximum.size=50000 4
   opa.authorizer.cache.expire.after.seconds=600000 5
   super.users=User:alice;User:bob 6

   1

   (Required) The OAuth 2.0 token endpoint URL for the policy the authorizer plugin will query. In this example, the policy is called allow.

   2

   Flag to specify whether a client is allowed or denied access by default if the authorizer plugin fails to connect with the OPA policy engine.

   3

   Initial capacity in bytes of the local cache. The cache is used so that the plugin does not have to query the OPA policy engine for every request.

   4

   Maximum capacity in bytes of the local cache.

   5

   Time in milliseconds that the local cache is refreshed by reloading from the OPA policy engine.

   6

   A list of user principals treated as super users, so that they are always allowed without querying the Open Policy Agent policy.

   Refer to the Open Policy Agent website for information on authentication and authorization options.

5. Verify the configured permissions by accessing Kafka brokers using clients that have and do not have the correct authorization.