kafka如何配置SSL

這是一個如何配置kafka支持SSL環(huán)境的例子。

我們的目標(biāo)：

  1. 使用kafka2.1.1
  2. 利用kafka自帶的zookeeper
  3. 使用docker來啟動kafka/zookeeper clustor

Step 1：生成jks文件

#!/bin/bash

# generate CA certificate
openssl req -new -x509 -keyout ca.key -out ca.pem -days 365 -passout pass:test1234 \
    -subj "/C=cn/O=mycomp/OU=mygroup/CN=ca"

# generate server keystore
keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA \
    -storepass test1234 -keypass test1234 -genkey \
    -dname "C=cn,O=mycomp,OU=mygroup,CN=server"
# generate client keystore
keytool -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA \
    -storepass test1234 -keypass test1234 -genkey \
    -dname "C=cn,O=mycomp,OU=mygroup,CN=client"

# import CA certificate into server truststore
keytool -keystore server.truststore.jks -alias CARoot -import -noprompt \
    -file ca.pem -storepass test1234
# import CA certificate into client truststore
keytool -keystore client.truststore.jks -alias CARoot -import -noprompt \
    -file ca.pem -storepass test1234

# export server certificate
keytool -keystore server.keystore.jks -alias localhost -certreq \
    -file server.csr -storepass test1234 -ext san="DNS:kafka.example.com"
# export client certificate
keytool -keystore client.keystore.jks -alias localhost -certreq \
    -file client.csr -storepass test1234 -ext san="DNS:kafka.example.com"

# sign server certificate with CA certificate
openssl x509 -req -CA ca.pem -CAkey ca.key -in server.csr \
    -out server.pem -days 365 -CAcreateserial -passin pass:test1234 \
    -extensions SAN \
    -extfile <(cat /etc/pki/tls/openssl.cnf <(printf "[SAN]\nsubjectAltName=DNS:kafka.example.com"))
# sign client certificate with CA certificate
openssl x509 -req -CA ca.pem -CAkey ca.key -in client.csr \
    -out client.pem -days 365 -CAcreateserial -passin pass:test1234 \
    -extensions SAN \
    -extfile <(cat /etc/pki/tls/openssl.cnf <(printf "[SAN]\nsubjectAltName=DNS:kafka.example.com"))

# import CA certificate into server keystore
keytool -keystore server.keystore.jks -alias CARoot -import -noprompt \
    -file ca.pem -storepass test1234
# import CA certificate into client keystore
keytool -keystore client.keystore.jks -alias CARoot -import -noprompt \
    -file ca.pem -storepass test1234

# import server certificate into server keystore
keytool -keystore server.keystore.jks -alias localhost -import \
    -file server.pem -storepass test1234
# import client certificate into client keystore
keytool -keystore client.keystore.jks -alias localhost -import \
    -file client.pem -storepass test1234

運(yùn)行完之后，會生成server.keystore.jks/server.truststore.jks，client.keystore.jks/client.truststore.jks一共四個文件。

有一個地方要注意：
SAN的值給定義上了kafka.example.com，這兒實(shí)際上keytool是有bug的，實(shí)際應(yīng)用中，我們可能只能定義域名，無法定義確切的主機(jī)名，所以最好的方式是在這里定義DNS的值為".example.com"，但是由于keytool存在bug(設(shè)計(jì)問題)，他不支持"'開頭，那么我們只能使用全名主機(jī)名了。
要解決這個問題，可以使用openssl工具來生成"*.example.com"，不要直接使用keytool。

Step 2：配置zookeeper和kafka的配置文件

增加/修改zookeeper配置文件/opt/kafka/config/zookeeper.properties：

dataDir=/data
clientPort=2181
maxClientCnxns=0
initLimit=5
dataLogDir=/datalog
syncLimit=2
autopurge.purgeInterval=1
autopurge.snapRetainCount=3
server.1=zookeeper.example.com:2888:3888
tickTime=2000

這個文件沒啥補(bǔ)充的。

增加/修改kafka配置文件/opt/kafka/config/server.properties文件：

broker.id=1
zookeeper.connect=zookeeper.example.com:2181
listeners=SSL://:9093
advertised.listeners=SSL://kafka.example.com:9093
listener.security.protocol.map=SSL:SSL
# security.inter.broker.protocol=ssl
ssl.client.auth=required
ssl.keystore.location=/work/ssl/server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.truststore.location=/work/ssl/server.truststore.jks
ssl.truststore.type=JKS
ssl.truststore.password=test1234
ssl.keystore.type=JKS

這個文件有幾點(diǎn)說明：

  1. ssl.client.auth是否驗(yàn)證client端的證書，即雙向驗(yàn)證。
  2. listeners支持kafka同時支持SSL和PLAINTXT等多種模式
    listeners=PLAINTEXT://:9092,SSL://:9093
    這樣kafka就能支持在端口9092上監(jiān)聽非SSL的請求，同時在9093端口監(jiān)聽SSL端口。
  3. security.inter.broker.protocol指定在kafka之間通信是是否使用SSL。
    在前面的listeners我們知道kafkakey監(jiān)聽多個端口，這樣運(yùn)行對client進(jìn)來的請求，我們讓他監(jiān)聽在9093的SSL端口，然后kafka之間的通信我們讓他監(jiān)聽的9092端口上，這樣保證了外網(wǎng)訪問的安全(SSL)，也提高了內(nèi)部通信的性能(PLAINTEXT)。
    這個內(nèi)部通信協(xié)議指得是kafka之間的通信，不包括kafka和zookeeper之間的通信，也就是說kafka和zookeeper之間的通信還是PLAINTEXT的。
  4. advertised.listeners廣播監(jiān)聽地址
    前面的listeners可以配置0.0.0.0地址，但是advertised.listeners不能使用這個地址，因?yàn)檫@個地址會被SSL連接的時候驗(yàn)證SAN值；當(dāng)然也可以把listeners配置成SAN地址，這樣advertised.listeners就不需要了，但是這種配置，無法區(qū)分內(nèi)網(wǎng)訪問，還是外部訪問。

Step 3: 配置docker-compose.yaml

這里需要把kafka達(dá)到image里面進(jìn)去，構(gòu)造兩個images：kafka和zookeeper都使用kafka的安裝包就行。

version: '2'
networks:
  byfn:

services:
  zookeeper.example.com:
    image: zookeeper
    container_name: zookeeper.example.com
    environment:
        - ZOO_MY_ID=1
        - ZOO_SERVERS=server.1=zookeeper.example.com:2888:3888
    volumes:
        - ./zookeeper.properties:/opt/kafka/config/zookeeper.properties
    networks:
      - byfn

  kafka.example.com:
    image: kafka
    container_name: kafka.example.com
    environment:
    volumes:
        - ./ssl:/ssl
        - ./server.properties:/opt/kafka/config/server.properties
    depends_on:
        - zookeeper.example.com
    networks:
        - byfn

Step 4: 啟動clustor

$ docker-compose up

Step 5: 客戶端訪問

配置client-ssl.properties文件

security.protocol=SSL 
ssl.truststore.location=/ssl/client.truststore.jks 
ssl.truststore.password=test1234
ssl.keystore.location=/ssl/client.keystore.jks 
ssl.keystore.password=test1234
ssl.key.password=test1234

啟動producer和consumer：

# producer
/opt/kafka/bin/kafka-console-producer.sh \
  --broker-list kafka.example.com:9093 \
  --topic test-topic \
  --producer.config /ssl/client-ssl.properties

# consumer
/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka1.example.com:9093 \
  --topic test-topic \
  --consumer.config /ssl/client-ssl.properties \
  --from-beginning

Step 6: 如果是雙端口配置

即9092是PLAINTEXT訪問， 9093是SSL訪問

# server.properties
listeners=PLAINTEXT://:9092,SSL://:9093
advertised.listeners=PLAINTEXT://kafka.example.com:9092,SSL://kafka.example.com:9093
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL

那么client端既可以訪問9092，也可以訪問9093，以producer為例：

# PLAINTEXT
/opt/kafka/bin/kafka-console-producer.sh \
  --broker-list kafka.example.com:9092 \
  --topic test-topic \

# SSL
/opt/kafka/bin/kafka-console-producer.sh \
  --broker-list kafka.example.com:9093 \
  --topic test-topic \
  --producer.config /ssl/client-ssl.properties

差別PLAINTEXT訪問9092端口，SSL訪問9093端口，并且配置上SSL證書信息。

總的來說，這個過程還是比較清楚的。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容