一、說明
在大數據處理和分析中 Apache Kafka 已經成爲了一個核心元件。然而在生產環境中部署 Kafka 時，安全性是一個必須要考慮的重要因素。SASL（簡單認證與安全層）和 SCRAM（基於密碼的認證機制的鹽化挑戰響應認證機制）提供了一種方法來增強 Kafka 叢集的安全性。
本文將從零開始部署 ZooKeeper 和 Kafka 並透過配置 SASL/SCRAM 和 ACL（訪問控制列表）來增強 Kafka 的安全性。
二、Kafka 的安全機制
kafka 社羣在 0.9.0.0 版本正式新增了安全特性，可以滿足各種安全性的要求，包括：
Kafka 與 ZooKeeper 之間的安全通訊；
Kafka 叢集 ZooKeeper 之間的安全通訊；
客戶端與服務端之間的安全通訊；
訊息級別的許可權控制，可以控制客戶端（生產者或消費者）的讀寫操作許可權。
| 認證方式 | 引入版本 | 適用場景 |
|---|---|---|
| SSL | 0.9.0 | SSL做通道加密比較多，SSL認證不如SASL所以一般都會使用SSL來做通訊加密。 |
| SASL/GSSAPI | 0.9.9 | 主要是給 Kerberos 使用的。如果你的公司已經做了 Kerberos 認證（比如使用 Active Directory），那麼使用 GSSAPI 是最方便的了。因為你不需要額外地搭建 Kerberos，只要讓你們的 Kerberos 管理員給每個 Broker 和要訪問 Kafka 叢集的作業系統使用者申請 principal 就好了。 |
| SASL/PLAIN | 0.10.2 | 簡單的使用者名稱密碼認證，通常與SSL結合使用，對於小公司來說，沒必要搭建公司級別的Kerberos，使用它就比較合適。 |
| SASL/SCRAM | 0.10.2 | PLAIN的加強版本，支援動態的使用者增減。 |
| Deleation Token | 1.1 | Delegation Token 是在 1.1 版本引入的，它是一種輕量級的認證機制，主要目的是補充現有的 SASL 或 SSL 認證。如果要使用 Delegation Token，你需要先配置好 SASL 認證，然後再利用 Kafka 提供的 API 去獲取對應的 Delegation Token。這樣 Broker 和客戶端在做認證的時候，可以直接使用這個 token，不用每次都去 KDC 獲取對應的 ticket（Kerberos 認證）或傳輸 Keystore 檔案（SSL 認證）。 |
| SASL/OAUTHBEARER | 2.0 | OAuth 2框架的整合。 |
三、環境和軟體準備
從 Apache Kafka 官網 下載對應版本的 Kafka 並解壓到你選擇的目錄。
確保已經安裝 Java 才能執行 Kafka，可以透過執行
java -version來檢查 Java 環境。
四、部署 Zookeeper
使用 Kafka 內建的 Zookeeper
4.1. 啟用 SASL 認證
進入 config 目錄，修改 zookeeper.properties 配置檔案增加以下內容：
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider jaasLoginRenew=3600000 requireClientAuthScheme=sasl zookeeper.sasl.client=true
4.2. 配置 JAAS
在 config 目錄下建立 zk_jaas.conf 檔案，內容如下：
Server {
org.apache.zookeeper.server.auth.DigestLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_zkclient="zkclient";
};
其作用是建立了一個 Server 節點，其中
org.apache.zookeeper.server.auth.DigestLoginModule required是認證邏輯的處理類；username、password是zookeeper之間通訊的使用者名稱和密碼；user_admin="admin"的結構是 user_[username]=[password] 定義 kafka-broker（zookeeper客戶端）連線到 zookeeper 時用的使用者名稱和密碼。
注意：Server 內部最後一行的
;和 } 後的;不能缺少！
4.3. 修改啟動檔案
進入 bin 目錄，修改 zookeeper-server-start.sh 檔案；
在 export KAFKA_HEAP_OPTS= 配置項的引數後新增 JAAS 的配置：
export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M -Djava.security.auth.login.config=../config/zk_jaas.conf"
4.4. 啟動 Zookeeper
執行命令：./zookeeper-server-start.sh -daemon ../config/zookeeper.properties
-daemon 引數配置後臺執行
4.5. 測試
可以從官網 Apache ZooKeeper 下載對應版本的 ZooKeeper 並解壓；
新增 JAAS 配置，在 confi 目錄下建立 zk_client_jaas.conf 檔案：
Client{
org.apache.zookeeper.server.auth.DigestLoginModule required
username="zkclient"
password="zkclient";
};
修改 bin 目錄下的 zkCli.sh 檔案，在啟動命令中增加 JAAS 的配置：
"$JAVA" "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" "-Dzookeeper.log.file=${ZOO_LOG_FILE}" \
-cp "$CLASSPATH" $CLIENT_JVMFLAGS $JVMFLAGS \
"-Djava.security.auth.login.config=../conf/zk_client_jaas.conf" \
org.apache.zookeeper.ZooKeeperMain "$@"
執行 zkCli.sh 連線本機已經啟動好的 ZooKeeper
進入 Kafka 的 log 目錄，檢視內建 zk 的日誌 zookeeper.out 顯示以下內容：
INFO adding SASL authorization for authorizationID: zkclient (org.apache.zookeeper.server.ZooKeeperServer)
代表 ZooKeeper 的 SASL 認證已經配置成功。
五、部署 Kafka
5.1. 配置 Kafka Broker
進入 config 目錄，修改 server.properties 配置檔案增加以下內容：
listeners=SASL_PLAINTEXT://:9092 advertised.listeners=SASL_PLAINTEXT://localhost:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 sasl.enabled.mechanisms=SCRAM-SHA-256 authorizer.class.name=kafka.security.authorizer.AclAuthorizer allow.everyone.if.no.acl.found=false super.users=User:admin
authorizer.class.name開啟 ACL 授權機制並指定實現類；allow.everyone.if.no.acl.found如果沒有找到ACL（訪問控制列表）配置，是否允許任何操作；這裏設定為false指除了超級管理員，其他使用者必須配置 ACL 才能訪問資源；super.users超級管理員，無需配置 ACL 擁有所有許可權的使用者。
5.2. 配置 JAAS
在 config 目錄下建立 kafka_server_jaas.conf 檔案，內容如下：
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin";
};
Client{
org.apache.kafka.common.security.plain.PlainLoginModule required
username="zkclient"
password="zkclient";
};
KafkaServer中的username，password用於 Kafka 叢集 Broker 節點之間通訊用的賬號密碼；KafkaServer中的user_test="test"用於 Kafka 客戶端（producer，consumer）連線broker時，用該配置下user_[username]=[password]結構配置的賬號密碼登入；Client用於 broker 和 zookeeper 之間的認證，對應 zk_jaas.conf 中的 【user_zkclient="zkclient"】 配置；user_admin="admin"的結構是 user_[username]=[password] 定義 kafka-broker（zookeeper客戶端）連線到 zookeeper 時用的使用者名稱和密碼。
5.3. 修改啟動檔案
進入 bin 目錄，修改 kafka-server-start.sh 檔案；
在 export KAFKA_HEAP_OPTS= 配置項的引數後新增 JAAS 的配置：
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=../config/kafka_server_jaas.conf"
5.4. 建立 SCRAM 使用者
在啟動 Kafka 之前需要先建立好使用者，在 bin 目錄下執行以下內容：
分別建立
admin(超級管理員) 和test(客戶端使用者)
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin]' --entity-type users --entity-name admin ./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=test]' --entity-type users --entity-name test
SASL/SCRAM認證的使用者資訊是動態建立儲存在 ZooKeeper 中， 由於上面的配置kafka_server_jaas.conf中 Broker 之間的通訊是透過admin使用者的，如果該使用者不存在會 啟動報錯。
5.5. 啟動 Kafka
執行命令：./kafka-server-start.sh -daemon ../config/server.properties
-daemon 引數配置後臺執行
六、驗證 SASL/SCRAM 鑑權
6.1. 客戶端認證配置
6.1.1. 管理員配置
進入 config 目錄建立 cmd.properties 內容如下：
security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";
配置認證的型別以及登入邏輯的處理類和使用者，使用超級管理員 admin
注意 最後的
;是必須加上的。
6.1.2. 生產者配置
修改 config 目錄下的 producer.properties 增加以下內容：
security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";
生產者也使用超級管理員 admin 來發送訊息。
6.1.3. 消費者配置
修改 config 目錄下的 consumer.properties 增加以下內容：
security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="test" password="test";
消費者使用 test 使用者來接收訊息。
6.2. 建立topic
在 bin 目錄下執行以下命令：
./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test --partitions 1 --replication-factor 1 --command-config ../config/cmd.properties
bootstrap-server配置 Kafka 服務端的地址topic指定topic名稱command-config指定命令的認證配置，這裏使用上面建立的 管理員配置
建立成功後可以透過以下命令檢視存在的 topic 清單：
./kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ../config/cmd.properties
6.3. 建立消費者
6.3.1. 執行 kafka-console-consumer
在 bin 目錄下執行以下命令：
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties
執行命令後會發現以下 報錯 資訊：
ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$) org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 Processed a total of 0 messages
Authentication failed認證失敗，由於消費者的認證使用的是 test 使用者，而該使用者還未配置任何 ACL 許可權。
6.3.2. 配置使用者 ACL 許可權
Kafka 的 ACL (Access Control Lists) 允許你定義哪些使用者可以訪問哪些主題，並且可以執行哪些操作（如讀、寫、建立、刪除等）。
執行以下命令：
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:test --operation Read --topic test --group test-consumer-group
為 test 使用者在資源
topic[test]下分配只讀許可權
執行成功，可以透過以下命令檢視資源所分配的所有 ACL 清單：
./kafka-acls.sh --bootstrap-server localhost:9092 --topic test --list --command-config ../config/cmd.properties
重新建立消費者：
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties
執行成功後該 shell 視窗會一直阻塞等待訊息。
6.4. 建立生產者
新開一個 shell 視窗 在 bin 目錄下執行以下命令：
./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config ../config/producer.properties
由於生產者的認證使用的是 admin 為 超級管理員 所以無需配置 ACL 許可權。
執行成功後會出現
>符號，輸入內容之後，切換到 消費者 視窗就可以看到了。
