kafka如何配置SSL

这是一个如何配置kafka支持SSL环境的例子。

我们的目标：

使用kafka2.1.1
利用kafka自带的zookeeper
使用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

运行完之后，会生成server.keystore.jks/server.truststore.jks，client.keystore.jks/client.truststore.jks一共四个文件。

有一个地方要注意：
SAN的值给定义上了kafka.example.com，这儿实际上keytool是有bug的，实际应用中，我们可能只能定义域名，无法定义确切的主机名，所以最好的方式是在这里定义DNS的值为".example.com"，但是由于keytool存在bug(设计问题)，他不支持"'开头，那么我们只能使用全名主机名了。
要解决这个问题，可以使用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

这个文件没啥补充的。

增加/修改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

这个文件有几点说明：

ssl.client.auth是否验证client端的证书，即双向验证。
listeners支持kafka同时支持SSL和PLAINTXT等多种模式
listeners=PLAINTEXT://:9092,SSL://:9093
这样kafka就能支持在端口9092上监听非SSL的请求，同时在9093端口监听SSL端口。
security.inter.broker.protocol指定在kafka之间通信是是否使用SSL。
在前面的listeners我们知道kafkakey监听多个端口，这样运行对client进来的请求，我们让他监听在9093的SSL端口，然后kafka之间的通信我们让他监听的9092端口上，这样保证了外网访问的安全(SSL)，也提高了内部通信的性能(PLAINTEXT)。
这个内部通信协议指得是kafka之间的通信，不包括kafka和zookeeper之间的通信，也就是说kafka和zookeeper之间的通信还是PLAINTEXT的。
advertised.listeners广播监听地址
前面的listeners可以配置0.0.0.0地址，但是advertised.listeners不能使用这个地址，因为这个地址会被SSL连接的时候验证SAN值；当然也可以把listeners配置成SAN地址，这样advertised.listeners就不需要了，但是这种配置，无法区分内网访问，还是外部访问。

Step 3: 配置docker-compose.yaml

这里需要把kafka达到image里面进去，构造两个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证书信息。

总的来说，这个过程还是比较清楚的。

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,590评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 86,808评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,151评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,779评论 1赞 277
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,773评论 5赞 367
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,656评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,022评论 3赞 398
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,678评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 41,038评论 1赞 299
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,659评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,756评论 1赞 330
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,411评论 4赞 321
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,005评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,973评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,203评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,053评论 2赞 350
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,495评论 2赞 343

kafka如何配置SSL

推荐阅读更多精彩内容