Kafka: Часть 8 — Аутентификация и авторизация с ACL

В предыдущей статье мы настроили и проверили аутентификацию по протоколу SASL с механизмом PLAIN через SSL. Настало время познакомиться с авторизацией в Kafka!

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

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

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

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

ACL в Kafka

Давайте вспомним основную информацию из первой статьи:

Авторизация — это процесс проверки — имеет ли право пользователь выполнять те или иные действия над ресурсом
Авторизация настраивается независимо от аутентификации
Авторизация, если она настроена, выполняется после аутентификации
Kafka имеет встроенный авторизатор на базе ACL

А теперь подробнее о некоторых пунктах.

Авторизация настраивается независимо от аутентификации. Т.е. независимо от любой схемы аутентификации мы можем добавить конфигурацию авторизации на сервер и она будет работать.

Kafka имеет встроенный авторизатор на базе ACL. ACL — это просто списки, в которых описано кому над каким ресурсом можно совершать действия.

Авторизатор включается так:

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

1	authorizer.class.name=kafka.security.authorizer.AclAuthorizer

Еще нужно добавить опцию:

allow.everyone.if.no.acl.found=true

1	allow.everyone.if.no.acl.found=true

Что она означает? Она делает вот что:

Если топик, группа или другой объект не имеют ни одной записи в списке ACL, то доступ к этому топику, группе или другому объекту разрешен для всех!

Если топик, группа или другой объект имеет хотя бы одну ACL запись, то доступ к этому топику, группе или объекту разрешен тем, у кого есть соответствующие права, и суперпользователям.

Если allow.everyone.if.no.acl.found=false, то:

Если топик, группа или другой объект не имеют ни одной записи в списке ACL, то доступ к этому топику, группе или другому объекту разрешен только суперпользователям.

Если топик, группа или другой объект имеет хотя бы одну ACL запись, то доступ к этому топику, группе или объекту разрешен тем, у кого есть соответствующие права, и суперпользователям.

Зачем же мы установили allow.everyone.if.no.acl.found в true? Логичным было бы запретить доступ ко всем топикам, если у пользователя нет прав. Это сделано сейчас для удобства, чтобы мы могли быстро настроить ACL.

Дело в том, что у нас есть межброкерное взаимодействие на 9092 порту. Для сервсных операций Kafka. (Подробнее в в этой статьей). И на 9092 порту нет аутентификации. Это значит, что сервер не сможет определить «личность» клиента. А значит и не сможет провести авторизацию по ACL! Ведь ACL авторизатор работает для всех соединений сразу. В итоге клиент на 9092 порту не получит вообще никаких прав. А это значит, что Kafka не сможет выполнять сервисные операции. И работать наш сервер корректно не будет. А в логах сможем увидеть такую ошибку:

org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(processor=0, connectionId=127.0.0.1:9092-127.0.0.1:57784-0, session=Session(User:ANONYMOUS,/127.0.0.1),

1

org.apache.kafka.common.errors.ClusterAuthorizationException: Request Request(processor=0, connectionId=127.0.0.1:9092-127.0.0.1:57784-0, session=Session(User:ANONYMOUS,/127.0.0.1),

Сервисные операции касаются конфигурации самой Kafka (конфигурации кластера, если быть точным). Поскольку мы им не указывали списки доступа ACL, то мы можем поставить опцию — allow.everyone.if.no.acl.found в true. Таким образом, неаутентифицированный сервисный клиент получит возможность выполнять ВСЕ операции над конфигурацией Kafka. Таким образом, межброкерное взаимодействие будет работать корректно.

Безусловно, такой подход — дыра в безопасности. И мы ее исправим в ближайших статьях, когда будем настраивать аутентификацию для межброкерного взаимодействия. А пока поработаем с этой опцией.

С конфигурационным файлом сервера все. Однако нам нужно теперь сконфигурировать сами списки ACL.

В этом нам помогут команды:

Просмотр списков ACL для топика [TOPICNAME]

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

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

Разрешить пользователю [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]

Обратите внимание, что в командах указывается соединение с ZooKeeper. Дело в том, что Kafka хранит списки ACL в ZooKeeper.

Теперь давайте приступим к практике.

Авторизация ACL с SASL_SSL и PLAIN

Для начала запустим Kafka и добавим в списки ACL право Алисе (у нас она alice) выполнять любые действия (обозначается как ALL) над топиком stats.store.

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

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 ALL --topic stats.store

После выполнения команды Kafka даст нам знать, что права изменены:

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=stats.store, patternType=LITERAL)`:
 	(principal=User:alice, host=*, operation=ALL, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=stats.store, patternType=LITERAL)`:
 	(principal=User:alice, host=*, operation=ALL, permissionType=ALLOW)

Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=stats.store, patternType=LITERAL)`:

(principal=User:alice, host=*, operation=ALL, permissionType=ALLOW)

Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=stats.store, patternType=LITERAL)`:

(principal=User:alice, host=*, operation=ALL, permissionType=ALLOW)

Отлично. Теперь скопируем конфигурацию SASL_SSL+PLAIN для сервера из предыдущей статьи. И добавим к ней конфигурацию авторизатора. Приведем полную конфигурацию:

broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
advertised.host.name=localhost

listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093
advertised.listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093

security.protocol=SASL_SSL
sasl.enabled.mechanisms=PLAIN

listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="admin-secret" \
  user_admin="admin-secret" \
  user_robin="robin-secret" \
  user_alice="alice-secret";

ssl.keystore.location=/path/to/server/keystore/server.keystore.jks
ssl.keystore.password=serverKeystorePassword

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

broker.id=0

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/tmp/kafka-logs

num.partitions=1

num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

zookeeper.connect=localhost:2181

zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

delete.topic.enable=true

advertised.host.name=localhost

listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093

advertised.listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093

security.protocol=SASL_SSL

sasl.enabled.mechanisms=PLAIN

listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \

username="admin" \

password="admin-secret" \

user_admin="admin-secret" \

user_robin="robin-secret" \

user_alice="alice-secret";

ssl.keystore.location=/path/to/server/keystore/server.keystore.jks

ssl.keystore.password=serverKeystorePassword

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

allow.everyone.if.no.acl.found=true

Не забывайте менять localhost, ssl.keystore.location и ssl.keystore.password на свои!

Настройки клиента остаются такие же, как и в предыдущей статье.

package com.blockwit.kafka.security.examples;

public class SimpleProducerTest_SASL_SSL_PLAIN_ACL {

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

}

package com.blockwit.kafka.security.examples;

public class SimpleProducerTest_SASL_SSL_PLAIN_ACL {

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

SimpleProducer.runProducer(Helper.of(

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

"sasl.mechanism", "PLAIN",

"security.protocol", "SASL_SSL",

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

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

"localhost:9093", "stats.store");

}

Не забывайте менять localhost, ssl.truststore.location и ssl.truststore.password на свои!

Запустим Kafka, потом клиент и получим вывод успешной отправки сообщений:

...
Message sends 1
Message sends 2
Message sends 3
Message sends 4
Message sends 5
Message sends 6
...

...

Message sends 1

Message sends 2

Message sends 3

Message sends 4

Message sends 5

Message sends 6

...

Резюме

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

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

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

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

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

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

inaword

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

Kafka: Часть 8 — Аутентификация и авторизация с ACL

ACL в Kafka

Авторизация ACL с SASL_SSL и PLAIN

Резюме

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

Kafka: Часть 8 — Аутентификация и авторизация с ACL

ACL в Kafka

Авторизация ACL с SASL_SSL и PLAIN

Резюме

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

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