Повышаем безопасность кластера Apache Kafka с помощью SSL, SASL и ACL
Необходимые условия: Начальные навыки работы с Apache Kafka, продюсерами и консюмерами Kafka.
Эта статья в первую очередь посвящена SASL, SSL и ACL.
В этом посте мы рассмотрим конфигурации для включения аутентификации с помощью SCRAM, авторизации с помощью SimpleAclAuthorizer и шифрования между клиентами и сервером с помощью SSL.
Помимо всего прочего Apache Kafka используется и для обмена конфиденциальной информацией как между различными системами внутри компании, так и между различными компаниями. Поэтому между клиентскими приложениями и сервером Kafka должны быть установлены безопасные соединения типа HTTPS. Сообщество разработчиков Apache Kafka с открытым исходным кодом предоставило множество вариантов повышения безопасности Kafka для аутентификации, авторизации и шифрования.
Аутентификация Kafka:
- SSL
- SASL: PLAIN, SCRAM(SHA-256 and SHA-512), OAUTHBEARER, GSSAPI(Kerberos)
Авторизация в Kafka: Kafka поставляется с простым классом авторизации kafka.security.auth.SimpleAclAuthorizer для работы с ACL (создание, чтение, запись, описание, удаление). Мы также можем добавить свою собственную реализацию класса Authorizer.
Шифрование: Шифрование данных по сети между клиентами и сервером Kafka, т.е. между продюсерами и сервером Kafka, консюмерами и сервером Kafka.
Аутентификация с помощью SCRAM:
Механизм аутентификации Salted Challenge Response Authentication Mechanism (SCRAM) аналогичен аутентификации с использованием имени пользователя/пароля. Apache Kafka поддерживает SCRAM-SHA-256 и SCRAM-SHA-512. Для этого механизма Kafka по умолчанию (ScramLoginModule) хранит учетные данные SCRAM в zookeeper, поэтому zookeeper должен быть надежно защищен. Если иные параметры не указаны, по умолчанию используется число итераций 4096. Идентификатор SCRAM, состоящий из итераций, StoredKey и ServerKey, хранится в ZooKeeper. В нашем случае для аутентификации мы будем использовать SCRAM-SHA-512
Авторизация с помощью SimpleAclAuthorizer:
Apache Kafka поставляется с простой формой авторизации kafka.security.auth.SimpleAclAuthorizer для всех операций ACL. SimpleAcl Authroizer предоставляет разрешения create, read, write, cluster_action, alter_configs, describe и delete. В примерах ниже мы увидим роль каждого из этих разрешений, а также ошибки приложения, возникающие в случае, если эти роли были назначены неправильно.
Шифрование с помощью SSL:
SSL (Secure Sockets Layer) можно настроить на шифрование, а также на двустороннюю аутентификацию между клиентом и сервером. То есть брокер аутентифицирует клиента, используя сертификат хранилища доверия клиента, а клиент аутентифицирует брокера, используя сертификат хранилища доверия брокера. В этой статье мы настроим SSL только на шифрование. SSL использует пары приватных ключей/сертификатов, которые используются в процессе SSL handshake.
- Каждому брокеру Kafka нужна своя пара частный ключ/сертификат, для аутентификации брокера клиент использует предварительно настроенный сертификат
Генерация SSL-сертификата и ключа: Создайте SSL keystore и сертификат truststore брокера Kafka с помощью скрипта confluent-platform-security-tools . Сохраните пароли keystore и truststore, который Вы вводите во время выполнения этого скрипта, они понадобятся нам для настройки Kafka server.properties.
С помощью специального скрипта скопируем truststore и keystore в папку config, находящуюся в каталоге загруженного бинарного файла Kafka.
Выполним все нижеприведенные команды, перейдя в терминале по адресу <kafka-binary-dir>/.
Запуск Zookeeper:
./bin/zookeeper-server-start.sh config/zookeeper.properties
Создание суперпользователя Kafka:
./bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='admin-secret']' --entity-type users --entity-name adminCompleted Updating config for entity: user-principal 'admin'.
Создание файла kafka_server_jaas.conf в папке config:
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret"; };
Создание файла ssl-user-config.properties в папке config:
security.protocol=SASL_SSL sasl.mechanism=SCRAM-SHA-512 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="demo-user" password="secret"; ssl.truststore.location= <kafka-binary-dir>/config/truststore/kafka.truststore.jks ssl.truststore.password=password
Добавьте приведенные ниже параметры безопасности к существующему файлу Kafka server.properties в папке config.
########### SECURITY using SCRAM-SHA-512 and SSL ###################listeners=PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093,SASL_SSL://localhost:9094advertised.listeners=PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093,SASL_SSL://localhost:9094security.inter.broker.protocol=SASL_SSLssl.endpoint.identification.algorithm=ssl.client.auth=requiredsasl.mechanism.inter.broker.protocol=SCRAM-SHA-512sasl.enabled.mechanisms=SCRAM-SHA-512# Broker security settings ssl.truststore.location= <kafka-binary-dir>/config/truststore/kafka.truststore.jksssl.truststore.password=passwordssl.keystore.location= <kafka-binary-dir>/config/keystore/kafka.keystore.jksssl.keystore.password=passwordssl.key.password=password# ACLs authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer super.users=User:admin#zookeeper SASL zookeeper.set.acl=false ########### SECURITY using SCRAM-SHA-512 and SSL ###################
Теперь с помощью файла jaas conf запустите Kafka:
export KAFKA_OPTS=-Djava.security.auth.login.config=<kafka-binary-dir>/config/kafka_server_jaas.conf./bin/kafka-server-start.sh ./config/server.properties
Теперь клиенты могут подключаться к кластеру Kafka тремя способами (для вышеуказанной конфигурации server.properties):
- localhost:9092: PLAIN_TEXT без аутентификации и авторизации.
- localhost:9093: PLAIN_TEXT с аутентификацией и авторизацией.
- localhost:9094: вместе с шифрованием, аутентификацией, авторизацией.
Теперь давайте рассмотрим различные рабочие процессы с использованием кластера Kafka.
Создание пользователя:
./bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password='secret']' --entity-type users --entity-name demouserCompleted Updating config for entity: user-principal 'demouser'.
Создание файла ssl-user-config.properties в папке config:
security.protocol=SASL_SSLsasl.mechanism=SCRAM-SHA-512sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="demouser" password="secret";ssl.truststore.location= <kafka-bin-dir>/config/truststore/kafka.truststore.jksssl.truststore.password=password
Создание топика:
./bin/kafka-topics.sh --create --bootstrap-server localhost:9094 --command-config ./config/ssl-user-config.properties --replication-factor 1 --partitions 1 --topic demo-topicError while executing topic command : org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] [2019-06-20 11:27:29,244] ERROR java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45) at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32) at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89) at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260) at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:175) at kafka.admin.TopicCommand$TopicService.createTopic(TopicCommand.scala:134) at kafka.admin.TopicCommand$TopicService.createTopic$(TopicCommand.scala:129) at kafka.admin.TopicCommand$AdminClientTopicService.createTopic(TopicCommand.scala:157) at kafka.admin.TopicCommand$.main(TopicCommand.scala:60) at kafka.admin.TopicCommand.main(TopicCommand.scala) Caused by: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [Topic authorization failed.] (kafka.admin.TopicCommand$)
Назначьте demouser разрешения «Создать» и «Описать», а затем создайте топик:
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:demouser --operation Create --operation Describe --topic demo-topicAdding ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Create from hosts: * User:demouser has Allow permission for operations: Describe from hosts: *Current ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Create from hosts: * User:demouser has Allow permission for operations: Describe from hosts: *./bin/kafka-topics.sh --create --bootstrap-server localhost:9094 --command-config ./config/ssl-user-config.properties --replication-factor 1 --partitions 1 --topic demo-topicTopic demo-topic created
Создание файла ssl-producer.properties в папке config.
bootstrap.servers=localhost:9094compression.type=none### SECURITY ######security.protocol=SASL_SSLsasl.mechanism=SCRAM-SHA-512sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="demouser" password="secret";ssl.truststore.location= <kafka-binary-dir>/config/truststore/kafka.truststore.jksssl.truststore.password=password
Использование продюсера с неправильным паролем: измените пароль на другой в ssl-producer.properties, а затем попробуйте произвести данные на в топике demo-topic:
./bin/kafka-console-producer.sh --broker-list localhost:9094 --topic demo-topic --producer.config config/ssl-producer.properties[2019-06-22 18:41:27,406] ERROR [Producer clientId=console-producer] Connection to node -1 (localhost/127.0.0.1:9094) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512 (org.apache.kafka.clients.NetworkClient)
Использование продюсера с правильным паролем и без прав на производство в топике demo-topic:
./bin/kafka-console-producer.sh --broker-list localhost:9094 --topic demo-topic --producer.config config/ssl-producer.properties>ERROR Error when sending message to topic demo-topic with key: null, value: 8 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [demo-topic]
Назначьте разрешения Producer для demouser в топике demo-topic:
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:demouser --producer --topic demo-topicAdding ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Create from hosts: * User:demouser has Allow permission for operations: Describe from hosts: * User:demouser has Allow permission for operations: Write from hosts: *Current ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Create from hosts: * User:demouser has Allow permission for operations: Describe from hosts: * User:demouser has Allow permission for operations: Write from hosts: *
Создавайте сообщения с правильным паролем и правами доступа:
./bin/kafka-console-producer.sh --broker-list localhost:9094 --topic demo-topic --producer.config config/ssl-producer.properties >message1 >message2 >message3 ^C
Теперь мы создали 3 сообщения на demo-topic, давайте попробуем потребить эти сообщения из топика, создав консюмера.
Создание файла ssl-consumer.properties в папке config:
bootstrap.servers=localhost:9094# consumer group idgroup.id=demo-consumer-group### SECURITY ###### security.protocol=SASL_SSLsasl.mechanism=SCRAM-SHA-512sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="demouser" password="secret";ssl.truststore.location= <kafka-binary-dir>/config/truststore/kafka.truststore.jksssl.truststore.password=password
Потребление данных из топика demo-topic с неверным паролем в ssl-consumer.properties
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic demo-topic --from-beginning --consumer.config config/ssl-consumer.properties[2019-06-22 18:46:41,983] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$) org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512 Processed a total of 0 messages
Потребление данных из топика demo-topic с правильным паролем в ssl-consumer.properties:
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic demo-topic --from-beginning --consumer.config config/ssl-consumer.properties[2019-06-22 18:47:37,461] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$) org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: demo-consumer-group Processed a total of 0 messages
Назначить для demouser потребительские права на топик demo-topic:
./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:demouser --consumer --topic demo-topic --group demo-consumer-groupAdding ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Describe from hosts: * User:demouser has Allow permission for operations: Read from hosts: *Adding ACLs for resource `Group:LITERAL:demo-consumer-group`: User:demouser has Allow permission for operations: Read from hosts: *Current ACLs for resource `Topic:LITERAL:demo-topic`: User:demouser has Allow permission for operations: Create from hosts: * User:demouser has Allow permission for operations: Describe from hosts: * User:demouser has Allow permission for operations: Write from hosts: * User:demouser has Allow permission for operations: Read from hosts: *Current ACLs for resource `Group:LITERAL:demo-consumer-group`: User:demouser has Allow permission for operations: Read from hosts: *
Потребляйте сообщения из топика demo-topic с правильным паролем и правами на потребление:
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic demo-topic --from-beg inning --consumer.config config/ssl-consumer.properties message1 message2 message3
