自0.9.0.0.版本引入Security之后,Kafka一直在完善security的功能。當前Kafka security主要包含3大功能:認證(authentication)、信道加密(encryption)和授權(authorization)。信道加密就是為client到broker、broker到broker以及工具腳本與broker之間的數據傳輸配置SSL;認證機制主要是指配置SASL,而授權是通過ACL接口命令來完成的。
生產環境中,用戶若要使用SASL則必須配置Kerberos,但對於一些小公司而言,他們的用戶系統並不復雜(特別是專門為Kafka集群服務的用戶可能不是很多),顯然使用Kerberos有些大材小用,而且由於運行在內網環境,SSL加密也不是很必要。因此一個SASL+PLAINTEXT的集群環境足以應付一般的使用場景。本文給出一個可運行的實例來演示一下如何在不使用Kerberos的情況下配置SASL + ACL來構建secured Kafka集群。
在開始之前,我們簡單學習下Kafka ACL的格式。根據官網的介紹,Kafka中一條ACL的格式如下:“Principal P is [Allowed/Denied] Operation O From Host H On Resource R”。它們的含義描述如下:
- principal:表示一個Kafka user
- operation:表示一個具體的操作類型,如WRITE, READ, DESCRIBE等。完整的操作列表詳見:http://docs.confluent.io/current/kafka/authorization.html#overview
- Host:表示連向Kafka集群的client的IP地址,如果是‘*’則表示所有IP。注意:當前Kafka不支持主機名,只能指定IP地址
- Resource:表示一種Kafka資源類型。當前共有4種類型:TOPIC、CLUSTER、GROUP、TRANSACTIONID
下面我使用Kafka 0.11.0.0版本來演示下如何構建支持SASL + PLAINTEXT + ACL的Kafka集群環境。
1. Broker端配置
要配置SASL和ACL,我們需要在broker端進行兩個方面的設置。首先是創建包含所有認證用戶信息的JAAS文件。本例中,我們假設有3個用戶:admin, reader和writer,其中admin是管理員,reader用戶讀取Kafka集群中topic數據,而writer用戶則負責向Kafka集群寫入消息。我們假設這3個用戶的密碼分別與用戶名相同(在實際場景中,管理員需要單獨把密碼發給各自的用戶),因此我們可以這樣編寫JAAS文件:
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_reader="reader"
user_writer="writer";
};
保存該文件為kafka_cluster_jaas.conf(本例中的完整路徑是/Users/huxi/SourceCode/newenv/kafka_cluster_jaas.conf),之后我們需要把該文件的完整路徑作為一個JVM參數傳遞給Kafka的啟動腳本。不過由於bin/kafka-server-start.sh只接收server.properties的位置,不再接收其他任何參數,故我們需要修改該啟動腳本。具體做法如下:
$ pwd
/Users/huxi/SourceCode/newenv/kafka_0.11
$ cp bin/kafka-server-start.sh bin/secured-kafka-server-start.sh$ vi bin/secured-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=/Users/huxi/SourceCode/newenv/kafka_cluster_jaas.conf kafka.Kafka "$@"
做完上面的步驟后,我們就在bin/目錄下新建了一個secured-kafka-server-start.sh啟動腳本。
配置好JAAS文件后,我們開始修改broker啟動所需的server.properties文件,你至少需要配置(或修改)以下這些參數:
# 配置ACL入口類
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 本例使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol= SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
# 設置本例中admin為超級用戶
super.users=User:admin
Okay,現在我們可以啟動broker了(當前肯定要先啟動Zookeeper):
bin/secured-kafka-server-start.sh ../config_files/server.properties
.......
[2017-08-17 16:07:30,417] INFO [Kafka Server 0], started (kafka.server.KafkaServer)
可見,Kafka broker已經成功啟動了。不過當前該broker只會接收已認證client發來的請求。下面我們繼續clients端的配置。
2. Client端配置
我們先來創建一個測試topic,名為test,單分區,副本因子是1。
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic test --partitions 1 --replication-factor 1
Created topic "test".
咦?稍等下,我們不是啟用ACL了嗎?這里為什么會創建成功呢?這里我簡要說明一下: 首先我們啟用ACL了嗎?當然!由於我們沒有顯式地設置allow.everyone.if.no.acl.found,故這個參數默認是false,也就是說對於目前的這個Kafka集群而言,除了超級用戶之外的其他任何用戶都無法執行任何操作。其次,那為什么創建topic成功了呢?這是因為當前kafka-topics.sh腳本直接連接Zookeeper,故不受ACL的限制。所以無論是否配置了security,用戶總是可以使用kafka-topics來管理topic。
言歸正傳,本例中我們的目的是要測試:
- 用戶writer向test topic寫入消息
- 用戶reader從test topic讀取消息
下面我們先來啟動一個console-consumer和一個console-producer來看下當前是個什么狀況:
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>hello, kafka
[2017-08-17 16:16:09,626] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:09,685] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:09,740] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:09,799] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)......
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
[2017-08-17 16:16:47,936] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:47,992] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:48,052] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:16:48,163] WARN Bootstrap broker localhost:9092 disconnected (org.apache.kafka.clients.NetworkClient)
......
看到了吧,當前的console producer/consumer都無法工作,報的錯誤就是無法連接broker(localhost:9092)。這就是因為我們設置了security的緣故。下面我們為writer用戶配置安全,首先需要創建一個屬於writer用戶的JAAS文件,該文件中指定了writer用戶的credentials,如下所示:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="writer"
password="writer";
};
把上述內容保存到writer_jaas.conf文件(/Users/huxi/SourceCode/newenv/writer_jaas.conf)中。和broker類似,我們需要拷貝一份新的bin/kafka-console-producer.sh將該JAAS文件作為一個JVM參數傳給console producer:
$ cp bin/kafka-console-producer.sh bin/writer-kafka-console-producer.sh
$ vi bin/writer-kafka-console-producer.sh
把該文件中的這行:
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
修改為下面這行,然后保存退出
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/Users/huxi/SourceCode/newenv/writer_jaas.conf kafka.tools.ConsoleProducer "$@"
然后,我們創建一個producer.config為該console producer指定以下兩個屬性:
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
現在我們使用新的腳本來啟動console producer:
$ bin/writer-kafka-console-producer.sh --broker-list localhost:9092 --topic test --producer.config producer.config
>hello, kafka
[2017-08-17 16:28:13,930] WARN Error while fetching metadata with correlation id 1 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:28:14,036] WARN Error while fetching metadata with correlation id 3 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:28:14,143] WARN Error while fetching metadata with correlation id 4 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:28:14,246] WARN Error while fetching metadata with correlation id 5 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)
依然有錯誤,不過錯誤變成“無法獲取元數據”了。這說明我們運行的console producer通過了認證,但是沒有通過授權,因此我們需要配置ACL來讓writer用戶有權限寫入topic:
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:writer --operation Write --topic test
再試producer:
$ bin/writer-kafka-console-producer.sh --broker-list localhost:9092 --topic test --producer.config producer.config
>hello
>hello, kafka
>message abc......
producer生產消息成功了。等等!子曰:沒有消費之前永遠不要斷言生產成功了!下面我們就來配置下consumer,即用戶reader。
和writer用戶類似,我們首先創建reader用戶的JAAS文件,保存在/Users/huxi/SourceCode/newenv/kafka_0.11/reader_jaas.conf中:
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="reader"
password="reader";
};
然后拷貝一份新的console consumer來指定上面的reader_jaas.conf:
$ cp bin/kafka-console-consumer.sh bin/reader-kafka-console-consumer.sh
$ vi bin/reader-kafka-console-consumer.sh
把該文件中的這行:
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"
修改為下面這行,然后保存退出
exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config=/Users/huxi/SourceCode/newenv/reader_jaas.conf kafka.tools.ConsoleConsumer "$@"
然后,我們創建一個consumer.config為該console producer指定以下3個屬性:
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
group.id=test-group
現在運行console consumer:
$ bin/reader-kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config consumer.config
[2017-08-17 16:37:54,124] WARN Error while fetching metadata with correlation id 2 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)
[2017-08-17 16:37:54,127] ERROR Unknown error when running consumer: (kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: test-group
可以看到這次出現了兩個錯誤:第一個問題依然是無法獲取元數據——這表明reader用戶通過了認證但沒有通過授權;第二個問題表明reader用戶無權訪問consumer group——這同樣是授權的問題。我們需要設置ACL來解決它們。首先,我們為reader用戶設置test topic的讀權限:
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --operation Read --topic test
然后設置訪問group的權限:
$ bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --operation Read --group test-group
做完這些之后,我們重新運行console consumer程序:
$ bin/reader-kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config consumer.config
hello
hello, kafka
message abc
.......
可以看到,這次reader用戶成功地讀取了writer用戶生產的消息。這證明了writer和reader用戶都可以正常地工作了。
最后總結一下,這種方案適用於用戶數很少但又必須啟用安全的Kafka集群,不過此方案比較麻煩的地方在於需要為每個腳本都重新定制,加上-Djava.security.auth.login.config參數以識別JAAS文件,不如Kerberos來得簡單。另外用戶完全可以根據官網的教程配置SSL,然后很輕易地把本文中的例子改成SASL_SSL。
