ACL in Kafka

This article describes the basic steps for working with ACL authorization in Kafka.

Access Control List (ACL) — an access control list that defines who can access topics or groups in Kafka and what operations are allowed or denied to be performed by a user or group of users.

Kafka has AclAuthorizer — a built-in ACL-based authorizer.

ACLs for Kafka are stored in ZooKeeper.

Enable ACL in Kafka

NOTE

Before configuring authorization for an ADS cluster, authentication must be enabled and configured, and users must be created using one of the following protocols:

To enable and use AclAuthorizer, do the following:

  1. On the host with the Kafka broker enter the command:

    $ sudo vim /usr/lib/kafka/config/server.properties
  2. Add new parameters to the file:

    authorizer.class.name=kafka.security.authorizer.AclAuthorizer
    allow.everyone.if.no.acl.found=true

    where:

    • authorizer.class.name — setting the authorizer class.

    • allow.everyone.if.no.acl.found is a parameter that determines whether to enable authorization only for topics listed in the ACL.

  3. After entering the data, restart the Kafka service on all hosts with the Kafka broker and check the status of the server. To do this, enter the commands one by one:

    $ sudo systemctl restart kafka
    $ sudo systemctl status kafka

    As a result, information about the status of the Kafka service is displayed:

  kafka.service - Apache Kafka
   Loaded: loaded (/usr/lib/systemd/system/kafka.service; enabled; vendor preset: disabled)
  Drop-In: /etc/systemd/system/kafka.service.d
           └─custom.conf
   Active: active (running) since Wed 2022-10-12 12:48:49 UTC; 52s ago
  Process: 8782 ExecStop=/usr/lib/kafka/bin/kafka stop (code=exited, status=0/SUCCESS)
  Process: 8857 ExecStart=/usr/lib/kafka/bin/kafka start (code=exited, status=0/SUCCESS)
  Process: 8854 ExecStartPre=/bin/chown kafka:kafka /run/kafka (code=exited, status=0/SUCCESS)
  Process: 8852 ExecStartPre=/bin/mkdir -p /run/kafka (code=exited, status=0/SUCCESS)
 Main PID: 8860 (java)
    Tasks: 82
   Memory: 298.6M
   CGroup: /system.slice/kafka.service
           └─8860 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOcc...

Oct 12 12:48:49 sov-ads-test-1.ru-central1.internal systemd[1]: Starting Apache Kafka...
Oct 12 12:48:49 sov-ads-test-1.ru-central1.internal systemd[1]: Started Apache Kafka.
CAUTION

Restarting the Kafka service using the ADCM user interface tools may result in the set value of authorizer.class.name being removed.

Prepare the authentication environment to create ACLs

SASL PLAINTEXT

To create the user environment specified during authentication using the SASL PLAINTEXT protocol, do the following:

  1. On the host with the Kafka broker, enter the command:

    $ sudo vim /etc/kafka/conf/client.properties
  2. In the file that opens, enable (uncomment) the authentication mechanism and fill in the data for the created user:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
# Uncomment and set necessary username/password
#sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule requi$
#  username="username" \
#  password=password_for_username;

For more information about working with users in the SASL PLAINTEXT authentication environment, see the article Use SASL PLAINTEXT in Kafka.

Kerberos SASL

To create the environment for the user specified during authentication using the Kerberos SASL protocol, do the following:

  1. Create a JAAS (Java Authentication and Authorization Service) file for the user principal that manages the ACLs. Run command:

    $ sudo vim /tmp/client.jaas
  2. Write data to file:

    KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useTicketCache=true;
    };
  3. Create a ticket for user user:

    $ kinit -p user@ADS-KAFKA.LOCAL
  4. Check ticket:

    $ klist
    Ticket cache: FILE:/tmp/krb5cc_1000
    Default principal: user@ADS-KAFKA.LOCAL
    
    Valid starting Expires Service principal
    08/10/2022 20:44:12 08/11/2022 20:44:12 krbtgt/ADS-KAFKA.LOCAL@ADS-KAFKA.LOCAL
  5. Export the generated client.jaas file as a JVM option for the given user using the KAFKA_OPTS environment variable:

$ export  KAFKA_OPTS="-Djava.security.auth.login.config=/tmp/client.jaas"

For more information about working with users in a Kerberos SASL authentication environment, see the article Use MIT Kerberos in Kafka.

ACL configuration

Working with the ACL-based authorizer occurs by running the kafka-acls.sh script on the command line from any cluster host on which the Kafka service is installed.

NOTE

To view all options for the kafka-acls.sh script, run the command:

$ /usr/lib/kafka/bin/kafka-acls.sh

Assign rights and prohibitions to individual actions in Kafka for users

Assignment of rights or prohibitions to individual actions to the user occurs by running the kafka-acls.sh script with the --operation option, and specifying the operation to which access is given.

Permissions are assigned using the --add option.

The --allow-principal option enables the action, and --deny-principal disables the action.

Removing a user from the ACL is done with the command option --remove.

The main operations for which user rights can be set are:

  • Describe — view topic or group metadata.

  • DescribeConfigs — view topic configuration.

  • Alter — change a topic or a group.

  • IdempotentWrite — write to a topic for idempotent and transactional APIs.

  • Read — the right to read from a topic or from a group.

  • Delete — delete a topic or a group.

  • Create — create a topic or a group.

  • All — all rights.

  • Write —  write to a topic or a group.

  • AlterConfigs — change the topic configuration.

NOTE

For users authenticated using the Kerberos SASL or LDAP protocol, the ACLs use the username without the authentication realm.

In order to allow the username user to perform the WRITE operation on the test-topic topic, on the host with the Kafka broker, enter the command:

$ /usr/lib/kafka/bin/kafka-acls.sh  --bootstrap-server localhost:9092  --command-config /etc/kafka/conf/client.properties  --add --allow-principal User:username  --operation WRITE --test-topic

In order to prevent the username1 user from performing the WRITE operation on the test-topic topic, on the host with the Kafka broker, enter the command:

$ /usr/lib/kafka/bin/kafka-acls.sh  --bootstrap-server localhost:9092  --command-config /etc/kafka/conf/client.properties  --add --deny-principal User:username1  --operation WRITE --test-topic

As a result of each addition to the list, a message appears about the successful addition of rights or a ban and an ACL list with current user rights for the topic:

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:username, host=*, operation=WRITE, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:username, host=*, operation=CREATE, permissionType=ALLOW)
	(principal=User:username1, host=*, operation=CREATE, permissionType=DENY)

Assign producer and consumer rights to a user

It is possible to assign several producer rights to the user at once — create, write, view metadata (describe).

To assign producer rights to the writer user for the test-topic topic, enter the command on the host with the Kafka broker:

$ /usr/lib/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /etc/kafka/conf/client.properties  --add --allow-principal User:writer --producer --topic test-topic

As a result, a message appears about the successful addition of producer rights and an ACL list with current user rights for the topic:

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:writer, host=*, operation=CREATE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=WRITE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=DESCRIBE, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:writer, host=*, operation=WRITE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=CREATE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=DESCRIBE, permissionType=ALLOW)

It is also possible to assign several consumer rights at once — read, view metadata (describe).

To assign consumer rights to the reader user for the test-topic topic, enter the command on the host with the Kafka broker:

$ /usr/lib/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /etc/kafka/conf/client.properties  --add --allow-principal User:reader --consumer --topic test-topic --group 1
NOTE
When assigning consumer rights, you must specify both the topic name (topic) and the name of the consumer group (group), even if there is only one consumer.

As a result, a message appears about the successful addition of consumer rights and an ACL list with current user rights for the test-topic topic, including those created earlier for the writer user:

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:reader, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=READ, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
	(principal=User:writer, host=*, operation=WRITE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=CREATE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=READ, permissionType=ALLOW)

It also displays data about the created rights for the consumer group:

Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=1, patternType=LITERAL)`:
 	(principal=User:reader, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=READ, permissionType=ALLOW)

View ACLs

Viewing ACLs is done with the --list option. In order to view the list of permissions for the test-topic topic, enter the command:

$ /usr/lib/kafka/bin/kafka-acls.sh --bootstrap-server localhost:9092 --command-config /etc/kafka/conf/client.properties --list --topic test-topic

The result is a listing of all assigned permissions for the test-topic topic:

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=test-topic, patternType=LITERAL)`:
 	(principal=User:writer, host=*, operation=WRITE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=CREATE, permissionType=ALLOW)
	(principal=User:writer, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=DESCRIBE, permissionType=ALLOW)
	(principal=User:reader, host=*, operation=READ, permissionType=ALLOW)

To view the list of rights for all topics, use the same command without specifying the topic name.

How an authorized user works in Kafka

After setting up ACLs for users, work in Kafka proceeds according to the steps described in the articles:

Any action by an unauthorized user will cause an error. For example, attempting to write messages to the test-topic topic for a user who does not have WRITE permission in the ACL for that topic will result in the following errors:

[2022-10-12 13:38:57,301] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 9 : {test-topic=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient)
[2022-10-12 13:38:57,302] ERROR [Producer clientId=console-producer] Topic authorization failed for topics [test-topic] (org.apache.kafka.clients.Metadata)
[2022-10-12 13:38:57,302] ERROR Error when sending message to topic test-topic with key: null, value: 1 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [test-topic]
Found a mistake? Seleсt text and press Ctrl+Enter to report it