1. 云图网首页
  2. 技术专区
  3. 大数据

实战Kafka ACL机制详解大数据

• 大数据

1.概述

　　在Kafka0.9版本之前，Kafka集群时没有安全机制的。Kafka Client应用可以通过连接Zookeeper地址，例如zk1:2181:zk2:2181,zk3:2181等。来获取存储在Zookeeper中的Kafka元数据信息。拿到Kafka Broker地址后，连接到Kafka集群，就可以操作集群上的所有主题了。由于没有权限控制，集群核心的业务主题时存在风险的。

2.内容

2.2 身份认证

　　Kafka的认证范围包含如下：

  • Client与Broker之间
  • Broker与Broker之间
  • Broker与Zookeeper之间

　　当前Kafka系统支持多种认证机制，如SSL、SASL（Kerberos、PLAIN、SCRAM）。

2.3  SSL认证流程

　　在Kafka系统中，SSL协议作为认证机制默认是禁止的，如果需要使用，可以手动启动SSL机制。安装和配置SSL协议的步骤，如下所示：

  1. 在每个Broker中Create一个Tmp密钥库
  2. 创建CA
  3. 给证书签名
  4. 配置Server和Client

　　执行脚本如下所示：

#! /bin/bash 
 
# 1.Create rsa 
keytool -keystore server.keystore.jks -alias dn1 -validity 365 -genkey -keyalg RSA 
# 2.Create CA 
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365 
# 3.Import client 
keytool -keystore client.truststore.jks -alias CAROOT -import -file ca-cert 
# 4.Import server 
keytool -keystore server.truststore.jks -alias CAROOT -import -file ca-cert 
# 5.Export 
keytool -keystore server.keystore.jks -alias dn1 -certreq -file cert-file 
# 6.Signed 
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:123456 
# 7.Import ca-cert 
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert 
# 8.Import cert-signed 
keytool -keystore server.keystore.jks -alias dn1 -import -file cert-signed

2.4 SASL认证流程

　　在Kafka系统中，SASL机制包含三种，它们分别是Kerberos、PLAIN、SCRAM。以PLAIN认证为示例，下面给大家介绍PLAIN认证流程。

2.4.1  配置Server

　　首先，在$KAFKA_HOME/config目录中新建一个文件，名为kafka_server_jaas.conf，配置内容如下：

KafkaServer { 
   org.apache.kafka.common.security.plain.PlainLoginModule required 
   username="smartloli" 
   password="smartloli-secret" 
   user_admin="smartloli-secret"; 
}; 
 
Client { 
   org.apache.kafka.common.security.plain.PlainLoginModule required 
   username="smartloli" 
   password="smartloli-secret"; 
};

　　然后在Kafka启动脚本（kafka-server-start.sh）中添加配置文件路径，设置内容如下：

[[email protected] bin]$ vi kafka-server-start.sh 
 
# Add jaas file 
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka/config/kafka_server_jaas.conf"

　　接下来，配置server.properties文件，内容如下：

# Set ip & port 
listeners=SASL_PLAINTEXT://dn1:9092 
advertised.listeners=SASL_PLAINTEXT://dn1:9092 
# Set protocol 
security.inter.broker.protocol=SASL_PLAINTEXT 
sasl.enabled.mechanisms=PLAIN 
sasl.mechanism.inter.broker.protocol=PLAIN 
 
# Add acl 
allow.everyone.if.no.acl.found=true 
auto.create.topics.enable=false 
delete.topic.enable=true 
advertised.host.name=dn1 
super.users=User:admin 
 
# Add class 
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

2.4.2 配置Client

　　当Kafka Server端配置启用了SASL/PLAIN，那么Client连接的时候需要配置认证信息，Client配置一个kafka_client_jaas.conf文件，内容如下：

KafkaClient { 
  org.apache.kafka.common.security.plain.PlainLoginModule required 
  username="admin" 
  password="admin-secret"; 
};

　　然后，在producer.properties和consumer.properties文件中设置认证协议，内容如下：

security.protocol=SASL_PLAINTEXT  
sasl.mechanism=PLAIN

　　最后，在kafka-console-producer.sh脚本和kafka-console-producer.sh脚本中添加JAAS文件的路径，内容如下：

# For example: kafka-console-producer.sh 
[email protected] bin]$ vi kafka-console-producer.sh 
 
# Add jaas file 
export KAFKA_OPTS="-Djava.security.auth.login.config=/data/soft/new/kafka/ 
/config/kafka_client_jaas.conf"

2.5 ACL操作

　　在配置好SASL后，启动Zookeeper集群和Kafka集群之后，就可以使用kafka-acls.sh脚本来操作ACL机制。

　　（1）查看：在kafka-acls.sh脚本中传入list参数来查看ACL授权新

[[email protected] bin]$ kafka-acls.sh --list --authorizer-properties zookeeper.connect=dn1:2181

　　（2）创建：创建待授权主题之前，在kafka-acls.sh脚本中指定JAAS文件路径，然后在执行创建操作

[[email protected] bin]$ kafka-topics.sh --create --zookeeper dn1:2181 --replication-factor 1 --partitions 1 --topic kafka_acl_topic

　　（3）生产者授权：对生产者执行授权操作

[[email protected] ~]$ kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=dn1:2181 --add --allow-principalUser:producer --operation Write --topic kafka_acl_topic

　　（4）消费者授权：对生产者执行授权后，通过消费者来进行验证

[[email protected] ~]$ kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=dn1:2181 --add --allow-principalUser:consumer --operation Read --topic kafka_acl_topic

　　（5）删除：通过remove参数来回收相关权限

[[email protected] bin]$ kafka-acls.sh --authorizer-properties zookeeper.connect=dn1:2181 --remove --allow-principal User:producer --operation Write --topic kafka_acl_topic3

3.总结

　　在处理一些核心的业务数据时，Kafka的ACL机制还是非常重要的，对核心业务主题进行权限管控，能够避免不必要的风险。

4.结束语

　　这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

原创文章，作者：奋斗，如若转载，请注明出处：https://blog.ytso.com/9877.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论