kafka zk 使用SASL,PLAIN认证

zookeeper配置

  1. 修改zoo.cfg增加三行配置
  1. authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  2. requireClientAuthScheme=sasl
  3. jaasLoginRenew=360000
  1. 配置JAAS文件：conf目录下创建zk_server_jaas.conf（定义了需要链接到Zookeeper服务器的用户名和密码）

vim zk_server_jaas.conf

  1. Server {
  2.   org.apache.kafka.common.security.plain.PlainLoginModule required
  3.   username="admin"
  4.   password="admin-sec"
  5.   user_admin="admin-sec";
  6. };
  1. 加入需要的包：（从kafka/libs下拷贝到zookeeper/lib）
  1. kafka-clients-0.10.0.1.jar
  2. lz4-1.3.0.jar
  3. slf4j-api-1.7.21.jar
  4. slf4j-log4j12-1.7.21.jar
  5. snappy-java-1.1.2.6.jar
  1. 修改zkEnv.sh(在最后一行新增)
  1. export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=$ZOOKEEPER_PREFIX/conf/zk_server_jaas.conf"
  1. 启动Zookeeper
  1. /opt/lingmou/zookeeper/bin/zkServer.sh start

kafka服务的配置

  1. 配置server.properties
  1. listeners=SASL_PLAINTEXT://192.168.101.130:6667
  2. advertised.listeners=SASL_PLAINTEXT://192.168.101.130:6667
  4. # set protocol
  5. security.inter.broker.protocol=SASL_PLAINTEXT
  7. # SASL机制
  8. sasl.enabled.mechanisms=PLAIN
  9. sasl.mechanism.inter.broker.protocol=PLAIN
  11. # 完成身份认证的类
  12. authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
  14. # 如果没有找到ACL（访问控制列表）配置，则允许任何操作
  15. allow.everyone.if.no.acl.found=true
  17. super.users=User:admin
  18. security.protocol=SASL_PLAINTEXT
  1. kafka增加认证信息：config/kafka_server_jaas.conf
  1. KafkaServer {
  2.   org.apache.kafka.common.security.plain.PlainLoginModule required
  3.   username="admin"            # username是broker用于初始化连接到其他的broker
  4.   password="admin-sec"
  5.   user_admin="admin-sec"      # user_userName定义了所有连接到broker和broker验证的所有的客户端连接包括其他broker的用户密码
  6.   user_producer="prod-sec"
  7.   user_consumer="cons-sec";
  8. };
  10. Client {
  11.   org.apache.kafka.common.security.plain.PlainLoginModule required
  12.   username="admin"            # 就是broker与zookeeper通信时登录zookeeper的用户名密码
  13.   password="admin-sec";
  14. };
  1. 修改启动脚本 bin/kafka-server-start.sh
  1. exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf kafka.Kafka "$@"
  1. 启动kafka
  1. /opt/lingmou/kafka/bin/kafka-server-start.sh -daemon /opt/lingmou/kafka/config/server.properties
  1. kafka客户端配置
  1. # 消费者：conf/kafka-consumer-jaas.conf
  3. KafkaClient {
  4.   org.apache.kafka.common.security.plain.PlainLoginModule required
  5.   username="consumer"
  6.   password="cons-sec";
  7. };
  9. Client {
  10.   org.apache.kafka.common.security.plain.PlainLoginModule required
  11.   username="consumer"
  12.   password="cons-sec";
  13. };
  15. # 生产者：conf/kafka-producer-jaas.conf
  17. KafkaClient {
  18.   org.apache.kafka.common.security.plain.PlainLoginModule required
  19.   username="producer"
  20.   password="prod-sec";
  21. };
  23. Client {
  24.   org.apache.kafka.common.security.plain.PlainLoginModule required
  25.   username="producer"
  26.   password="prod-sec";
  27. };
  1. 修改客户端配置信息
  1. # 消费者：config/producer.properties
  3. bootstrap.servers=192.168.101.130:6667
  4. compression.type=none
  5. # 添加认证机制
  6. security.protocol=SASL_PLAINTEXT
  7. sasl.mechanism=PLAIN
  9. # 生产者：config/consumer.properties 
  11. zookeeper.connect=192.168.101.130:2181
  12. zookeeper.connection.timeout.ms=6000
  13. group.id=test-group
  14. # 添加认证机制
  15. sasl.mechanism=PLAIN
  16. security.protocol=SASL_PLAINTEXT
  1. 修改客户端脚本指定JAAS文件加载：
  1. # 消费者 bin/kafka-console-consumer.sh
  3. 修改
  4. exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
  5. 为
  6. exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka-consumer-jaas.conf kafka.tools.ConsoleConsumer "$@"
  8. # 生产者 bin/kafka-console-producer.sh
  10. 修改
  11. exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
  12. 为
  13. exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=$(dirname $0)/../config/kafka-producer-jaas.conf kafka.tools.ConsoleProducer "$@"
  1. 进行授权
  1. 1）创建主题 haha
  2. ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic haha
  4. 2）增加生产权限
  5. ./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:producer --operation Write --topic haha
  7. 3）配置消费权限
  8. ./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --topic haha
  10. 4）配置消费分组权限
  11. ./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --group test-group
  13. 5）查看配置的权限
  14. ./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list
  16. 6）取消权限
  17. ./bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --remove --allow-principal User:producer --operation Write --topic haha
  19. 7) 拒绝来自 ip 为192.168.1.100账户为 zhangsan 进行 read 操作，其他用户都允许
  20. ./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:* --allow-host * --deny-principal User:zhangsan --deny-host 192.168.1.100 --operation Read --topic haha
  22. 8) 允许来自 ip 为192.168.1.100或者192.168.1.101的读写请求
  23. ./bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:zhangsan --allow-principal User:alice --allow-host 192.168.1.100 --allow-host 192.168.1.101 --operation Read --operation Write --topic haha
  1. 测试
  1. 1）生产数据
  2. ./bin/kafka-console-producer.sh --topic haha --broker-list localhost:6667 --producer.config config/producer.properties
  4. 2）消费数据
  5. ./bin/kafka-console-consumer.sh --topic haha --bootstrap-server localhost:6667 --consumer.config config/consumer.properties