Apache zookeeper kafka 开启SASL安全认证-Toy模板网

这篇具有很好参考价值的文章主要介绍了Apache zookeeper kafka 开启SASL安全认证。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

背景：我之前安装的kafka没有开启安全鉴权，在没有任何凭证的情况下都可以访问kafka。搜了一圈资料，发现有关于sasl、acl相关的，准备试试。

简介

Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发，支持多语言（如Java、Python、Go等）客户端，它可以水平扩展和具有高吞吐量特性而被广泛使用，并与多类开源分布式处理系统进行集成使用。

Kafka作为一款开源的、轻量级的、分布式、可分区和具备复制备份的、基于ZooKeeper协调管理的分布式流平台的功能强大的消息系统。与传统消息系统相比，Kafka能够更好的处理活跃的流数据，让数据在各个子系统中高性能、低延迟地不停流转。

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

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服务

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

SCRAM认证配置的优点:

如果使用PLAIN认证有个问题，就是不能动态新增用户，每次添加用户后，需要重启正在运行的Kafka集群才能生效。

因此，在生产环境中，这种认证方式不符合实际业务场景，不利于后期扩展。然而使用SCRAM认证，可以动态新增用户，添加用户后，可以不用重启正在运行的Kafka集群即可进行鉴权。所以生产环境推荐使用SCRAM+PLAIN搭配的认证方案。

配置zookeeper集群启用SASL

1. 配置zookeeper，启用sasl认证，cat zoo.cfg查看到如下内容：

tickTime=2000
initLimit=1
syncLimit=5
dataDir=/tmp/zookeeper/data
dataLogDir=/tmp/zookeeper/log
clientPort=2181
admin.serverPort=8888
maxClientCnxns=3000
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
server.1=localhost:2888:3888
4lw.commands.whitelist=conf,stat,srvr,mntr.envi
#zk SASL
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

2. 配置zookeeper JAAS

cat zk_jaas.conf文件内容如下，如果没有改文件则使用vi命令编辑


Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="admin123"
    user_kafka="kafka123";
};

注意：admin用户是zk 集群之间使用的。kafka用户是 broker 与 zk 之间使用的。

3. 修改zkEnv.sh

将上一步添加的 jaas 配置文件添加到zookeeper的环境变量中，zkEnv.sh文件最后添加一行：
 
vim zkEnv.sh
 
ZOOBINDIR="${ZOOBINDIR:-/usr/bin}"
ZOOKEEPER_PREFIX="${ZOOBINDIR}/.."
 
# 添加如下 新增变量SERVER_JVMFLAGS：
 
export SERVER_JVMFLAGS="-Djava.security.auth.login.config=../conf/zk_jaas.conf"

配置kafka sasl动态认证

SASL/SCRAM认证是把凭证(credential)存储在Zookeeper，使用kafka-configs.sh在Zookeeper中创建凭据。对于每个SCRAM机制，必须添加具有机制名称的配置来创建凭证，所以在启动Kafka broker之前需要创建代理间通信的凭据。

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

1. 创建用户

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

./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name admin
 
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name producer
 
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin123],SCRAM-SHA-512=[password=admin123]' --entity-type users --entity-name consumer

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

2. 查看创建的用户信息

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

./kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users 
 
#（可以单独指定某个用户 --entity-name producer，如下）
./kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name producer

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

ZK客户端命令行查看：
 
./zkCli.sh  -server localhost:2181
 
ls /config/users

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

3. 配置kafka jaas文件

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

Kafka 的 jaas认证配置文件，配置的是登录类，超管密码和管理的帐号密码列表

vim kafka_server_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username ="admin"
    password="admin123"
    user_admin="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
Client {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka"
    password="kafka123";
};

KafkaServer中usename配置的是kafka服务端使用的账号和密码，后面的user_xxx事预设的普通帐号认证信息。

中间部分配置的是PLAIN认证方式的账户和密码，其中producer1是账户名，producer123是密码。
Client配置了broker到Zookeeper的连接用户名密码，这里要和前面zookeeper配置中的zk_jaas.conf.conf 中 user_kafka 的账号和密码相同。

关于这个文件内容，需要注意以下两点：

1）不要忘记最后一行和倒数第二行结尾处的分号；

2）JAAS 文件中不需要任何空格键。

4. kafka 配置文件启用SASL认证

Kafka 服务配置文件 server.propertis，配置认证协议及认证实现类
cat server.properties其它内容都注释掉，然后追加如下内容：

broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://localhost:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka/logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
delete.topic.enable=true
auto.create.topics.enable=false
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

Host.name=43.138.0.199

5. kafka 启动脚本添加认证文件路径的环境变量

Kafka 安全认证可以直接通过环境变量 -Djava.security.auth.login.config 设置，修改 Kafka 启动脚本 kafka-start-server.sh 文件最后一行，增加一个参数指向 jaas 配置文件的绝对路径

vi kafka-server-start.sh

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/home/lighthouse/kafka_2.12-2.2.1/config/kafka_server_jaas.conf  kafka.Kafka "$@"

6. kafka客户端配置

1) 配置consumer.properties和producer.properties,都要加入以下配置

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512

2) 生产者配置

使用kafka-console-producer.sh脚本测试生产者，由于开启安全认证和授权，此时使用console-producer脚本来尝试发送消息，那么消息会发送失败，原因是没有指定合法的认证用户，因此客户端需要做相应的配置，需要创建一个名为producer.conf的配置文件给producer程序使用。

config目录下创建一个producer.conf的文件，cat producer.conf文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.Scra
mLoginModule required username="producer" password="producer123";

注意：Topic设置写权限

3) 消费者配置

使用kafka-console-consumer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。需要为 consumer 用户创建consumer.conf给消费者程序，同时设置对topic的读权限。

config目录下创建一个consumer.conf的文件，cat consumer.conf文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.Scra
mLoginModule required username="consumer" password="consumer123";

注意：Topic设置读权限。

4) 在生产者和消费者启动脚本中引入JAAS文件

vim bin/kafka-console-producer.sh
 
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
 
# 添加这行
export KAFKA_OPTS="-Djava.security.auth.login.config=../config/kafka_server_jaas.conf"
 
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
 
 
# vim bin/kafka-console-consumer.sh
 
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
 
# 添加这行
export KAFKA_OPTS="-Djava.security.auth.login.config=../config/kafka_server_jaas.conf"
 
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"

启动kafka

nohup kafka-server-start.sh /path-to-kafka/config/server.properties &

我自己写了一个脚本，同时启动zookeeper和kafka

文件名称叫startup.sh内容如下：

cd /home/lighthouse/zk-3.4.14/zookeeper-3.4.14/bin
./zkServer.sh start
cd /home/lighthouse/kafka_2.12-2.2.1/
nohup bin/kafka-server-start.sh config/server.properties > output.txt &

zookeeper没有启动成功，我找下原因。把整个过程重新整理了一遍，发现zookeeper、kafka启动成功了。

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

检查验证

./zkCli.sh  -server localhost:2181

ls /brokers/ids

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

1. 生产者测试

1）创建一个测试主题test_topic

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

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全

2）查看创建的Topic

./kafka-topics.sh  --list --zookeeper localhost:2181 test_topic

Apache zookeeper kafka 开启SASL安全认证,JAVA,zookeeper,kafka,安全文章来源地址https://www.toymoban.com/news/detail-839621.html

到了这里，关于Apache zookeeper kafka 开启SASL安全认证的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

Apache zookeeper kafka 开启SASL安全认证

简介

配置zookeeper集群启用SASL

1. 配置zookeeper，启用sasl认证，cat zoo.cfg查看到如下内容：

2. 配置zookeeper JAAS

3. 修改zkEnv.sh

配置kafka sasl动态认证

1. 创建用户

2. 查看创建的用户信息

3. 配置kafka jaas文件

4. kafka 配置文件启用SASL认证

5. kafka 启动脚本添加认证文件路径的环境变量

6. kafka客户端配置

启动kafka

检查验证

1. 生产者测试

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

支付宝扫一扫领取红包，优惠每天领

二维码1

二维码2