0


在Kafka3.7.0中进行SASL_SCRAM认证配置_不使用自带zookeeper

在Kafka3.7.0中进行SASL_SCRAM认证配置_不使用自带zookeeper

1. 启动Zookeeper

​ 这里不使用kafka安装包自带的zookeeper,从官网下载apache-zookeeper-3.8.0-bin.tar.gz这个版本使用

​ 另外注意./conf/zoo.cfg这个配置文件内容,dataDir配置,这个存放zookeeper的快照文件,这个路径默认值固定,如果相关机器历史上有配置过zookeeper并且没有修改此配置,那本次配置的zookeeper启动时会加载历史的数据日志文件(不想修改,且确定历史数据不重要的情况下,可以删除该目录下的数据)。

  1. # the directory where the snapshot is stored.
  2. dataDir=/data/bigdata/zookeeper/data

​ 在没有设置任何权限的配置下启动Zookeeper:

  1. ./bin/zkServer.sh start ./conf/zoo.cfg

2 创建SCRAM证书

2.1 创建broker通信用户admin

​ 在使用sasl之前必须先创建,否则启动报错

  1. bin/kafka-configs.sh --zookeeper127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-succ],SCRAM-SHA-512=[password=admin-succ]' --entity-type users --entity-name admin

2.2 创建生产用户producer

  1. bin/kafka-configs.sh --zookeeper127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=producer-succ],SCRAM-SHA-512=[password=producer-succ]' --entity-type users --entity-name producer

2.3 创建消费用户:consumer

  1. bin/kafka-configs.sh --zookeeper127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=consumer-succ],SCRAM-SHA-512=[password=consumer-succ]' --entity-type users --entity-name consumer

​ SCRAM-SHA-256/SCRAM-SHA-512是对密码加密的算法,二者有其一即可。

3 维护SCRAM证书

3.1 查看SCRAM证书

  1. bin/kafka-configs.sh --zookeeper127.0.0.1:2181 --describe --entity-type users --entity-name producer
  2. bin/kafka-configs.sh --zookeeper127.0.0.1:2181 --describe --entity-type users --entity-name consumer

3.2 删除SCRAM证书

  1. bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-512' --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name producer

4 服务端配置

​ 在用户证书创建完毕之后开始Kafka服务端的配置

4.1 创建JAAS文件

​ 进入kafka/config文件夹,执行以下命令:

  1. cat> kafka_server_jaas.conf <<EOF
  2. KafkaServer {
  3. org.apache.kafka.common.security.scram.ScramLoginModule required
  4. username="admin"
  5. password="admin-succ";
  6. };
  7. EOF

4.2 启动加载jaas配置文件

​ 进入kafka/bin文件夹,在 kafka-server-start.sh 最上方添加(路径换成自己的)

  1. exportKAFKA_OPTS=" -Djava.security.auth.login.config=/home/bigdata/apps/kafka/config/kafka_server_jaas.conf"

4.3 配置server.properties【config/server.properties】

​ 将下面配置添加至其中

  1. #认证配置
  2. listeners=SASL_PLAINTEXT://:9092
  3. security.inter.broker.protocol=SASL_PLAINTEXT
  4. sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
  5. sasl.enabled.mechanisms=SCRAM-SHA-256
  6. #ACL配置
  7. allow.everyone.if.no.acl.found=false
  8. super.users=User:admin
  9. authorizer.class.name=kafka.security.authorizer.AclAuthorizer

​ 可以根据自己的需求选择SASL_SSL或SASL_PLAINTEXT, PLAINTEXT为不加密明文传输,性能好一点。

​ 另外注意,log.dirs配置,这个存放kafka的数据日志文件,这个路径默认值固定,如果相关机器历史上有配置过kafka并且没有修改此配置,那本次配置的kafka启动时会加载历史的数据日志文件(不想修改,且确定历史数据不重要的情况下,可以删除该目录下的数据)。

  1. # A comma separated list of directories under which to store log files
  2. log.dirs=/tmp/kafka-logs

4.4 配置完后重启Zookeeper,再启动Kafka

  1. ./bin/zkServer.sh stop
  2. ./bin/zkServer.sh start ./conf/zoo.cfg
  3. nohupsh ./bin/kafka-server-start.sh ./config/server.properties >/dev/null 2>&1&

5 客户端配置

5.1 创建的三个用户的三个JAAS文件

  1. kafka_client_scram_admin_jaas.conf
  2. kafka_client_scram_producer_jaas.conf
  3. kafka_client_scram_consumer_jaas.conf

​ 进入kafka/config文件夹,执行以下命令:

  1. cat> kafka_client_scram_admin_jaas.conf <<EOF
  2. KafkaClient {
  3. org.apache.kafka.common.security.scram.ScramLoginModule required
  4. username="admin"
  5. password="admin-succ";
  6. };
  7. EOFcat> kafka_client_scram_producer_jaas.conf <<EOF
  8. KafkaClient {
  9. org.apache.kafka.common.security.scram.ScramLoginModule required
  10. username="producer"
  11. password="producer-succ";
  12. };
  13. EOFcat> kafka_client_scram_consumer_jaas.conf <<EOF
  14. KafkaClient {
  15. org.apache.kafka.common.security.scram.ScramLoginModule required
  16. username="consumer"
  17. password="consumer-succ";
  18. };
  19. EOF

5.2 在启动脚本中引入JAAS文件

​ 进入kafka/bin文件夹,在 kafka-console-producer.sh 最上方添加(路径换成自己的):

  1. exportKAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/home/bigdata/apps/kafka/config/kafka_client_scram_producer_jaas.conf"

​ 进入kafka/bin文件夹,在 kafka-console-consumer.sh 最上方添加(路径换成自己的):

  1. exportKAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/home/bigdata/apps/kafka/config/kafka_client_scram_consumer_jaas.conf"

5.3 配置consumer.properties和producer.properties

  1. cat> consumer.properties <<EOF
  2. security.protocol=SASL_PLAINTEXT
  3. sasl.mechanism=SCRAM-SHA-256
  4. group.id= test-consumer-group
  5. EOFcat> producer.properties <<EOF
  6. security.protocol=SASL_PLAINTEXT
  7. sasl.mechanism=SCRAM-SHA-256
  8. EOF

5.4 创建配置文件command_config.properties

​ 该文件其实也是配置的连接Kafka的用户配置文件,但是该文件用于kafka-acls.sh,kafka-configs.sh等文件,同样的名称和路径可自定义,我这里还是把该文件放到解压的config目录中,内容如下:

  1. sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-succ";
  2. security.protocol=SASL_PLAINTEXT
  3. sasl.mechanism=SCRAM-SHA-256
  1. cat> command_config.properties <<EOF
  2. sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-succ";
  3. security.protocol=SASL_PLAINTEXT
  4. sasl.mechanism=SCRAM-SHA-256
  5. EOF

5.5 创建主题

  1. bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --topictest--partitions3 --replication-factor 1 --command-config ./config/command_config.properties
  2. bin/kafka-topics.sh --list --bootstrap-server 127.0.0.1:9092 --command-config ./config/command_config.properties

5.6 对生产者赋予写的权限

  1. bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 -add --allow-principal User:producer --operation Write --topictest --command-config ./config/command_config.properties

5.7 查看权限

  1. bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --list--topictest --command-config ./config/command_config.properties

5.8 对消费者赋予读的权限

  1. bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --allow-principal User:consumer --operation Read --topictest --command-config ./config/command_config.properties

5.9 对消费者赋予组的权限

  1. bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --allow-principal User:consumer --operation Read --topictest--group test-consumer-group --command-config ./config/command_config.properties

5.10 启动生产

  1. bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topictest--producer.config config/producer.properties

5.11 启动消费者

  1. bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topictest --from-beginning --consumer.config config/consumer.properties

6 使用ACL进行用户管理并进行队列相关操作

6.1 创建主题

  1. ./bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --topic tp_usera --partitions 3 --replication-factor 1 --command-config ./config/command_config.properties

6.2 添加用户

  1. ./bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 --alter --add-config 'SCRAM-SHA-256=[password=abcd],SCRAM-SHA-512=[password=abcd]' --entity-type users --entity-name usera --command-config ./config/command_config.properties

6.3 给指定用户添加指定topic【读和写权限】

  1. ./bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --allow-principal User:usera --operation Read --operation Write --topic tp_usera --command-config ./config/command_config.properties

6.4 给用户添加指定topic【消息生产权限】

  1. ./bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --allow-principal User:usera --producer --topic tp_usera --command-config ./config/command_config.properties

6.5 给用户添加指定topic【消息消费权限,并对消费者赋予组的权限】

  1. ./bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --add --allow-principal User:usera --consumer --topic tp_usera --group test-consumer-group --command-config ./config/command_config.properties

6.6 查看指定topic的开放权限信息(主要是查看对哪些用户开放了哪些权限)

  1. ./bin/kafka-acls.sh --bootstrap-server 127.0.0.1:9092 --list --topic tp_usera --command-config ./config/command_config.properties

6.7 创建的两个用户的两个JAAS文件

  1. kafka_client_scram_producer_usera_jaas.conf
  2. kafka_client_scram_consumer_usera_jaas.conf

​ 进入kafka/config文件夹,执行以下命令:

  1. cat> kafka_client_scram_producer_usera_jaas.conf <<EOF
  2. KafkaClient {
  3. org.apache.kafka.common.security.scram.ScramLoginModule required
  4. username="usera"
  5. password="abcd";
  6. };
  7. EOFcat> kafka_client_scram_consumer_usera_jaas.conf <<EOF
  8. KafkaClient {
  9. org.apache.kafka.common.security.scram.ScramLoginModule required
  10. username="usera"
  11. password="abcd";
  12. };
  13. EOF

6.8 在启动脚本中引入JAAS文件

​ 进入kafka/bin文件夹,新增 kafka-console-producer-usera.sh 文件,注意最上方添加(路径换成自己的):

  1. cp kafka-console-producer.sh kafka-console-producer-usera.sh
  2. vi kafka-console-producer-usera.sh

​ 修改“KAFKA_HEAP_OPTS”所在的行为下面的内容:

  1. export KAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/home/bigdata/apps/kafka/config/kafka_client_scram_producer_usera_jaas.conf"

​ 进入kafka/bin文件夹,新增 kafka-console-consumer-usera.sh 文件,注意最上方添加(路径换成自己的):

  1. cp kafka-console-consumer.sh kafka-console-consumer-usera.sh
  2. vi kafka-console-consumer-usera.sh

​ 修改“KAFKA_HEAP_OPTS”所在的行为下面的内容:

  1. export KAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=/home/bigdata/apps/kafka/config/kafka_client_scram_consumer_usera_jaas.conf"

6.9 启动生产者

  1. ./bin/kafka-console-producer-usera.sh --broker-list 127.0.0.1:9092 --topic tp_usera --producer.config config/producer.properties

6.10 启动消费者

  1. ./bin/kafka-console-consumer-usera.sh --bootstrap-server 127.0.0.1:9092 --topic tp_usera --from-beginning --consumer.config config/consumer.properties

参考文档

(1):Kafka配置动态SASL_SCRAM认证


本文转载自: https://blog.csdn.net/u010782920/article/details/140094740
版权归原作者 武舞悟 所有, 如有侵权,请联系我们删除。

“在Kafka3.7.0中进行SASL_SCRAM认证配置_不使用自带zookeeper”的评论:

还没有评论