DRAFT
В предыдущей статье мы рассмотрели права ACL и попробовали ими управлять. В этой статье мы добавим настройки для межброкерного взаимодействия.
А если хотите узнать зачем вообще нужна межброкерная коммуникация в Kafka с одним сервером, то Вам сюда.
Полный список статей по теме тут.
Заходите в наш телеграмм канал — Enterprise Stack Helper! Делитесь опытом или задавайте вопросы, если что-то непонятно.
Репозитории с примерам из статей по способам аутентификаци и авторизации — https://github.com/BlockWit/kafka-security-examples.
Настройка межброкерного взаимодействия
Итак, из предыдущих статей мы знаем, что межброкерное взаимодействие у нас по умолчанию незащищенное. Т.е. как PLAINTEXT. И, если мы добавляем для клиентов аутентификацию по другому протоколу, то должны оставить явно PLAINTEXT помимо клиентского соединения. Это мы все время и делали. У нас всегда было описано два соединения:
|
1 2 |
listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093 advertised.listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093 |
- SASL_SSL://localhost:9093 — на 9093 порту мы описывали соединения для наших экспериментов с клиентом. И потом описывали выбор протокола для клиента с помощью опции security.protocol.
- PLAINTEXT://localhost:9092 — а это соединение мы оставили для межброкерного взаимодействия. Выбор именно этого соединения для межброкерного взаимодействия мы явно не задавали. Потому как PLAINTEXT выбирается по умолчанию.
Что же это за межброкерное взаимодействие? Это обмен информацией между серверами Kafka и некоторые сервисные операции.
Как же выбрать для межброкерного взаимодействия не PLAINTEXT, а другое? Например то же SASL_SSL? Для этого есть аналоги опций security.protocol и sasl.enabled.mechanisms:
- security.inter.broker.protocol — делает то же, что и security.protocol, только для межброкерного взаимодействия. Т.е. выбирает протокол.
- sasl.mechanism.inter.broker.protocol — делает то же, что и sasl.enabled.mechanisms, только для межброкерного взаимодействия. Т.е. выбирает конкретный механизм аутентификации SASL.
Т.е. в строках advertised.listeners и listeners мы описываем все типы протоколов. А дальше опциями выбираем какой из описанных протоколов будет работать для клиентов, а какой для межброкерного взаимодействия. При этом настройки авторизации будут работать для всех соединений! Давайте возьмем пример из 8-ой статьи. Там у нас была аутентификация клиента по протоколу SASL_SSL с механизмом PLAIN и авторизацией. И сделаем так, чтобы межброкерное взаимодействие также шло через 9093 порт по протоколу SASL_SSL. Для этого мы уберем PLAINTEXT описание из соединений. Тогда соединения у нас станут такими:
|
1 2 |
listeners=SASL_SSL://localhost:9093 advertised.listeners=SASL_SSL://localhost:9093 |
И для межброкерного взаимодействия выберем SASL_SSL
|
1 2 |
security.inter.broker.protocol=SASL_SSL sasl.mechanism.inter.broker.protocol=PLAIN |
Почти все! Осталась одна важная деталь!
Поскольку у нас соединение SSL, то нужно предоставить еще и хранилище доверенных сертификатов для сервера. Ведь должны же брокеры проверять сертификаты друг-друга как-то.
Для этого возьмем наш сертификат сервера, который мы делали в этой статье, и создадим с ним хранилище для сервера:
|
1 |
keytool -keystore server.truststore.jks -import -file domain.pem |
Где domain.pem — сертификат сервера (он либо Вам выдан, либо вы его создавали в этой статье). После выполнения операции появится файл server.truststore.jks. Также запомните пароль к хранилищу.
Добавим наше хранилище доверенных сертификатов к настройкам сервера:
|
1 2 |
ssl.truststore.location=/path/to/server/truststore/server.truststore.jks ssl.truststore.password=serverTruststorePassword |
Не забудьте сделать владельцем хранилища пользователя kafka!
Теперь последний штрих. Поскольку брокеры у нас теперь полноценно аутентифицируются, то можно убрать опцию ACL:
|
1 |
allow.everyone.if.no.acl.found=true |
И нужно добавить сделать пользователя, с помощью которого общаются брокеры, суперпользователем:
|
1 |
super.users=User:admin |
Давайте посмотрим на наш полный конфиг сервера Kafka:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 |
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=SASL_SSL://localhost:9093 advertised.listeners=SASL_SSL://localhost:9093 security.inter.broker.protocol=SASL_SSL sasl.mechanism.inter.broker.protocol=PLAIN security.protocol=SASL_SSL sasl.enabled.mechanisms=PLAIN super.users=User:admin 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.p12 ssl.keystore.password=serverKeystorePassword ssl.truststore.location=/path/to/server/truststore/server.truststore.jks ssl.truststore.password=serverTruststorePassword authorizer.class.name=kafka.security.authorizer.AclAuthorizer |
Отлично, теперь запустим Kafka! Клиент править нет необходимости. Его также берем из 8-ой статьи и запускаем. Вывод должен быть:
|
1 2 3 4 5 |
... Message sends 2 Message sends 3 Message sends 4 .... |
Резюме
В этой статье мы настроили межброкерное взаимодействие по защищенному протоколу.
Полный список статей по теме тут.
Заходите в наш телеграмм канал — Enterprise Stack Helper! Делитесь опытом или задавайте вопросы, если что-то непонятно.
Репозитории с примерам из статей по способам аутентификаци и авторизации — https://github.com/BlockWit/kafka-security-examples.
