單機kafka添加SSL認證

參考：https://www.cnblogs.com/smartloli/p/12950761.html

https://www.icode9.com/content-4-136457.html

1.概述

最近有同學咨詢說，Kafka的SSL安全認證如何安裝與使用？今天筆者將通過以下幾個方面來介紹Kafka的SSL：

• Kafka 權限介紹
• Kafka SSL的安裝與使用
• Kafka Eagle中如何配置SSL？

2.內容

2.1 什么是Kafka權限認證？

在Kafka 0.9.0.0之后，Kafka社區增加了一系列的功能，其中包含對Kafka集群進行安全管控。支持的權限認證方式如下：

1. Broker與Client之間的權限認證（例如Producer和Consumer）。可以使用SSL或SASL，而SASL支持如下方案：

• • SASL/GSSAPI（Kerberos），開始於0.9.0.0版本
  • SASL/PLAIN，開始於0.10.0.0版本
  • SASL/SCRAM-SHA-256和SASL/SCRAM-SHA-512，開始於0.10.2.0版本
  • SASL/OAUTHBEARER，開始於2.0版本

　　 2. Broker和Zookeeper之間建立權限認證

　　 3. 在Broker和Client之間、Broker和Broker之間使用SSL建立權限認證時，性能會有所下降，其程度取決於CPU類型和JVM的實現

　　 4. 對Client進行讀寫認證

在實際生產環境中，對於權限認證使用的較多的是SCRAM認證，其原因在《Kafka SCRAM和PLAIN實戰》這篇博客中詳細解釋。

2.2 Kafka SSL安裝與使用

Kafka允許客戶端使用SSL來連接，默認情況下，SSL是禁止的，但是可以通過手動開啟。安裝Kafka SSL的流程如下所示：

 執行步驟如下所示：

1、創建腳本create_ssl.sh

[root@database-zongshuai kafka_2.12-2.2.2]# vim  create_ssl.sh
#! /bin/bash
set -e
#初始化環境變量（在腳本中聲明變量）
echo "Step1: Config env"

BASE_DIR=/data/kafka_2.12-2.2.2/ssl
CERT_OUTPUT_PATH="$BASE_DIR/certificates"
PASSWORD=ke123456
KEY_STORE="$CERT_OUTPUT_PATH/kafka.keystore"
TRUST_STORE="$CERT_OUTPUT_PATH/kafka.truststore"
KEY_PASSWORD=$PASSWORD
STORE_PASSWORD=$PASSWORD
TRUST_KEY_PASSWORD=$PASSWORD
TRUST_STORE_PASSWORD=$PASSWORD
CLUSTER_NAME=ke-cluster-01
CERT_AUTH_FILE="$CERT_OUTPUT_PATH/ca-cert"
CLUSTER_CERT_FILE="$CERT_OUTPUT_PATH/${CLUSTER_NAME}-cert"
DAYS_VALID=365
D_NAME="CN=database-zongshuai.novalocal, OU=bonc, O=bonc, L=China, ST=China, C=database-zongshuai.novalocal"

mkdir -p $CERT_OUTPUT_PATH

#創建證書到KeyStore
echo "Step2: Create certificate to keystore"
keytool -keystore $KEY_STORE -alias $CLUSTER_NAME -validity $DAYS_VALID -genkey -keyalg RSA -storepass $STORE_PASSWORD -keypass $KEY_PASSWORD -dname "$D_NAME"

#創建CA
echo "Step3: Create CA"
openssl req -new -x509 -keyout $CERT_OUTPUT_PATH/ca-key -out "$CERT_AUTH_FILE" -days "$DAYS_VALID" -passin pass:"$PASSWORD" -passout pass:"$PASSWORD" -subj "/C=CN/S
T=XX/L=XX/O=XX/CN=XX"

#導入CA到TrustStore中
echo "Step4: Import CA into truststore"
keytool -keystore "$TRUST_STORE" -alias CARoot -import -file "$CERT_AUTH_FILE" -storepass "$TRUST_STORE_PASSWORD" -keypass "$TRUST_KEY_PASS" -noprompt

#導出證書
echo "Step5: Export certificate from keystore"
keytool -keystore "$KEY_STORE" -alias "$CLUSTER_NAME" -certreq -file "$CLUSTER_CERT_FILE" -storepass "$STORE_PASSWORD" -keypass "$KEY_PASSWORD" -noprompt

#給證書簽名
echo "Step6: Signing the certificate"
openssl x509 -req -CA "$CERT_AUTH_FILE" -CAkey $CERT_OUTPUT_PATH/ca-key -in "$CLUSTER_CERT_FILE" -out "${CLUSTER_CERT_FILE}-signed" -days "$DAYS_VALID" -CAcreateser
ial -passin pass:"$PASSWORD"

#導入CA到KeyStore
echo "Setp7: Import CA into keystore"
keytool -keystore "$KEY_STORE" -alias CARoot -import -file "$CERT_AUTH_FILE" -storepass "$STORE_PASSWORD" -keypass "$KEY_PASSWORD" -noprompt

#導入證書到KeyStore
echo "Setp8: Import signed certificate into keystore"
keytool -keystore "$KEY_STORE" -alias "${CLUSTER_NAME}" -import -file "${CLUSTER_CERT_FILE}-signed" -storepass "$STORE_PASSWORD" -keypass "$KEY_PASSWORD" -noprompt

　成功執行腳本后，會在對應的目錄（$BASE_DIR/certificates）生成對應文件清單：

 2、修改kafka安裝目錄下config目錄下的server.properties文件

listeners=SSL://database-zongshuai.novalocal:9095
advertised.listeners=SSL://database-zongshuai.novalocal:9095
ssl.keystore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.keystore
ssl.keystore.password=ke123456
ssl.key.password=ke123456
ssl.truststore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.truststore
ssl.truststore.password=ke123456
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS 
ssl.truststore.type=JKS 
ssl.endpoint.identification.algorithm=HTTPS
security.inter.broker.protocol=SSL
broker.id=0
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=/data/kafka_2.12-2.2.2/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=database-zongshuai.novalocal:2182
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

注：SSL配置最好寫在配置文件的最上面，否者可能導致Kafka配置SSL失敗。

3、修改zookeeper配置文件

[root@database-zongshuai kafka_2.12-2.2.2]# grep  '^[a-z]'  config/zookeeper.properties            
dataDir=/data/kafka_2.12-2.2.2/zookeeper
clientPort=2182
maxClientCnxns=0

4、創建目錄

[root@database-zongshuai kafka_2.12-2.2.2]# pwd
/data/kafka_2.12-2.2.2
[root@database-zongshuai kafka_2.12-2.2.2]#mkdir  zookeeper  kafka-logs

5、啟動zookeeper

[root@database-zongshuai kafka_2.12-2.2.2]# ./bin/zookeeper-server-start.sh  -daemon config/zookeeper.properties

6、啟動kafka

[root@database-zongshuai kafka_2.12-2.2.2]# ./bin/kafka-server-start.sh   -daemon  config/server.properties

7、使用Linux自帶的openssl測試一下，驗證我們配置的ssl有效

[root@database-zongshuai kafka_2.12-2.2.2]# openssl s_client -debug -connect  database-zongshuai.novalocal:9095 -tls1

 

 

 2.3驗證

1、創建topic

#創建topic
[root@database-zongshuai kafka_2.12-2.2.2]#bin/kafka-topics.sh --create --zookeeper database-zongshuai.novalocal:2182 --replication-factor 1 --partitions 1 --topic test01

#查看topic
[root@database-zongshuai kafka_2.12-2.2.2]#bin/kafka-topics.sh --list --zookeeper database-zongshuai.novalocal:2182

 2、模擬生產者

創建一個SSL下的消費者配置文件p.properties

[root@database-zongshuai kafka_2.12-2.2.2]# cat  p.properties 
bootstrap.servers=database-zongshuai.novalocal:9095
security.protocol=SSL
ssl.truststore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.truststore
ssl.truststore.password=ke123456   
ssl.keystore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.keystore
ssl.keystore.password=ke123456
ssl.key.password=ke123456

啟動生產者：

[root@database-zongshuai kafka_2.12-2.2.2]#bin/kafka-console-producer.sh --broker-list database-zongshuai.novalocal:9095 --topic test01 --producer.config  /data/kafka_2.12-2.2.2/p.properties

 3、模擬消費者

創建一個SSL下的消費者配置文件c.properties

[root@database-zongshuai kafka_2.12-2.2.2]# cat  c.properties 
security.protocol=SSL
group.id=test-group
ssl.truststore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.truststore
ssl.truststore.password=ke123456
ssl.keystore.password=ke123456
ssl.keystore.location=/data/kafka_2.12-2.2.2/ssl/certificates/kafka.keystore

啟動消費者：

[root@database-zongshuai kafka_2.12-2.2.2]# bin/kafka-console-consumer.sh --bootstrap-server database-zongshuai.novalocal:9095 --topic test01 --from-beginning --consumer.config  c.properties