Kafka: Часть 9 — Описание ACL прав Kafka

В предыдущей статье мы узнали, как настраивать авторизацию ACL с аутентификацией. В этой статье мы расскажем о основных операциях (правах) в списках ACL Kafka.

Предыдущая статья. Следующая статья.

Полный список статей по теме тут.

Заходите в наш телеграмм канал — Enterprise Stack Helper! Делитесь опытом или задавайте вопросы, если что-то непонятно.

Репозитории с примерам из статей по способам аутентификаци и авторизации — https://github.com/BlockWit/kafka-security-examples.

Операции в списках ACL Kafka

В предыдущей статье для проверки работы авторизации AC, мы давали все (ALL) права клиенту, чтобы не разбираться. Но, как Вы понимаете, это не безопасно. Зачем вводить систему авторизации, если мы все равно даем все права? Это было сделано для простоты рассмотрения примера. Сейчас настало время разобраться с операциями и настроить списки ACL правильно.

Мы рассмотрим пока только некоторые операции необходимые для управления топиками и группами. Названия у них говорят сами за себя, поэтому пояснять детально не будем.

All — все права
Read — право на чтение. Например, из топика или их группы
Write — право на запись. Например, в топик или группы
Create — создание. Например, право на запись в топик или группу
Delete — удаление
Alter — изменение

Список актуальных операций всегда можно посмотреть в репозитории в исходном коде Kafka (https://github.com/apache/kafka/blob/24f664aa1621dc70794fd6576ac99547e41d2113/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java)

Работу некоторых из этих операций мы рассмотрим на примере топиков и групп. Вспомним команды, которые мы рассмотрели в предыдущей статье для управления списками ACL:

Разрешить пользователю [USERNAME] выполнять операцию [OPERATION] над топиком [TOPICNAME]

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:[USERNAME] --operation [OPERATION] --topic [TOPICNAME]

1	sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:[USERNAME] --operation [OPERATION] --topic [TOPICNAME]

Запретить пользователю [USERNAME] выполнять операцию [OPERATION] над топиком [TOPICNAME]

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:[USERNAME] --operation [OPERATION] --topic [TOPICNAME]

1	sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:[USERNAME] --operation [OPERATION] --topic [TOPICNAME]

Просмотреть списки прав

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list

1

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list

Посмотрите внимательно на команду удаления и добавления прав. В них есть параметр
—topic [TOPICNAME]

Если Вы его замените на
—group [GROUPNAME]
то получите команду удаления или добавления прав выполнения операции над группой!

Отлично, теперь давайте рассмотрим настройку на конкретных примерах.

Примеры использования

Для выполнения примеров будем использовать настройки Kafka и клиент из предыдущей статьи. Приводить код тут не будем, поскольку он не изменится. За исключением, конечно, клиента. В нем нужно будет менять username и password для конкретного примера.

Прежде чем выполнять примеры, убедитесь что у пользователей нет никаких прав. И, если они есть, то удалите их.

Пример с топиком

Давайте дадим Алисе право на запись в топик topic.alice, а также право на создание топика topic.alice. Право на создание дается, потому что такого топика еще нет и при первой записи в него Алисе потребуется его создать. Обратите внимание, что мы даем право на две операции одной командой:

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --operation WRITE --operation CREATE --topic topic.alice

1	sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --operation WRITE --operation CREATE --topic topic.alice

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

...
Message sends 1
Message sends 2
...

...

Message sends 1

Message sends 2

...

Топик создался и сообщения благополучно в него пишутся.

А давайте попробуем писать в этот же топик, но от имени Робина. Поменяйте в клиенте учетные данные на логин и пароль Робина: robin, robin-secret и запустите клиент. В результате получим:

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [topic.alice]

1	org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [topic.alice]

Как видите, наши настройки прав успешно работают в случае записи в топик.

Пример с группой

С группами не все так однозначно, как с топиками. Для того, чтобы клиент мог читать из группы, необходимо дать клиенту право на чтение из топика, из которого будем читать и право на чтение из группы, из которой будем читать. Если такой группы нет, то она создастся автоматически и право на создание группы не потребуется.

Давайте дадим Робину право на чтение из топика Алисы topic.alice и право на чтение из группы group.robin.from.alice.

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:robin --operation READ --group group.robin.from.alice
sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:robin --operation READ --topic topic.alice

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:robin --operation READ --group group.robin.from.alice

Для чтения сообщение нам потребуется соответствующий клиент. В репозитории он называется SimpleConsumer. Давайте приведем код клиента:

package com.blockwit.kafka.security.examples;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class SimpleConsumer {

    private static Consumer<Long, String> createConsumer(Map<String, String> inProps, String server, String topicName, String groupName) {

        final Properties props = new Properties();

        props.putAll(inProps);

        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");
        props.put("max.poll.records", "1");

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                server);
        props.put(ConsumerConfig.GROUP_ID_CONFIG,
                groupName);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());


        final Consumer<Long, String> consumer =
                new KafkaConsumer<>(props);

        TopicPartition tp = new TopicPartition(topicName, 0);
        List<TopicPartition> tps = Arrays.asList(tp);
        consumer.assign(tps);
        consumer.seekToBeginning(tps);

        return consumer;
    }

    static void runConsumer(Map<String, String> inProps, String server, String topicName, String groupName) {
        final Consumer<Long, String> consumer = createConsumer(inProps, server, topicName, groupName);

        final int giveUp = 100;
        int noRecordsCount = 0;

        while (true) {
            ConsumerRecords<Long, String> consumerRecords = null;

            try {
                consumerRecords = consumer.poll(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }

            if (consumerRecords.count() == 0) {
                noRecordsCount++;
                if (noRecordsCount > giveUp) break;
                else continue;
            }

            consumerRecords.forEach(record -> {
                System.out.printf("Consumer Record:(%d, %s, %d, %d)\n",
                        record.key(), record.value(),
                        record.partition(), record.offset());
            });

            consumer.commitAsync();
            break;
        }
        consumer.close();
    }

}

package com.blockwit.kafka.security.examples;

import org.apache.kafka.clients.consumer.Consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.clients.consumer.ConsumerRecords;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.common.TopicPartition;

import org.apache.kafka.common.serialization.LongDeserializer;

import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;

import java.util.List;

import java.util.Map;

import java.util.Properties;

public class SimpleConsumer {

private static Consumer<Long, String> createConsumer(Map<String, String> inProps, String server, String topicName, String groupName) {

final Properties props = new Properties();

props.putAll(inProps);

props.put("enable.auto.commit", "false");

props.put("auto.offset.reset", "earliest");

props.put("max.poll.records", "1");

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,

server);

props.put(ConsumerConfig.GROUP_ID_CONFIG,

groupName);

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,

LongDeserializer.class.getName());

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,

StringDeserializer.class.getName());

final Consumer<Long, String> consumer =

new KafkaConsumer<>(props);

TopicPartition tp = new TopicPartition(topicName, 0);

List<TopicPartition> tps = Arrays.asList(tp);

consumer.assign(tps);

consumer.seekToBeginning(tps);

return consumer;

}

static void runConsumer(Map<String, String> inProps, String server, String topicName, String groupName) {

final Consumer<Long, String> consumer = createConsumer(inProps, server, topicName, groupName);

final int giveUp = 100;

int noRecordsCount = 0;

while (true) {

ConsumerRecords<Long, String> consumerRecords = null;

try {

consumerRecords = consumer.poll(1000);

} catch (Exception e) {

e.printStackTrace();

}

if (consumerRecords.count() == 0) {

noRecordsCount++;

if (noRecordsCount > giveUp) break;

else continue;

}

consumerRecords.forEach(record -> {

System.out.printf("Consumer Record:(%d, %s, %d, %d)\n",

record.key(), record.value(),

record.partition(), record.offset());

});

consumer.commitAsync();

break;

}

consumer.close();

}

И код, который будет его запускать:

package com.blockwit.kafka.security.examples;

public class SimpleConsumerTest_SASL_SSL_PLAIN_ACL {

    public static void main(String[] args) throws InterruptedException {
        SimpleConsumer.runConsumer(Helper.of(
                "sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required serviceName=\"Kafka\" username=\"robin\" password=\"robin-secret\";",
                "sasl.mechanism", "PLAIN",
                "security.protocol", "SASL_SSL",
                "ssl.truststore.location", "/path/to/client/truststore/client.truststore.jks",
                "ssl.truststore.password", "clientTruststorePassword"),
                "localhost:9093", "topic.alice", "group.robin.from.alice");
    }

}

package com.blockwit.kafka.security.examples;

public class SimpleConsumerTest_SASL_SSL_PLAIN_ACL {

public static void main(String[] args) throws InterruptedException {

SimpleConsumer.runConsumer(Helper.of(

"sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required serviceName=\"Kafka\" username=\"robin\" password=\"robin-secret\";",

"sasl.mechanism", "PLAIN",

"security.protocol", "SASL_SSL",

"ssl.truststore.location", "/path/to/client/truststore/client.truststore.jks",

"ssl.truststore.password", "clientTruststorePassword"),

"localhost:9093", "topic.alice", "group.robin.from.alice");

}

И давайте запустим клиент. В итоге, мы получим информацию о том, что прочитано одно сообщение:

...
Consumer Record:(null, msg0, 0, 0)
...

...

Consumer Record:(null, msg0, 0, 0)

...

Что собой представляет операция чтения из группы? Это операция считывания сообщения и операция сохранения в группе номера следующего за считываемым сообщением, чтобы при следующем чтении из группы Kafka отдавала уже следующее сообщение. Номер следующего сообщения называется смещением, а операция установки смещения — коммит.

А теперь давайте попробуем считать Алисой сообщения из своего топика — alice.topic из группы group.robin.from.alice. Но для этого она должна иметь право на чтение из этого топика. Ранее мы давали только право на запись и создание. Давайте дадим право на чтение:

sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --operation READ --topic topic.alice

1	sudo /opt/kafka/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --operation READ --topic topic.alice

У Алисы нет прав на чтение из группы group.robin.from.alice . Как же поведет себя клиент? Измените учетные данные клиента на логин и пароль Алисы и запустите. В результате мы получим несколько странный вывод. Сообщение прочитается:

...
Consumer Record:(null, msg0, 0, 0)
...

...

Consumer Record:(null, msg0, 0, 0)

...

Но коммит упадет с ошибкой:

16:00:36,653 ERROR [KafkaConsumer] [Consumer clientId=consumer-1, groupId=group.robin.from.alice] Failed to close coordinator
org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: group.robin.from.alice

1 2	16:00:36,653 ERROR [KafkaConsumer] [Consumer clientId=consumer-1, groupId=group.robin.from.alice] Failed to close coordinator org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: group.robin.from.alice

Получается что запрет на чтение из группы — это вовсе не запрет просмотра сообщения из топика. Сообщения из топика мы можем просматривать, если у нас есть права на чтение из топика. А вот изменять параметр оффсета в группе мы не можем. Неоднозначно, но вполне логично.

Резюме

В этой статье мы ознакомились с основными правами ACL в Kafka и рассмотрели пару примеров работы с ними. В следующей статье мы добавим защиту к межброкерному взаимодействию.

Предыдущая статья. Следующая статья.

Полный список статей по теме тут.

Заходите в наш телеграмм канал — Enterprise Stack Helper! Делитесь опытом или задавайте вопросы, если что-то непонятно.

Репозитории с примерам из статей по способам аутентификаци и авторизации — https://github.com/BlockWit/kafka-security-examples.

inaword

Блог о разработке и современных технологиях

Kafka: Часть 9 — Описание ACL прав Kafka

Операции в списках ACL Kafka

Примеры использования

Пример с топиком

Пример с группой

Резюме

Добавить комментарий Отменить ответ

Kafka: Часть 9 — Описание ACL прав Kafka

Операции в списках ACL Kafka

Примеры использования

Пример с топиком

Пример с группой

Резюме

Читайте также

Добавить комментарий Отменить ответ