使用SASL / SCRAM進行身份驗證

請先在不配置任何身份驗證的情況下啟動Kafka

1. 創建SCRAM Credentials

1.1 創建broker通信用戶(或稱超級用戶) bash Emacs bin/kafka-configs.sh --zookeeper centos1:2181 --alter --add-config 'SCRAM-SHA-256=[password=adminpwd],SCRAM-SHA-512=[password=adminpwd]' --entity-type users --entity-name admin

1.2 創建客戶端用戶dbcUser bash Emacs bin/kafka-configs.sh --zookeeper centos1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=changeit],SCRAM-SHA-512=[password=changeit]' --entity-type users --entity-name dbcUser

1.3 查看SCRAM證書 bash Emacs bin/kafka-configs.sh --zookeeper centos1:2181 --describe --entity-type users --entity-name dbcUser

刪除SCRAM證書 (只是說明功能，這里不執行) bash Emacs bin/kafka-configs.sh --zookeeper centos1:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name dbcUser

2. 配置Kafka Brokers

2.1 在每個Kafka broker的config目錄中添加一個kafka_server_jaas.conf，內容如下 bash Emacs KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="adminpwd"; };

注意：不要少寫了分號

2.2 將JAAS配置文件位置作為JVM參數傳遞給每個Kafka broker

修改 kafka/bin/kafka-server-start.sh

將exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" 注釋, 增加下面的內容 bash Emacs #exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf kafka.Kafka "$@"

或者不修改kafka-server-start.sh腳本, 而是將下面的內容添加到~/.bashrc bash Emacs export KAFKA_PLAIN_PARAMS="-Djava.security.auth.login.config=/usr/local/kafka/config/kafka_server_jaas.conf" export KAFKA_OPTS="$KAFKA_PLAIN_PARAMS $KAFKA_OPTS"

2.3 在server.properties中配置SASL端口和SASL機制。 bash Emacs # 認證配置 listeners=SASL_PLAINTEXT://centos1:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 sasl.enabled.mechanisms=SCRAM-SHA-256 # ACL配置 allow.everyone.if.no.acl.found=false super.users=User:admin authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

2.4 重啟ZK/Kafka

重啟ZK / Kafka服務. 所有broker在連接之前都會引用’kafka_server_jaas.conf’. bash Emacs #重啟所有zookeeper bin/zookeeper-server-stop.sh bin/zookeeper-server-start.sh -daemon config/zookeeper.properties #重啟說有broker bin/kafka-server-stop.sh bin/kafka-server-start.sh -daemon config/server.properties

3. 客戶端配置

先使用kafka-console-producer 和 kafka-console-consumer 測試一下

kafka-console-producer

1. 創建 config/client-sasl.properties 文件 bash Emacs security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256

2. 創建 config/kafka_client_jaas_admin.conf 文件 bash Emacs KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="adminpwd"; };

3. 修改kafka-console-producer.sh腳本

這里我們復制一份新文件來改 bash Emacs cp bin/kafka-console-producer.sh bin/kafka-console-producer-admin.sh vi bin/kafka-console-producer-admin.sh #exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@" exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka_client_jaas_admin.conf kafka.tools.ConsoleProducer "$@"

4. 創建測試topic(之前以及創建則不用創建) bash Emacs bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 1 --replication-factor 1 --topic test

5. 測試生產消息 bash Emacs bin/kafka-console-producer-admin.sh --broker-list centos1:9092 --topic test --producer.config config/client-sasl.properties [wanghy@centos1 kafka]$ bin/kafka-console-producer-admin.sh --broker-list centos1:9092 --topic test --producer.config config/client-sasl.properties >hello, I am admin >

可以看到admin用戶無需配置ACL就可以發送消息

6. 測試 dbcUser 用戶

創建一個bin/kafka-console-producer-dbc.sh文件 bash Emacs cp bin/kafka-console-producer-admin.sh bin/kafka-console-producer-dbc.sh exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka_client_jaas_dbc.conf kafka.tools.ConsoleProducer "$@"

創建kafka_client_jaas_dbc.conf文件 bash Emacs cp config/kafka_client_jaas_admin.conf config/kafka_client_jaas_dbc.conf KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="dbcUser" password="changeit"; };

生產消息 bash Emacs [wanghy@centos1 kafka]$ bin/kafka-console-producer-dbc.sh --broker-list centos1:9092 --topic stest --producer.config config/client-sasl.properties >hello, I am dbcUser [2019-03-15 09:47:28,483] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {stest=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient) [2019-03-15 09:47:28,486] ERROR Error when sending message to topic stest with key: null, value: 19 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [stest] >

報錯了，提示沒有訪問權限

kafka-console-consumer

1. 創建 config/consumer-dbc.properties 文件 bash Emacs security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 group.id=dbc-group

2. 創建 bin/kafka-console-consumer-dbc.sh 文件 bash Emacs cp bin/kafka-console-consumer.sh bin/kafka-console-consumer-dbc.sh vi bin/kafka-console-consumer-dbc.sh

修改內容如下 bash Emacs #exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@" exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka_client_jaas_dbc.conf kafka.tools.ConsoleConsumer "$@"

3. 測試消費者 bash Emacs bin/kafka-console-consumer-dbc.sh --bootstrap-server centos1:9092 --topic test --consumer.config config/consumer-dbc.properties --from-beginning [2019-03-15 10:03:27,794] WARN [Consumer clientId=consumer-1, groupId=dbc-group] Error while fetching metadata with correlation id 2 : {test=TOPIC_AUTHORIZATION_FAILED} (org.apache.kafka.clients.NetworkClient) [2019-03-15 10:03:27,796] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$) org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [test] Processed a total of 0 messages

沒有權限

總結

以上是生活随笔為你收集整理的kafka身份认证 maxwell_Kafka 使用SASL / SCRAM进行身份验证的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。