1. 準備證書文件

mkdir /opt/kafka/pkicd !$# 生成CA證書
openssl req -x509 -nodes -days 3650 -newkey rsa:4096 -keyout ca.key -out ca.crt -subj "/CN=Kafka-CA"# 生成私鑰
openssl genrsa -out kafka.key 4096# 生成證書簽名請求 (CSR)
openssl req -new -key kafka.key -out kafka.csr -subj "/CN=kafka-cluster"# 創建包含所有節點的SAN 配置文件
cat > san.cnf << EOF
[ req ]
distinguished_name = req_distinguished_name
req_extensions = req_ext
prompt = no[ req_distinguished_name ]
CN = kafka-cluster[ req_ext ]
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth, clientAuth
subjectAltName = @alt_names[ alt_names ]
# 節點主機名與ip
DNS.1 = kafka-1
DNS.2 = kafka-2
DNS.3 = kafka-3
IP.1 = 192.168.100.131
IP.2 = 192.168.100.132
IP.3 = 192.168.100.133
EOF# 簽署證書
openssl x509 -req -in kafka.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out kafka.crt \
-days 3650 -extfile san.cnf -extensions req_ext# 檢查驗證證書
openssl x509 -in kafka.crt -text -noout | grep -A 1 "Subject Alternative Name"ls -ltotal 28
-rw-r--r-- 1 root root 1805 Jan 15 15:54 ca.crt
-rw------- 1 root root 3272 Jan 15 15:54 ca.key
-rw-r--r-- 1 root root   41 Jan 15 15:54 ca.srl
-rw-r--r-- 1 root root 1777 Jan 15 15:54 kafka.crt
-rw-r--r-- 1 root root 1590 Jan 15 15:49 kafka.csr
-rw------- 1 root root 3247 Jan 15 15:49 kafka.key
-rw-r--r-- 1 root root  259 Jan 15 15:51 san.cnf

2.?創建 Keystore

# 將證書和私鑰轉換為PKCS12文件openssl pkcs12 -export -in kafka.crt -inkey kafka.key -out kafka.p12 -name kafka-cert -CAfile ca.crt -caname root -passout pass:123.com# 使用 keytool 將 kafka.p12 文件導入到 Keystorekeytool -importkeystore \-deststorepass 123.com \-destkeypass 123.com\-destkeystore kafka.keystore.jks \-srckeystore kafka.p12 \-srcstoretype PKCS12 \-srcstorepass 123.com \-alias kafka-cert

3.?創建 Truststore

# 使用keytool創建Truststore并導入CA證書keytool -import \-file ca.crt \-keystore kafka.truststore.jks \-storepass 123.com \-alias root

4. 分發文件

將kafka.truststore.jks 和kafka.keystore.jks 文件分發到其他 kafka 節點

scp kafka.truststore.jks 192.168.100.132:/opt/kafka/pki/scp kafka.keystore.jks  192.168.100.132:/opt/kafka/pki/scp kafka.truststore.jks 192.168.100.133:/opt/kafka/pki/scp kafka.keystore.jks  192.168.100.133:/opt/kafka/pki/

?5.?Kafka服務端配置 TLS

# 在Kafka KRaft模式下的server.properties文件中，添加以下配置vim /opt/kafka/config/kraft/server.properties# 修改SSL配置
listeners=SSL://:9092,CONTROLLER://:9093
inter.broker.listener.name=SSL
advertised.listeners=SSL://192.168.100.131:9092,CONTROLLER://192.168.100.131:9093# 新增Keystore配置
ssl.keystore.location=/opt/kafka/pki/kafka.keystore.jks
ssl.keystore.password=123.com
ssl.key.password=123.com
# 新增Truststore配置
ssl.truststore.location=/opt/kafka/pki/kafka.truststore.jks
ssl.truststore.password=123.com
# 客戶端連接時啟用ssl
ssl.client.auth=required# 重啟
systemctl restart kafka

6.?客戶端配置 TLS

# 創建客戶端配置文件，指定證書信息admin.properties文件內容如下cat > /opt/kafka/config/admin.properties << EOF
security.protocol=SSL
ssl.keystore.location=/opt/kafka/pki/kafka.keystore.jks
ssl.keystore.password=123.com
ssl.truststore.location=/opt/kafka/pki/kafka.truststore.jks
ssl.truststore.password=123.com
ssl.endpoint.identification.algorithm=
ssl.key.password=123.com
EOF# 連接測試## 查看節點信息bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.100.131:9092 --command-config /opt/kafka/config/admin.properties ## 查看topic信息bin/kafka-topics.sh --describe --bootstrap-server 192.168.100.131:9092 --command-config /opt/kafka/config/admin.properties# 生產者生產消息bin/kafka-console-producer.sh --bootstrap-server 192.168.100.131:9092 --topic test --producer.config /opt/kafka/config/admin.properties
>hello boy# 消費者消費消息bin/kafka-console-consumer.sh --bootstrap-server 192.168.10.31:9092 --topic test --from-beginning --consumer.config /opt/kafka/config/admin.propertieshello boy

7.?kafka-ui 配置 TLS

# 修改kafka-ui配置文件cat > config.yml << EOF
kafka:clusters:-name: kafka-clusterbootstrapServers: 192.168.100.131:9092,192.168.100.132:9092,192.168.100.133:9092metrics:port: 9997type: JMXproperties:security:protocol: SSLssl:keystore:location: /opt/kafka/pki/kafka.keystore.jkspassword: 123.comssl_endpoint_identification_algorithm: ''ssl:truststorelocation: /opt/kafka/pki/kafka.truststore.jkstruststorepassword: 123.com
EOF# 重啟systemctl restart kafka-ui