一 概述

1.1 Kafka的权限分类

1）身份认证（Authentication）：对client 与服务器的连接进行身份认证，brokers和zookeeper之间的连接进行Authentication（producer 和 consumer）、其他 brokers、tools与 brokers 之间连接的认证。

2）权限控制（Authorization）：实现对于消息级别的权限控制，clients的读写操作进行Authorization：（生产/消费/group)数据权限。

1.2 实现方式

自0.9.0.0版本开始Kafka社区添加了许多功能用于提高Kafka集群的安全性，Kafka提供SSL或者SASL两种安全策略。SSL方式主要是通过CA令牌实现，此方案主要介绍SASL方式。
1）SASL验证:

验证方式	Kafka版本	特点
SASL/PLAIN	0.10.0.0	不能动态添加用户
SASL/SCRAM	0.10.2.0	可以动态添加用户
SASL/Kerberos	0.9.0.0	需要独立部署验证服务
SASL/OAUTHBEARER	2.0.0	需自己实现接口实现token的创建和验证，需要额外的oauth服务

2）SSL加密: 使用SSL加密在代理和客户端之间，代理之间或代理和工具之间传输的数据。

二 安全认证授权配置

本文档主要演示SASL/SCRAM和SASL/PLAIN 搭配使用的方案。版本：kafka_2.12-2.5.1.tgz、apache-zookeeper-3.5.8-bin.tar.gz

2.1 zookeeper配置

2.1.1 引入kafka依赖包

将kafka包下面的相关依赖包复制到zookeeper的目录下，不同的kafka
版本jar包版本会不一样。

cp /mnt/datadisk/kafka/libs/kafka-clients-2.5.1.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/lz4-java-1.7.1.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/slf4j-api-1.7.30.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/slf4j-log4j12-1.7.30.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/snappy-java-1.1.7.3.jar /mnt/datadisk/zookeeper/lib

2.1.2修改ZK配置文件

Zookeeper的配置文件zoo.cfg中添加如下内容：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

2.1.3 创建jaas文件

在zookeeper的conf目录下添加安全认证的jaas文件，该文件定义需要链接到Zookeeper服务器的用户名和密码。这里定义文件名为zk-server-jaas.conf，文件内容为：

Server {
    
org.apache.kafka.common.security.plain.PlainLoginModule required
username="zoo_admin"
password="zoo_admin"
user_kafka="kafka";
};

该文件中username和password是定义zookeeper集群节点之间认证的用户名和密码；user_开头配置的用户kafka和密码kafka是提供给外部应用连接zookeeper使用的，后面kafka使用此用户连接zookeeper。

2.1.4 修改zkEnv.sh

将上一步添加的jaas配置文件添加到zookeeper的环境变量中，zkEnv.sh文件最后添加一行：

export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/mnt/datadisk/zookeeper/conf/zk_server_jaas.conf"

2.1.5 启动zk

nohup bin/zkServer.sh start &

2.2 Kafka配置

Kafka和生产者/消费者之间采用SASL/PLAIN和SASL/SCRAM两种方式共同完成认证，授权使用ACL方式。PLAIN方式的用户是在jaas文件中写死的，不能动态的添加；SCRAM支持动态的添加用户。

2.2.1 创建超级用户

配置SASL/SCRAM认证的第一步，是配置可以连接到kafka集群的用户。本方案演示创建3个用户：kafka_server_admin，writer，reader。kafka_server_admin用户用于broker之间的认证通信，writer用户用于生产者连接kafka，reader用户用于消费者连接kafka。

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name kafka_server_admin

正常情况打印信息：Completed updating config for entity: user-principal ‘admin’.

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --alter --add-config 'SCRAM-SHA-256=[password=writer],SCRAM-SHA-512=[password=writer]' --entity-type users --entity-name writer

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --alter --add-config 'SCRAM-SHA-256=[password=reader],SCRAM-SHA-512=[password=reader]' --entity-type users --entity-name reader

在zk客户端中可以查看创建成功的用户：

bin/zkCli.sh
# 客户端打开后执行下面命令，查看用户节点，可以看到创建成功的用户
ls /config/users

kafka-configs 脚本是用来设置主题级别参数的。其实，它的功能还有很多。比如在这个例子中，我们使用它来创建 SASL/SCRAM 认证中的用户信息。可以使用下列命令来查看刚才创建的用户数据。

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --describe --entity-type users --entity-name writer

这段命令包含了 writer 用户加密算法 SCRAM-SHA-256 以及 SCRAM-SHA-512 对应的盐值 (Salt)、ServerKey 和 StoreKey，这些都是 SCRAM 机制的术语。

2.2.2 创建jaas文件

配置了用户之后，我们需要为 Broker 创建一个对应的 JAAS 文件。在实际场景中，需要为每台单独的物理 Broker 机器都创建一份 JAAS 文件。

# 在kafka目录下创建文件
vi config/kafka_server_jaas.conf

kafka_server_jaas.conf文件的内容为：

KafkaServer {
    
org.apache.kafka.common.security.scram.ScramLoginModule required
username="kafka_server_admin"
password="admin";

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

Client {
    
org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafka"
password="kafka";
};

KafkaServer配置的是kafka的账号和密码。Client配置了broker到Zookeeper的连接用户名密码，这里要和前面2.1.3节zookeeper配置中的zk_server_jaas.conf中user_kafka的账号和密码相同。中间部分配置的是PLAIN认证方式的账户和密码，其中filebeat是账户名，123456是密码。
关于这个文件内容，需要注意以下两点：

1. 不要忘记最后一行和倒数第二行结尾处的分号；
2. JAAS 文件中不需要任何空格键。

2.2.3 修改kafka配置文件

server.properties文件中添加如下内容:

# 启用ACL
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
# 设置本例中admin为超级用户；在Zookeeper的“/kafka/config/users”下存在用户
super.users=User:kafka_server_admin
# 同时启用SCRAM和PLAIN机制
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
# 为broker间通讯开启SCRAM机制，采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
# broker间通讯使用PLAINTEXT，本例中不演示SSL配置
security.inter.broker.protocol=SASL_PLAINTEXT
# 配置listeners使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://:9092
# 配置advertised.listeners
advertised.listeners=SASL_PLAINTEXT://192.168.13.202:9092

启动脚本kafka-server-start.sh文件最后一行注释掉，修改为：

#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=/mnt/datadisk/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

kafka配置文件server.properties完整配置如下：

# kafka不同节点的唯一标识
broker.id=0

#认证配置
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
#ACL配置
allow.everyone.if.no.acl.found=false
super.users=User:kafka_server_admin
authorizer.class.name=kafka.security.authorizer.AclAuthorizer

# ip为本机ip
advertised.listeners=SASL_PLAINTEXT://192.168.13.202:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 必改项 数据和日志分离, kafka中数据的存放目录
log.dirs=/mnt/datadisk/kafka/data
# 默认的分区数量
num.partitions=3
# 默认的副本数量
default.replication.factor=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
# zookeeper集群地址
zookeeper.connect=192.168.13.202:2181
zookeeper.connection.timeout.ms=90000
# 可以删除topic
delete.topic.enable=true
group.initial.rebalance.delay.ms=0
# 消息体大小
message.max.bytes=10485760
socket.request.max.bytes=524288000
replica.lag.time.max.ms=120000

2.2.4启动kafka

nohup bin/kafka-server-start.sh config/server.properties &

三 测试连接

3.1 生产者测试

使用kafka-console-producer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。config目录下创建一个producer.conf的文件，文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="writer" password="writer";

创建一个测试主题test_topic：

bin/kafka-topics.sh --zookeeper 192.168.13.202:2181 --create --replication-factor 1 --partitions 1 --topic test_topic

这里使用writer用户认证授权，通过ACL为writer用户分配操作test_topic权限（下面两个命令二选一即可）：
1.分配写的权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:writer --operation Write --topic 'test_topic'

2.分配producer权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:writer --producer --topic 'test_topic'

启动生产者发送消息：

bin/kafka-console-producer.sh --broker-list 192.168.13.202:9092 --topic test_topic --producer.config /mnt/datadisk/kafka/config/producer.conf

3.2 消费者测试

使用kafka-console-consumer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。config目录下创建一个consumer.conf的文件，文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="reader";

这里使用reader用户认证授权，通过ACL为reader用户分配操作test_topic权限（下面两个命令二选一即可）：
1.分配读的权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --operation Read --topic 'test_topic'

2.分配consumer的权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --consumer --topic test_topic --group test_group

启动消费者消费消息：

bin/kafka-console-consumer.sh --bootstrap-server 192.168.13.202:9092 --topic test_topic --group 'test_group' --from-beginning --consumer.config /mnt/datadisk/kafka/config/consumer.conf

3.3 beats工具连接

kafka官网提供了许多beats采集工具，用于不同场景下采集数据，感兴趣的可以去官网了解下，这里就不介绍了。笔者这里使用的是beats工具是7.9版本的，采集的数据如果要发送到kafka，只支持SASL/PLAIN认证方式，据说高版本的beats工具支持SCRAM方式的。这里介绍下filebeat-7.9.1的配置方式。PLAIN类型的用户在2.2.2章节已经创建，在filebeat中用此用户名和密码即可完成认证，授权参考3.1章节。
filebeat的配置文件filebeat.yml中输出kafka配置中加入用户名和密码，如下：

可以启动一个消费者监控filebeat采集的数据，启动filebeat，并在filebeat采集的文件中输入内容，可以看到消费者可以获取对应的消息：

3.4 logstash消费

logstash可以使用SCRAM方式认证，logstash的config目录下新增kafka-client-jaas.conf文件，文件内容如下：

KafkaClient {
    
org.apache.kafka.common.security.scram.ScramLoginModule required
username="reader"
password="reader";
};

修改congfig目录下对应的conf处理文件，在input模块中引入SCRAM认证：

security_protocol => "SASL_PLAINTEXT"
sasl_mechanism => "SCRAM-SHA-256"
jaas_path => "/mnt/datadisk/logstash/config/kafka-client-jaas.conf"

启动生产者往logstash监控的topic中发送数据，测试如下：

下一篇：【Java版 Kafka ACL使用实战】