背景：我之前安裝的kafka沒有開啟安全鑒權(quán)，在沒有任何憑證的情況下都可以訪問kafka。搜了一圈資料，發(fā)現(xiàn)有關(guān)于sasl、acl相關(guān)的，準備試試。
簡介
Kafka是一個高吞吐量、分布式的發(fā)布-訂閱消息系統(tǒng)。Kafka核心模塊使用Scala語言開發(fā)，支持多語言（如Java、Python、Go等）客戶端，它可以水平擴展和具有高吞吐量特性而被廣泛使用，并與多類開源分布式處理系統(tǒng)進行集成使用。
? Kafka作為一款開源的、輕量級的、分布式、可分區(qū)和具備復(fù)制備份的、基于ZooKeeper協(xié)調(diào)管理的分布式流平臺的功能強大的消息系統(tǒng)。與傳統(tǒng)消息系統(tǒng)相比，Kafka能夠更好的處理活躍的流數(shù)據(jù)，讓數(shù)據(jù)在各個子系統(tǒng)中高性能、低延遲地不停流轉(zhuǎn)。
自0.9.0.0版本開始Kafka社區(qū)添加了許多功能用于提高Kafka集群的安全性，Kafka提供SSL或者SASL兩種安全策略。SSL方式主要是通過CA令牌實現(xiàn)，此方案主要介紹SASL方式。
SASL驗證分類
驗證方式?? ?kafka版本?? ?特點
SASL/PLAIN?? ?0.10.0.0?? ?不能動態(tài)添加用戶
SASL/SCRAM?? ?0.10.2.0?? ?可以動態(tài)添加用戶
SASL/Kerberos?? ?0.9.0.0?? ?需要獨立部署驗證服務(wù)
SASL/oauthbearer?? ?2.0.0?? ?需要自己實現(xiàn)接口，實現(xiàn)token的創(chuàng)建和驗證，需要額外的oauth服務(wù)
使用SSL加密在代理和客戶端之間，代理之間或代理和工具之間傳輸?shù)臄?shù)據(jù)
SCRAM認證配置的優(yōu)點:
? 如果使用PLAIN認證有個問題，就是不能動態(tài)新增用戶，每次添加用戶后，需要重啟正在運行的Kafka集群才能生效。
? 因此，在生產(chǎn)環(huán)境中，這種認證方式不符合實際業(yè)務(wù)場景，不利于后期擴展。然而使用SCRAM認證，可以動態(tài)新增用戶，添加用戶后，可以不用重啟正在運行的Kafka集群即可進行鑒權(quán)。所以生產(chǎn)環(huán)境推薦使用SCRAM+PLAIN搭配的認證方案。
配置zookeeper集群啟用SASL
1. 配置zookeeper，啟用sasl認證，cat zoo.cfg查看到如下內(nèi)容：
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=true2. 配置zookeeper?JAAS
cat zk_jaas.conf文件內(nèi)容如下，如果沒有改文件則使用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的環(huán)境變量中，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動態(tài)認證
SASL/SCRAM認證是把憑證(credential)存儲在Zookeeper，使用kafka-configs.sh在Zookeeper中創(chuàng)建憑據(jù)。對于每個SCRAM機制，必須添加具有機制名稱的配置來創(chuàng)建憑證，所以在啟動Kafka broker之前需要創(chuàng)建代理間通信的憑據(jù)。
這里配置的 Kafka和生產(chǎn)者/消費者之間 采用SASL/PLAIN和SASL/SCRAM兩種方式共同完成認證，授權(quán)使用ACL方式。PLAIN方式的用戶是在jaas文件中寫死的，不能動態(tài)的添加；SCRAM支持動態(tài)的添加用戶。
1. 創(chuàng)建用戶
配置SASL/SCRAM認證的第一步，是配置可以連接到kafka集群的用戶。本案例創(chuàng)建了3個用戶：admin，producer，consumer。kafka_server_admin用戶用于broker之間的認證通信，producer用戶用于生產(chǎn)者連接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
?
?
2. 查看創(chuàng)建的用戶信息
kafka-configs 腳本是用來設(shè)置主題級別參數(shù)的。其實，它的功能還有很多。比如在這個例子中，我們使用它來創(chuàng)建 SASL/SCRAM 認證中的用戶信息?？梢允褂孟铝忻顏聿榭磩偛艅?chuàng)建的用戶數(shù)據(jù)。
./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
ZK客戶端命令行查看：
./zkCli.sh -server localhost:2181
ls /config/users
3.?配置kafka jaas文件
配置了用戶之后，我們需要為 Broker 創(chuàng)建一個對應(yīng)的 JAAS 文件。在實際場景中，需要為每臺單獨的物理 Broker 機器都創(chuàng)建一份 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服務(wù)端使用的賬號和密碼，后面的user_xxx事預(yù)設(shè)的普通帳號認證信息。
?
中間部分配置的是PLAIN認證方式的賬戶和密碼，其中producer1是賬戶名，producer123是密碼。
Client配置了broker到Zookeeper的連接用戶名密碼，這里要和前面zookeeper配置中的zk_jaas.conf.conf 中 user_kafka 的賬號和密碼相同。
?
關(guān)于這個文件內(nèi)容，需要注意以下兩點：
?
1）不要忘記最后一行和倒數(shù)第二行結(jié)尾處的分號；
?
2）JAAS 文件中不需要任何空格鍵。
4.?kafka 配置文件啟用SASL認證
Kafka 服務(wù)配置文件 server.propertis，配置認證協(xié)議及認證實現(xiàn)類
cat server.properties其它內(nèi)容都注釋掉，然后追加如下內(nèi)容：
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.1995.?kafka 啟動腳本添加認證文件路徑的環(huán)境變量
Kafka 安全認證可以直接通過環(huán)境變量 -Djava.security.auth.login.config 設(shè)置，修改 Kafka 啟動腳本 kafka-start-server.sh 文件最后一行，增加一個參數(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-5122)?生產(chǎn)者配置
使用kafka-console-producer.sh腳本測試生產(chǎn)者，由于開啟安全認證和授權(quán)，此時使用console-producer腳本來嘗試發(fā)送消息，那么消息會發(fā)送失敗，原因是沒有指定合法的認證用戶，因此客戶端需要做相應(yīng)的配置，需要創(chuàng)建一個名為producer.conf的配置文件給producer程序使用。
config目錄下創(chuàng)建一個producer.conf的文件，cat producer.conf文件內(nèi)容如下：
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設(shè)置寫權(quán)限
3)?消費者配置
使用kafka-console-consumer.sh腳本測試生產(chǎn)者，由于開啟安全認證和授權(quán)，因此客戶端需要做相應(yīng)的配置。需要為 consumer 用戶創(chuàng)建consumer.conf給消費者程序，同時設(shè)置對topic的讀權(quán)限。
config目錄下創(chuàng)建一個consumer.conf的文件，cat consumer.conf文件內(nèi)容如下：
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設(shè)置讀權(quán)限。
4)?在生產(chǎn)者和消費者啟動腳本中引入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內(nèi)容如下：
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沒有啟動成功，我找下原因。把整個過程重新整理了一遍，發(fā)現(xiàn)zookeeper、kafka啟動成功了。
檢查驗證
./zkCli.sh -server localhost:2181
ls /brokers/ids
1. 生產(chǎn)者測試
1）?創(chuàng)建一個測試主題test_topic
./kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 1 --partitions 1 --topic test_topic
2）查看創(chuàng)建的Topic文章來源：http://www.zghlxwxcb.cn/news/detail-839621.html
./kafka-topics.sh --list --zookeeper localhost:2181 test_topic
文章來源地址http://www.zghlxwxcb.cn/news/detail-839621.html
到了這里，關(guān)于Apache zookeeper kafka 開啟SASL安全認證的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
