Kafka安全认证机制详解之SASL_PLAIN

一、概述

官方文档：
https://kafka.apache.org/documentation/#security
在官方文档中，kafka有五种加密认证方式，分别如下：

• SSL：用于测试环境
  
• SASL/GSSAPI (Kerberos) ：利用kerberos认证，暗码是加密的，也是当前企业中利用最多的，最小支持版本0.9
  
• SASL/PLAIN ：利用简单用户名和暗码情势，生产环境中一般不利用，主要用于测试，最小支持版本0.10
  
• SASL/SCRAM：主要办理PLAIN动态更新问题以及安全机制，最小支持版本0.10.2
  
• SASL/OAUTHBEARER：基于OAuth2认证框架，最小支持版本2.0
  

几种认证方式的详细比较

• SASL/GSSAPI 主要是给 Kerberos 利用的。GSSAPI 适用于本身已经做了 Kerberos 认证的场景，如许的话，SASL/GSSAPI 可以实现无缝集成。
  
• SASL/PLAIN 是一个简单的用户名 / 暗码认证机制，通常与 SSL 加密搭配利用。对于一些小公司而言，搭建公司级的 Kerberos 可能并没有什么必要，他们的用户系统也不复杂，特别是访问 Kafka 集群的用户可能不是很多。对于 SASL/PLAIN 而言，这就是一个非常符合的应用场景。总体来说，SASL/PLAIN 的设置和运维成本相对较小，适合于小型公司中的 Kafka 集群。SASL/PLAIN 有如许一个毛病：它不能动态地增减认证用户，必须重启 Kafka 集群才能令变动生效。因为所有认证用户信息全部保存在静态文件中，所以只能重启 Broker，才能重新加载变动后的静态文件。
  
• SASL/SCRAM 通过将认证用户信息保存在 ZooKeeper 的方式，避免了动态修改需要重启 Broker 的毛病。在实际利用过程中，可以利用 Kafka 提供的命令动态地创建和删除用户，无需重启整个集群。因此，如果打算利用 SASL/PLAIN，不妨改用 SASL/SCRAM 试试。不过要注意的是，后者是 0.10.2 版本引入的。
  
• SASL/OAUTHBEARER 是 2.0 版本引入的新认证机制，主要是为了实现与 OAuth 2 框架的集成。 Kafka 不提倡单纯利用 OAUTHBEARER，因为它生成的不安全的 JSON Web Token，必须配以 SSL 加密才能用在生产环境中。
  
• Delegation Token 是在 1.1 版本引入的，它是一种轻量级的认证机制，主要目的是增补现有的 SASL 或 SSL 认证。 如果要利用 Delegation Token，需要先设置好 SASL 认证，然后再利用 Kafka 提供的 API 去获取对应的 Delegation Token。如许，Broker 和客户端在做认证的时候，可以直接利用这个 token，不用每次都去 KDC 获取对应的 ticket（Kerberos 认证）或传输 Keystore 文件（SSL 认证）。
  

二、SASL/PLAIN

SASL/PLAIN是基于用户名暗码的认证方式，是比较常用的一种认证，通常与TLS一起用于加密以实现安全认证。
增长kafka设置

1. vim /opt/kafka_2.13-2.6.0/config/server.properties
3. # 增加以下配置，每台节点都要配置
4. listeners=SASL_PLAINTEXT://host.name:port
5. security.inter.broker.protocol=SASL_PLAINTEXT
6. sasl.mechanism.inter.broker.protocol=PLAIN
7. sasl.enabled.mechanisms=PLAIN
9. # 分发配置,需要注意用户名和路径
10. scp /opt/kafka_2.13-2.6.0/config/server.properties root@kafka2-73085:/opt/kafka_2.13-2.6.0/config/
12. scp /opt/kafka_2.13-2.6.0/config/server.properties root@kafka3-73085:/opt/kafka_2.13-2.6.0/config/

复制代码

增长kafka的权限设置信息

1. # 编辑配置文件
2. vim /opt/kafka_2.13-2.6.0/config/kafka_server_jaas.conf
4. # 增加以下配置
5. # 增加了两个用户 admin用户和tly用户,配置文件中间不能有注释
6. KafkaServer {
7.      org.apache.kafka.common.security.plain.PlainLoginModule required
8.      username="admin"
9.      password="123456"
10.      user_admin="123456"
11.      user_tly="123456";
12. };
14. # 分发配置文件
15. scp kafka_server_jaas.conf root@kafka2-73085:/opt/kafka_2.13-2.6.0/config/
17. scp kafka_server_jaas.conf root@kafka3-73085:/opt/kafka_2.13-2.6.0/config/

复制代码

上面设置是新增了两个用户，admin和tly，这两个用户都是普通用户，KafkaServer中的username、password设置的用户和暗码，是用来broker和broker连接认证。在本例中，admin是署理broker间通讯的用户。user_userName设置为连接到broker的所有用户界说暗码，broker利用这些验证所有客户端连接，包括来自其他broker的连接。
将JAAS位置作为JVM参数通报给broker

1. # 编辑 kafka-run-class.sh
2. vim /opt/kafka_2.13-2.6.0/bin/kafka-run-class.sh
4. # 更改第219行
5. export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka_2.13-2.6.0/config/kafka_server_jaas.conf"
7. # 分发脚本
8. scp kafka-run-class.sh root@kafka2-73085:/opt/kafka_2.13-2.6.0/bin/kafka-run-class.sh      
9. scp kafka-run-class.sh root@kafka2-73085:/opt/kafka_2.13-2.6.0/bin/kafka-run-class.sh   

复制代码

启动kafka集群

1. kafka-server-start.sh -daemon /opt/kafka_2.13-2.6.0/config/server.properties

复制代码

这个时候集群启动完毕已经带有权限
客户端连接

检察topic列表

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092 --list

复制代码

迟迟不出结果，卡住了，实际上是没有权限

检察日志如下：

设置权限信息

1. # 编辑权限文件
2. vim /root/auth.conf
4. # 输入如下内容
5. security.protocol=SASL_PLAINTEXT
6. sasl.mechanism=PLAIN
7. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="tly" password="123456";

复制代码

username和password必须是kafka_server_jaas.conf中设置的。
利用带账号暗码设置文件连接

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092 --list
2. --command-config /root/auth.conf

复制代码

已经可以正常返回了，阐明有权限认证通过了：

创建topic

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092 --create --topic topic_1 --partitions 3 --replication-factor 3 --command-config /root/auth.conf

复制代码

生产消息

1. kafka-console-producer.sh  --bootstrap-server \
2. kafka1-73085:9092 --topic topic_1 --producer.config /root/auth.conf

复制代码

消耗消息

1. kafka-console-consumer.sh --bootstrap-server kafka1-73085:9092 \
2. --topic topic_1 --consumer.config /root/auth.conf

复制代码

权限设置

1. # 修改server.properties 增加如下配置
2. # 指定通过ACL来控制权限
3. authorizer.class.name=kafka.security.authorizer.AclAuthorizer
5. # 指定超级管理员
6. super.users=User:admin

复制代码

利用权限设置而且超级用户是admin，这个用户和我们上面设置的列表中的admin用户相互映射，每个机器都需要设置然后重启所有节点

1. # 配置用户认证文件信息
2. vim /root/admin.conf
3. # 输入如下内容
4. security.protocol=SASL_PLAINTEXT
5. sasl.mechanism=PLAIN
6. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="123456";
8. # 重启所有节点,每个节点都要执行
9. # 停止节点
10. kafka-server-stop.sh
12. # 启动节点
13. kafka-server-start.sh -daemon /opt/kafka_2.13-2.6.0/config/server.properties

复制代码

此时每个节点都有了权限认证

1. # 配置用户认证文件
2. vim /root/admin.conf
4. # 输入以下内容
5. security.protocol=SASL_PLAINTEXT
6. sasl.mechanism=PLAIN
7. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="123456";

复制代码

创建topic

这个时候利用auth.conf已经不能创建topic了，利用admin.conf可以创建，因为admin账号和设置的超级管理员一致；而auth.conf设置的是tly账号；

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092 --create --topic topic_2 --partitions 3 --replication-factor 3 --command-conf /root/auth.conf

复制代码

不利用认证创建会不停卡主不动，其实也是没权限的，时间久了就会报timeout异常

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092 --create --topic topic_3 --partitions 3 --replication-factor 3

复制代码

kafka-acls控制权限

命令为：kafka-acls.sh
参数如下：

• –add 增长权限 --remove 删除权限
  
• –allow-host 允许操作的主机 --deny-host 拒绝的主机
  
• –allow-principal 允许操作的用户 --deny-principal 拒绝用户
  
• –bootstrap-server 集群地点
  
• –command-config 认证文件信息
  
• –group 可以消耗topic的消耗者组
  
• –list 查询所有权限
  
• –operation 增长操作权限
  
  
  
  • describe
    
  • describeConfigs
    
  • alter
    
  • read
    
  • delete
    
  • create
    
  • all
    
  • write
    
  • alterconfig
    
  
  
• –topic 指定topic
  

查询topic所有权限

1. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --list  --command-config /root/admin.conf

复制代码

删除所有权限

1. kafka-acls.sh --bootstrap-server  kafka1-73085:9092 --topic topic_1 --remove  --command-config /root/admin.conf

复制代码

给用户增长权限

给之前的用户tly授权

1. kafka-acls.sh --bootstrap-server  kafka1-73085:9092  --add --allow-principal User:tly --topic topic_1 --operation all --command-config /root/admin.conf

复制代码

利用tly用户检察当前topic

1. kafka-topics.sh --bootstrap-server kafka1-73085:9092  --list --command-config /root/auth.conf

复制代码

采取权限

1. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --remove --allow-principal user:tly --topic topic_1 --operation all --command-config /root/admin.conf

复制代码

增长权限而且增长IP限定

1. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --add  --deny-principal User:tly --topic topic_1 --operation all --deny-host 127.0.0.1 --command-config admin.conf

复制代码

生产者和消耗者的权限设置

1. # 删除所有权限信息
2. kafka-acls.sh --bootstrap-server kafka1-73085:9092  --topic topic_1 --remove --command-config /root/admin.conf
4. # 添加生产者权限
5. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --add --allow-principal User:tly --operation Write --topic topic_1 --command-config /root/admin.conf
7. # 添加消费权限
8. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --add --allow-principal User:tly --operation Read --topic topic_1 --command-config /root/admin.conf
10. # 消费消息
11. kafka-console-consumer.sh --bootstrap-server kafka1-73085:9092 --topic topic_1 --from-beginning --consumer.config /root/auth.conf
13. # 发现消费不到，原因是消费者组没有权限
14. # 增加消费者组权限
15. kafka-acls.sh --bootstrap-server kafka1-73085:9092 --add --allow-principal User:tly --operation Read --topic topic_1 --group group_1 --command-config /root/admin.conf
17. # 再次执行消费消息命令并增加分组信息即可以消费到数据
18. kafka-console-consumer.sh --bootstrap-server kafka1-73085:9092 --topic topic_1 --from-beginning --consumer.config /root/auth.conf --group group_1

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。