ACL в Kafka

Ольга Семме

В данной статье описаны основные действия для работы с авторизацей AСL в Kafka.

Access Control List (ACL) — список управления доступом, который определяет, кто может получать доступ к топикам или группам в Kafka и какие именно операции разрешено или запрещено выполнять пользователю или группе пользователей.

Kafka имеет AclAuthorizer — встроенный авторизатор на базе ACL.

Cписки авторизации ACL для Kafka хранятся в ZooKeeper.

Включение ACL в Kafka

ПРИМЕЧАНИЕ

Перед настройкой авторизации для кластера ADS должна быть включена и настроена аутентификация и созданы пользователи при помощи одного из протоколов:

Чтобы включить и использовать AclAuthorizer, необходимо выполнить:

  1. На хосте с брокером Kafka ввести команду:

    $ sudo vim /usr/lib/kafka/config/server.properties

  2. Добавить в файл новые параметры :

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

    где:

    • authorizer.class.name — установка класса авторизатора.

    • allow.everyone.if.no.acl.found — параметр, определяющий, включать ли авторизацию только для топиков, внесенных в список ACL.

  3. После внесения данных рестартовать сервис Kafka на всех хостах с брокером Kafka и проверить статус сервера. Для этого поочередно ввести команды:

    $ sudo systemctl restart kafka
$ sudo systemctl status kafka

    В результате выводится информация о статусе сервиса Kafka:

  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.
ВНИМАНИЕ

Перезапуск сервиса Kafka при помощи средств пользовательского интерфейса ADCM может привести к удалению установленного значения authorizer.class.name.

Подготовка среды аутентификации к созданию списков авторизации ACL

SASL PLAINTEXT

Для создания среды пользователя, указанного при аутентификации по протоколу SASL PLAINTEXT, необходимо:

  1. На хосте с брокером Kafka ввести команду:

    $ sudo vim /etc/kafka/conf/client.properties

  2. В открывшемся файле включить (раскомментировать) механизм аутентификации и заполнить данные для созданного пользователя:

    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;

Дополнительные сведения о работе с пользователями в среде аутентификации SASL PLAINTEXT можно найти в статье Использование SASL PLAINTEXT в Kafka.

Kerberos SASL

Для создания среды пользователя, указанного при аутентификации по протоколу Kerberos SASL, необходимо:

  1. Cоздать JAAS-файл (Java Authentication and Authorization Service) для принципала user, управляющего списками AСL. Выполнить команду:

    $ sudo vim /tmp/client.jaas

  2. Внести данные в файл:

    KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=true;
};

  3. Создать тикет для пользователя user:

    $ kinit -p user@ADS-KAFKA.LOCAL

  4. Проверить тикет:

    $ 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. Экспортировать созданный файл client.jaas как параметр JVM для данного пользователя при помощи переменной среды KAFKA_OPTS:

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

Дополнительные сведения о работе с пользователями в среде аутентификации Kerberos SASL можно найти в статье Использование MIT Kerberos в Kafka.

Конфигурация списков ACL

Работа с авторизацией на основе ACL происходит путем запуска скрипта kafka-acls.sh в командной строке с любого узла кластера, на котором установлен сервис Kafka.

ПРИМЕЧАНИЕ

Для просмотра всех опций для скрипта kafka-acls.sh необходимо выполнить команду:

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

Назначение прав и запретов на отдельные действия в Kafka для пользователей

Назначение прав или запретов на отдельные действия пользователю осуществляется при помощи скрипта kafka-acls.sh с указанием опции --operation и операции, на которую дается доступ.

Назначение прав или запретов осуществляется при помощи опции --add.

Опция --allow-principal — разрешает действие, а --deny-principal — запрещает действие.

Удаление пользователя из списка ACL выполняется при помощи опции --remove.

Основные операции, для которых можно установить права пользователя:

  • Describe — просмотр метаданных топика или группы.

  • DescribeConfigs — просмотр конфигурации топика.

  • Alter —  изменение топика или группы.

  • IdempotentWrite — запись в топик для идемпотентных и транзакционных API.

  • Read — право на чтение из топика или из группы.

  • Delete — удаление топика или группы.

  • Create — создание топика или группы.

  • All — все права.

  • Write — право на запись в топик или группы.

  • AlterConfigs — изменение конфигурации топика.

ПРИМЕЧАНИЕ

Для пользователей, прошедших аутентификацию по протоколу Kerberos SASL или LDAP, в составлении списков ACL используется имя пользователя без обозначения области аутентификации (realm).

Для того, чтобы разрешить пользователю username выполнять операцию WRITE над топиком test-topic, необходимо на хосте с брокером Kafka ввести команду:

$ /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

Для того, чтобы запретить пользователю username1 выполнять операцию WRITE над топиком test-topic, необходимо на хосте с брокером Kafka ввести команду:

$ /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

В результате каждого добавления в список появляется сообщение об успешном добавлении прав или запрете, а также список ACL c текущими правами пользователей для топика:

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)

Назначение прав producer и consumer пользователю

Существует возможность назначить пользователю сразу несколько прав производителя (producer) — создание (create), запись (write), просмотр метаданных (describe).

Чтобы добавить права producer для топика test-topic пользователю writer, необходимо на хосте с брокером Kafka ввести команду:

$ /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

В результате появляется сообщение об успешном добавлении прав producer и список ACL c текущими правами пользователей для топика:

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)

Также существует возможность назначить сразу несколько прав потребителя (consumer) — чтение (read), просмотр метаданных (describe).

Чтобы добавить права consumer для топика test-topic пользователю reader, необходимо на хосте с брокером Kafka ввести команду:

$ /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
ПРИМЕЧАНИЕ
При назначении прав потребителя (consumer) необходимо указать и название топика (topic), и название группы потребителей (group), даже если потребитель всего один.

В результате появляется сообщение об успешном добавлении прав consumer и список ACL c текущими правами пользователей для топика test-topic, включая созданные ранее для пользователя writer:

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)

Также выводятся данные о созданных правах для группы потребителей:

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)

Просмотр списков ACL

Просмотр списков ACL осуществляется при помощи опции --list. Для того, чтобы просмотреть список прав для топика test-topic, необходимо выполнить команду:

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

В результате выводится список всех назначенных прав для топика test-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)

Для просмотра список прав для всех топиков используется такая же команда без указания имени топика.

Работа авторизованного пользователя в Kafka

После настройки списков ACL для пользователей работа в Kafka происходит в соответствии с шагами, описанными в статьях:

Любое действие неавторизованного пользователя вызовет ошибку. Например, попытка записать сообщения в топик test-topic для пользователя, не имеющего разрешения WRITE в списке ACL для данного топика, вызовет ошибки следующего содержания:

[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]
Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней