Chapter 4. Configuring Kafka

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

Procedure

1. Edit the /opt/kafka/config/server.properties Kafka configuration file to use the AclAuthorizer.

   authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

2. (Re)start the Kafka brokers.

Procedure

1. Run kafka-acls.sh with the --add option.

   Examples:

   • Allow user1 and user2 access to read from myTopic using the MyConsumerGroup consumer group.

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Read --topic myTopic --allow-principal User:user1 --allow-principal User:user2
     
     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Describe --topic myTopic --allow-principal User:user1 --allow-principal User:user2
     
     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Read --operation Describe --group MyConsumerGroup --allow-principal User:user1 --allow-principal User:user2

   • Deny user1 access to read myTopic from IP address host 127.0.0.1.

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Describe --operation Read --topic myTopic --group MyConsumerGroup --deny-principal User:user1 --deny-host 127.0.0.1

   • Add user1 as the consumer of myTopic with MyConsumerGroup.

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --consumer --topic myTopic --group MyConsumerGroup --allow-principal User:user1

Procedure

1. Edit the /opt/kafka/config/server.properties Kafka configuration 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 /opt/kafka/config/server.properties Kafka configuration 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.

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

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 /opt/kafka/config/server.properties Kafka configuration 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 truststore containing the public key of the certification authority used to sign the user certificates.

2. Edit the /opt/kafka/config/server.properties Kafka configuration file on all cluster nodes for the following:

   • Set the ssl.truststore.location option to the path to the JKS truststore with the certification authority of the user certificates.
   • Set the ssl.truststore.password option to the password you used to protect the truststore.

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

     For example:

     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. This file should contain all your users and their passwords. Make sure this file is the same on all Kafka brokers.

   For example:

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

2. Edit the /opt/kafka/config/server.properties Kafka configuration file on all cluster nodes for the following:

   • Change the listener.security.protocol.map field to specify the SASL_PLAINTEXT or SASL_SSL protocol for the listener where you want to use SASL PLAIN authentication.

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

     For example:

     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. Enable the ScramLoginModule for the KafkaServer context. Make sure this file is the same on all Kafka brokers.

   For example:

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

2. Edit the /opt/kafka/config/server.properties Kafka configuration file on all cluster nodes for the following:

   • Change the listener.security.protocol.map field to specify the SASL_PLAINTEXT or SASL_SSL protocol for the listener where you want to use SASL SCRAM authentication.

   • 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

1. Configure an OAuth 2.0 authorization server for the AMQ Streams 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.KeycloakRBACAuthorizer
   principal.builder.class=io.strimzi.kafka.oauth.server.authorizer.JwtKafkaPrincipalBuilder

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.

   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.

   For example:

   strimzi.authorization.delegate.to.kafka.acl="false" 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

   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".

10. 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.

Procedure

1. Switch to the kafka user:

   su - kafka

2. List all the broker loggers for a broker by using the kafka-configs.sh tool:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server BOOTSTRAP-ADDRESS --describe --entity-type broker-loggers --entity-name BROKER-ID

   For example, for broker 0:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0

   This returns the logging level for each logger: TRACE, DEBUG, INFO, WARN, ERROR, or FATAL. For example:

   #...
   kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={}
   kafka.log.TimeIndex=INFO sensitive=false synonyms={}

3. Change the logging level for one or more broker loggers. Use the --alter and --add-config options and specify each logger and its level as a comma-separated list in double quotes.

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server BOOTSTRAP-ADDRESS --alter --add-config "LOGGER-ONE=NEW-LEVEL,LOGGER-TWO=NEW-LEVEL" --entity-type broker-loggers --entity-name BROKER-ID

   For example, for broker 0:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0

   If successful this returns:

   Completed updating config for broker: 0.