#作者 ：張桐瑞

1Kafka安全認證技術介紹

Kafka作為一款高吞吐量的分布式消息系統，在數據傳輸和存儲過程中的安全性至關重要。目前，Kafka支持多種認證方式，每種方式都有其獨特的特點和適用場景，在生產環境中常見應用的SASL相關認證方式如下：
SSL/TLS認證：基于SSL/TLS加密技術，通過SSL/TLS證書對客戶端和服務器進行雙向身份驗證。在數據傳輸過程中，利用證書加密通道，確保數據在傳輸層的安全性和完整性，有效防止數據被竊取或篡改。這種認證方式廣泛應用于對數據安全和隱私要求極高的場景，例如金融行業的數據傳輸。
SASL/PLAIN認證：簡單身份驗證和授權層應用程序接口（Simple Authentication and Security Layer Application Programming Interface）的PLAIN認證，是一種極為簡單的用戶名和密碼認證方式。其優勢在于配置和維護簡單，易于在Kafka與其他應用程序之間建立認證機制，適用于對安全性要求相對較低、部署和維護希望簡便的場景，如一些小型企業的內部系統。
SASL/SCRAM認證：包括SCRAM-SHA-256、SCRAM-SHA-512等認證方式。該認證過程需要客戶端和服務器密切協同完成，涉及到多次交互驗證。雖然使用和維護相對復雜，但它具備動態增加用戶的顯著優勢，無需重啟Kafka組件服務端即可完成用戶的添加，為系統的用戶管理提供了極大的靈活性，適合對用戶管理靈活性要求較高的企業。
SASL/GSSAPI認證（Kerberos認證）：主要適用于大型企業的生產環境，通常與Kerberos協議結合使用。通過集成目錄服務（如Active Directory，AD），實現用戶身份的統一管理和認證。這種認證機制不僅提供了卓越的安全性，還能為用戶帶來良好的體驗，例如實現單點登錄（SSO）功能，方便用戶在多個相關系統間無縫切換。
Kafka自帶ACL訪問控制方式：Kafka提供的ACL（Access Control Lists，訪問控制列表）功能，可針對特定的Topic或Topic Partition精細控制用戶的訪問權限，如讀、寫、刪除等操作權限。通過合理配置ACL，可以有效保障業務數據的安全性，確保敏感數據僅被授權用戶訪問。
企業在選擇認證方式時，需要綜合考慮多方面因素。既要確保業務數據的安全性，又要權衡部署后的性能影響和部署復雜度。在實際生產部署過程中，應根據企業的安全政策、網絡架構、對系統性能的要求等因素，謹慎選擇最適合的認證方式，以確保所選的認證機制與企業的實際需求完美契合。

2基礎設置

Zookeeper版本：apache-zookeeper-3.6.3-bin。
Kafka版本：kafka_2.13-3.5.1。

3 配置SASL/SCRAM認證

3.1編寫server.properties配置

listeners=SASL_PLAINTEXT://kafkaip:9092
advertised.listeners=SASL_PLAINTEXT://kafkaip:9092
super.users=User:admin
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
security.inter.broker.protocol=SASL_PLAINTEXT
authorizer.class.name=kafka.security.authorizer.AclAuthorizer

配置說明：
listeners：指定Kafka服務器監聽的地址和協議，這里使用SASL_PLAINTEXT協議，監聽在kafkaip:9092地址上。
advertised.listeners：用于告知客戶端連接Kafka服務器的地址和協議，確保客戶端能夠正確連接。
super.users：設置超級用戶，這里定義了admin為超級用戶，超級用戶擁有更高的權限，可進行一些特殊操作。
sasl.enabled.mechanisms：啟用SCRAM-SHA-512認證機制，確保Kafka服務器支持該認證方式。
sasl.mechanism.inter.broker.protocol：指定Kafka集群內部節點間通信使用的認證機制為SCRAM-SHA-512。
security.inter.broker.protocol：明確集群內部節點間通信的安全協議為SASL_PLAINTEXT。
authorizer.class.name：配置Kafka的訪問控制授權器，這里使用kafka.security.authorizer.AclAuthorizer來實現基于ACL的訪問控制。

3.2編寫kafka.conf密碼文件

KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule requiredusername="admin"password="admin";
};  #kafka用戶名密碼；
Client {org.apache.zookeeper.server.auth.DigestLoginModule requiredusername="user1"password="user1";
};  #zookeeper連接；

配置說明：
KafkaServer部分：配置Kafka服務器認證所需的用戶名和密碼，這里設置用戶名為admin，密碼為admin，使用org.apache.kafka.common.security.scram.ScramLoginModule模塊進行認證。
Client部分：用于配置連接Zookeeper時所需的用戶名和密碼，設置用戶名為user1，密碼為user1，使用org.apache.zookeeper.server.auth.DigestLoginModule模塊進行認證。

3.3編寫user.properties配置文件

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

配置說明：
security.protocol：指定客戶端與Kafka服務器通信的安全協議為SASL_PLAINTEXT。
sasl.mechanism：明確使用SCRAM-SHA-512作為認證機制。
sasl.jaas.config：配置JAAS（Java Authentication and Authorization Service）認證所需的參數，包括使用的認證模塊、用戶名和密碼。

3.4編寫kafka-run-class.sh腳本文件

if [ -z "$KAFKA_OPTS" ]; thenKAFKA_OPTS="-Djava.security.auth.login.config=/kafka/config/kafka.conf"
Fi
…

配置說明：
在kafka-run-class.sh腳本中，添加上述代碼片段。這段代碼用于檢查KAFKA_OPTS環境變量是否為空，如果為空，則設置KAFKA_OPTS環境變量，指定Kafka服務器啟動時使用的JAAS配置文件路徑為/tmp/kafka/config/kafka.conf，確保Kafka服務器能夠正確加載認證配置。

3.5Zk中增加kafka用戶

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

命令說明：
上述命令用于在Zookeeper中為Kafka添加用戶。通過kafka-configs.sh腳本，連接到指定的Zookeeper服務器，為名為admin的用戶添加SCRAM-SHA-512認證方式的密碼配置。
啟動kafka后可使用kafka命令進行創建用戶

./bin/kafka-configs.sh --bootstrap-server kafkaip:9092 --command-config config/user.properties --alter --add-config 'SCRAM-SHA-512=[password=exporter]' --entity-type users --entity-name expoter

命令說明：
在Kafka啟動后，利用kafka-configs.sh腳本，通過指定的bootstrap-server，并依據config/user.properties配置文件中的認證信息，為名為exporter的用戶添加SCRAM-SHA-512認證方式的密碼配置。

3.6啟動kafka進程

./bin/kafka-server-start.sh ./config/server.properties
4Kafka-exporter配置監控采集
Kafka-exporter是一個用于采集Kafka監控指標的工具，為了實現對所有topic的監控數據采集，需要為其配置具有describe所有topic權限的用戶。
./kafka_exporter --kafka.server=kafkaip:9092 --sasl.enabled --sasl.mechanism=SCRAM-SHA-512 --sasl.username=exporter --sasl.password=exporter --topic.filter=topicname
–kafka.server：指定Kafka服務器的地址和端口
–sasl.enabled：啟用SASL認證
–sasl.mechanism：設置認證機制為SCRAM-SHA-512
–sasl.username和–sasl.password：指定用于認證的用戶名和密碼
–topic.filter：設置監控的topic過濾規則