kafka ssl & acl
一 KafkaSSL配置
1.1 建立金鑰和證書
以下在每臺kafka伺服器上執行
keytool -keystoreserver.keystore.jks -alias localhost -validity 365-keyalg RSA -genkey
以下在連線kafka的clientt端執行，如果不要求客戶端身份認證（伺服器端沒有配置ssl.client.auth=required），那麼這步可以省略
keytool -keystoreclient.keystore.jks -alias localhost -validity 365-keyalg RSA –genkey
【注】
1、需要將金鑰庫和金鑰密碼設定相同
2、keystore 金鑰庫儲存位置、alias證書實體唯一別名、validity 金鑰有效期、keyalg加密演算法
3、一次金鑰方法
keytool-genkey -alias test -keypass test1234-storepass test1234 -validity 365-keystore test.keystore -dname"CN=hadoop001, OU=test, O=test, L=test,ST=test, C=test"
1.2 利用openssl建立CA
openssl req -new-x509 -keyout ca-key -out ca-cert -days 365
【注】
1、keyout 私鑰檔案（預設當前目錄）、out 證書檔案(預設當前目錄)，days 證書有效期（天）
2、CA相當於證書發放機構，提供證書籤名服務
3、kafka叢集其中一臺伺服器中執行即可，也可任選叢集外機器充當CA
1.3 建立CA客戶端證書信任庫
kafka server端執行：
keytool -keystoreserver.truststore.jks -alias CARoot -import -file ca-cert
kafka client端執行：
keytool -keystoreclient.truststore.jks -alias CARoot -import -file ca-cert
【注】truststore庫儲存的是CAclient端應該信任的CA證書集合，匯入證書進入自己的信任庫意味著信任所有該CA證書機構簽發的證書。
1.4 CA對證書進行簽名（利用CA證書ca-key,ca-cert簽名）
1、遷出證書
keytool -keystoreserver.keystore.jks -alias localhost -certreq -file cert-file
keytool -keystoreclient.keystore.jks -alias localhost -certreq -file cert-file-client
2、CA簽名
openssl x509 -req-CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days365-CAcreateserial -passin pass:test1234
openssl x509-req-CA ca-cert -CAkey ca-key -in cert-file-client -out cert-signed-client-days 365-CAcreateserial -passin pass:test1234
1.5 簽名證書匯入kafka server端信任庫
將經過CA簽名過的kafkaserver端證書（cert-signed）、client端證書（cert-signed-client）匯入server端信任庫（server.truststore.jks）
keytool-keystore server.truststore.jks-alias localhost -import -file cert-signed
keytool-keystore server.truststore.jks-alias localhost -import -filecert-signed-client
1.6 簽名證書匯入kafka client端信任庫
將CA簽名後的kafkaserver端證書（cert-signed）、client端證書（cert-signed-client）匯入kafkaclient端信任庫（client.truststore.jks）
keytool-keystore client.truststore.jks-alias localhost -import -file cert-signed
keytool-keystore client.truststore.jks-alias localhost -import -filecert-signed-client
1.7 配置server.properties檔案
新增如下關於SSL資訊的配置：
#開啟9093埠ssl協議請求
listeners=PLAINTEXT://192.168.14.140:9092,SSL://192.168.14.140:9093
#指定kafkaserver祕鑰、信任庫儲存位置，ssl.keystore.location存自己的私鑰，ssl.truststore.location存放信任庫
ssl.keystore.location=/opt/kafka_2.10-0.10.0.0/ssl/server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.truststore.location=/opt/kafka_2.10-0.10.0.0/ssl/server.truststore.jks
ssl.truststore.password=test1234
#kafkabroker之間也使用ssl通訊
security.inter.broker.protocol=SSL
#客戶端也需要認證（看需要）
ssl.client.auth=required
1.8 client端配置（producer、consumer關於SSL的配置相同）
如果kafka server端沒有配置ssl.client.auth=required，如下是SSL最小化配置：
#通訊協議
security.protocol=SSL
ssl.truststore.location=/opt/kafka_2.10-0.10.0.0/ssl/client.truststore.jks
ssl.truststore.password=test1234
kakfa server端開啟了客戶端認證，新增如下配置：
#指定kafkaclient祕鑰、信任庫儲存位置，ssl.keystore.location存自己的私鑰，#ssl.truststore.location存放信任庫
ssl.keystore.location=/opt/kafka_2.10-0.10.0.0/ssl/client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
1.9 測試驗證
kafka-console-producer.sh--broker-listlocalhost:9093 --topic test --producer.configclient-ssl.properties
kafka-console-consumer.sh--bootstrap-serverlocalhost:9093 --topic test --new-consumer--consumer.configclient-ssl.properties
二 啟用 ACL
2.1 配置server.properties檔案
啟用kafka 自帶ACL授權
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
principal.builder.class=org.apache.kafka.common.security.auth.DefaultPrincipalBuilder
新增超級使用者，建議kafka server所有broker內部通訊採用相同的principle，並將其設為超級使用者
super.users=User:CN=hadoop001,OU=test,O=test,L=test,ST=test,C=test
【注】如果開啟ACL，預設情況除了超級使用者，沒有使用者允許訪問
三 ACL使用
acl的格式定義 "Principal P is [Allowed/Denied]Operation O From Host H OnResource R”，預設情況下，SSL的使用者名稱稱的形式是"CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown"。
預設kafka叢集操作許可權可以分為Cluster、producer、consumer，其中Cluster為叢集管理許可權，kafka server必須擁有該許可權。
3.1 檢視ACL列表
./kafka-acls.sh--list--authorizer-properties zookeeper.connect=localhost:2181/kafka
3.2 新增producer許可權
1、所有topic擁有producer許可權
kafka-acls.sh--allow-principalUser:CN=hadoop001,OU=test,O=test,L=test,ST=test,C=test--authorizer-propertieszookeeper.connect= localhost:2188/apache_kafka --producer--add –topic=*
2、指定topic擁有producer許可權
kafka-acls.sh--allow-principalUser:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test--authorizer-propertieszookeeper.connect= localhost:2188/apache_kafka--producer --add –topic=test01
3.3 新增consumer許可權
1、所有consumer group對所有topic擁有consumer許可權
kafka-acls.sh--authorizer-propertieszookeeper.connect=localhost:2188/apache_kafka--allow-principal User:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test --consumer --topic=* --group=* --add
2、所有consumer group對某個topic擁有consumer許可權
kafka-acls.sh--authorizer-propertieszookeeper.connect=localhost:2188/apache_kafka--allow-principal User:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test --consumer --topic=test01 --group=* --add
3.4 刪除許可權
kafka-acls.sh--allow-principalUser:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test--authorizer-propertieszookeeper.connect= localhost:2188/apache_kafka --producer--remove –topic=*
3.5 原子許可權管理（read、write、describe）
1、read
kafka-acls.sh--authorizer-propertieszookeeper.connect=localhost:2188/apache_kafka--allow-principal User:CN=hadoop001,OU=test,O=test,L=test,ST=test,C=test --operation read --topic=* --group=* --add
2、write
kafka-acls.sh--authorizer-propertieszookeeper.connect=localhost:2188/apache_kafka--allow-principal User:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test --operation write --topic=* --add
3、describe
kafka-acls.sh--authorizer-propertieszookeeper.connect=localhost:2188/apache_kafka--allow-principal User:CN= hadoop001,OU=test,O=test,L=test,ST=test,C=test --operation describe --topic=* --add
四 注意點
1、 Kafka SSL可以單獨使用，不開啟ACL模式下只要客戶端、服務端principle驗證通過即可以進行資料通訊
2、 啟用SSL後，Kafka sever端每臺broker的信任庫必須擁有叢集所有broker的證書資訊，推薦配置為叢集內部公用一個principle
3、 自己建立CA時，叢集內部需要使用同一個CA進行簽名和認證
4、 啟用ACL後，kafka叢集內部所有角色啟動使用者的principle必須擁有叢集的讀、寫、叢集管理許可權，推薦使用超級使用者管理叢集
5、 Kafka SSL相關的所有執行操作只針對new producer、new consumer有效，老版本的kafka不支援