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的流程如下所示：

 执行步骤如下所示：

步骤一：初始化环境变量（在脚本中声明变量）

BASE_DIR=/Users/smartloli/workspace/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=localhost, OU=YourDept, O=YourCompany, L=China, ST=China, C=localhost"

mkdir -p $CERT_OUTPUT_PATH

步骤二：创建证书到KeyStore

keytool -keystore $KEY_STORE -alias $CLUSTER_NAME -validity $DAYS_VALID -genkey -keyalg RSA -storepass $STORE_PASSWORD -keypass $KEY_PASSWORD -dname "$D_NAME"

步骤三：创建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/ST=XX/L=XX/O=XX/CN=XX"

步骤四：导入CA到TrustStore中

keytool -keystore "$TRUST_STORE" -alias CARoot -import -file "$CERT_AUTH_FILE" -storepass "$TRUST_STORE_PASSWORD" -keypass "$TRUST_KEY_PASS" -noprompt

步骤五：导出证书

keytool -keystore "$KEY_STORE" -alias "$CLUSTER_NAME" -certreq -file "$CLUSTER_CERT_FILE" -storepass "$STORE_PASSWORD" -keypass "$KEY_PASSWORD" -noprompt

步骤六：给证书签名

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" -CAcreateserial -passin pass:"$PASSWORD"

步骤七：导入CA到KeyStore

keytool -keystore "$KEY_STORE" -alias CARoot -import -file "$CERT_AUTH_FILE" -storepass "$STORE_PASSWORD"-keypass "$KEY_PASSWORD" -noprompt

步骤八：导入证书到KeyStore

keytool -keystore "$KEY_STORE" -alias "${CLUSTER_NAME}" -import -file "${CLUSTER_CERT_FILE}-signed"-storepass "$STORE_PASSWORD" -keypass "$KEY_PASSWORD" -noprompt

整个创建证书的脚本如下：

#! /bin/bash
echo "Step1: Config env"
BASE_DIR=/Users/smartloli/workspace/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=test-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=localhost, OU=YourDept, O=YourCompany, L=China, ST=China, C=localhost"

mkdir -p $CERT_OUTPUT_PATH

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"

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/ST=XX/L=XX/O=XX/CN=XX"

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" -CAcreateserial -passin pass:"$PASSWORD"

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


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）生成对应文件清单：

ca-cert
ca-cert.srl
ca-key
kafka.keystore
kafka.truststore
test-cluster-01-cert
test-cluster-01-cert-signed

2.3 Kafka Broker配置SSL

Kafka Broker是支持监听多个Port上的连接，通过在Kafka服务端的配置文件中（默认是server.properties）中进行配置，至少监听一个Port，用逗号进行分割。配置内容如下：

### SSL Start

listeners=SSL://localhost:9095
advertised.listeners=SSL://localhost:9095
ssl.keystore.location=/Users/smartloli/workspace/ssl/certificates/kafka.keystore
ssl.keystore.password=ke123456
ssl.key.password=ke123456
ssl.truststore.location=/Users/dengjie/workspace/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


### SSL End

然后，启动Kafka服务，执行脚本如下：

$KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties &

2.4 Kafka Eagle如何集成SSL？

目前Kafka Eagle支持SASL和SSL多种安全认证协议，这里我们来说说如何配置SSL协议（Kafka Eagle V1.4.7以后的版本支持SSL）。想要了解配置SASL协议的，可以阅读Kafka Eagle官方文档来进行配置。

下载最新版本，然后解压，接着配置环境变量，最后配置system-config.properties文件。配置内容如下所示：

######################################
# kafka ssl authenticate
######################################
cluster3.kafka.eagle.ssl.enable=true
cluster3.kafka.eagle.ssl.protocol=SSL
# kafka server.properties "ssl.truststore.location" value
cluster3.kafka.eagle.ssl.truststore.location=/data/kafka/ssl/certificates/kafka.truststore
# kafka server.properties "ssl.truststore.password" value
cluster3.kafka.eagle.ssl.truststore.password=ke123456
# kafka server.properties "ssl.keystore.location" value
cluster3.kafka.eagle.ssl.keystore.location=/data/kafka/ssl/certificates/kafka.keystore
# kafka server.properties "ssl.keystore.password" value
cluster3.kafka.eagle.ssl.keystore.password=ke123456
# kafka server.properties "ssl.key.password" value
cluster3.kafka.eagle.ssl.key.password=ke123456

这里需要注意的是，Kafka Eagle支持CGroups。例如，你的当前SSL认证信息，只能访问一部分的Topics，而Kafka集群除了这些你能访问的Topics，还有写Topics你是没有权限访问。这样当你开启了Kafka Eagle的CGroup特性后，Kafka Eagle会将你的SSL认证权限范围缩小你当前能访问的Topics集合里面，配置内容如下所示：

# SASL
cluster3.kafka.eagle.sasl.cgroup.enable=true
cluster3.kafka.eagle.sasl.cgroup.topics=topic1,topic2,topic3

2.5 Kafka  Eagle SSL协议预览

访问Kafka Eagle的页面，预览截图如下所示：

3.总结

在安装SSL时，需要使用keytool和openssl命令，如果提供命令不存在，需要提前安装好这个命令。而在使用Kafka Eagle集成SSL协议时，各个属性的值需要设置正确，比如密码、证书路径等。

4.结束语

这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！