大数据集群安全系列之kafka使用SSL加密认证

Apache kafka允许客户端通过SSL进行连接。默认情况下SSL是禁止状态，可以根据需要开启。本文就是浪尖亲测，开启SSL认证。

1，为每一个kafka Broker创建SSL key和证书

第一步，部署HTTPS需要为在集群中的每台Broker创建key和证书。可以使用java的keytool完成这个任务。我们最初会将密钥生成到一个临时密钥库，以便我们稍后可以导出和签名。

keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA

在上面的命令中，你需要指定两个参数：

A),keystore:保存证书的keystore file。由于keystore file保存证书的私钥，因此，必须安全的保存起来。

B),validity(合法性)：以天为单位的证书有效时间。

注释：

默认情况，ssl.endpoint.identification.algorithm 属性所没有定义，所以hostname认证不执行。为了使能hostname 认证，设置下列属性：

ssl.endpoint.identification.algorithm=HTTPS

一旦启用，客户端将根据以下两个字段之一验证服务器的完全限定域名（FQDN）：

A),通用名称（CN）

B),主题备用名称（SAN）

两个字段都有效，但RFC-2818建议使用SAN。 SAN也更灵活，允许声明多个DNS条目。 另一个优点是，CN可以设置为更有意义的价值用于授权目的。要添加SAN字段，请将以下参数-ext SAN = DNS：{FQDN}附加到keytool命令中：

keytool -keystore server.keystore.jks -alias localhost -validity {validity} -genkey -keyalg RSA -ext SAN=DNS:{FQDN}

以后可以运行以下命令来验证生成的证书的内容：

keytool -list -v -keystore server.keystore.jks

2，创建你自己的CA

通过第一步，在集群中的每台机器都有一对公钥私钥和一个证书去识别。然而，证书是无标记的，意味着攻击者可以通过创建相同的证书假装任何机器。

因此，通过为群集中的每台机器签名来防止伪造的证书很重要。认证机构（CA）负责签发证书。认证机构就像发行护照的政府，政府会对每张护照盖章，使得护照很难被伪造。其它，政府核实印章，以保证此护照是真实的。类似的，CA签署证书，密码保证签署的证书在计算上很难被伪造。因此，只要CA是一个真正值得信赖的权威机构，客户就可以很高的保证他们正在连接到真实的机器。

openssl req -new -x509 -keyout ca-key -out ca-cert -days 365

生成的CA仅仅是一个公钥 - 私钥对和证书，它用于签署其他证书。

下一步是将生成的CA添加到**clients' truststore（客户的信任库）**，以便client可以信任这个CA:

keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert

注释：

如果你通过设置kafka Broker 配置文件的ssl.client.auth属性为"requested" 或者"required"，来配置kafka Broker 要求客户端认证。那你必须为kafka Broker提供信托库及所有客户端签名了的CA证书密匙。使用下面的命令：

keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert

相反，在步骤1中密钥库存储每个机器自己的身份。client的信任库存储所有client应该信任的证书，将证书导入到一个信任库也意味着信任由该证书签名的所有证书，正如上面的比喻，信任政府（CA）也意味着信任所有护照（证书），这种属性成为信任链，当在一个大的kafka集群上部署SSL时，它是特别有用的。你可以用一个CA登录集群中的所有证书，并拥有所有机器相同的信任库，信任CA，这样所有的机器可以验证其他的机器了。

3，签名

下一步是使用步骤2中生成的CA对步骤1生成的所有证书进行签名。首先，需要从密钥库导出证书：

keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file

然后用CA签名：

openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days {validity} -CAcreateserial -passin pass:{ca-password}

最后，您需要将CA证书和签名的证书导入到密钥库中：

keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert

keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed

参数的定义如下：

keystore: 密钥库的位置

ca-cert: CA的证书

ca-key: CA的私钥

ca-password: CA的密码

cert-file: 出口，服务器的未签名证书

cert-signed: 服务器的签名证书

这是一个包含所有上述步骤的bash脚本的示例。请注意，其中一个命令假定密码为`test1234`，因此在运行该密码之前，请使用该密码或编辑该命令。密码：test1234。其它可以按照提示自填。

#!/bin/bash
#Step 1
keytool -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
#Step 2
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
#Step 3
keytool -keystore server.keystore.jks -alias localhost -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial -passin pass:test1234
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore server.keystore.jks -alias localhost -import -file cert-signed

4，配置kafka Brokers

Kafka Brokers支持监听多端口，多个之间必须使用逗号分开：

listeners=PLAINTEXT://host.name:port,SSL://host.name:port

对于，没有设置ssh inter-Broker交流， PLAINTEXT 和SSL 都是需要的

下列SSL配置是在Broker端必须的。

ssl.keystore.location=/var/private/ssl/server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.truststore.location=/var/private/ssl/server.truststore.jks
ssl.truststore.password=test1234

5，客户端的配置

SSL仅仅支持Kafka新版本的producer和Consumer，老的API并不支持。生产者和消费者的SSL相关配置是一样的。

如果客户端的认证Broker端不要求，那么下面是最简单的配置

security.protocol=SSL
ssl.truststore.location=/var/private/ssl/client.truststore.jks
ssl.truststore.password=test1234

注释：

ssl.truststore.password是技术上可以选但是强烈推荐的。如果客户端认证是必须的，那么keystore必须像步骤1一样创建，下面也必须配置：

ssl.keystore.location=/var/private/ssl/client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

6，使用案例

A),Console-producer和console-Consumer的使用案例。

kafka-console-producer.sh --broker-list localhost:9093 --topic ada --producer.config client-ssl.properties
kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic test --consumer.config client-ssl.properties

B),代码中的使用案例

生产者消费者一样，这里采用的是server端的client.truststore.jks和server.keystore.jks

props.put("security.protocol", "SSL");
props.put("ssl.truststore.location", "/opt/modules/kafka_2.10-0.10.0.1/client.truststore.jks");
props.put("ssl.truststore.password", "test1234");
props.put("ssl.keystore.location", "/opt/modules/kafka_2.10-0.10.0.1/server.keystore.jks");
props.put("ssl.keystore.password", "test1234");
props.put("ssl.key.password", "test1234");

只是很浪费性能，影响整体吞吐量，慎用。