# 配置Kafka的SSL双向认证 SSL双向认证是指在通信认证时，将同时验证客户端和服务端的证书，以确保通信中涉及的双方都是受信任的。 开启SSL双向认证，主要用于对安全性要求较高的场景。 **SSL双向认证功能为受限使用，需要提交工单申请开通。** 配置SSL双向认证的流程如[图1]所示。 图1配置SSL双向认证流程图 ![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0000001991764425.png) #### 约束与限制 - 开启/关闭SSL双向认证将重启实例，请谨慎操作。 - 开启/关闭SSL双向认证后，无法再回退使用Kafka的默认证书。 #### 前提条件 - 已准备一台Linux系统的服务器，并安装[Java Development Kit 1.8.111或以上版本](https://www.oracle.com/java/technologies/downloads/#java8)，配置JAVA_HOME与PATH环境变量。 - 实例已开启Kafka SASL_SSL。 #### 步骤一：制作客户端验证服务端的证书 1. 登录Linux系统的服务器，执行以下命令，生成server.keystore.jks证书的孵化器。 ``` keytool -genkey -keystore server.keystore.jks -alias localhost -validity 3650 -keyalg RSA ``` 按照提示信息输入keystore密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求： - 8\~32个字符。 - 至少包含以下字符中的3种：大写字母、小写字母、数字、特殊字符\`\~!@#$%\^\&\*()-_=+\\\|\[{}\]:'",\<.\>/?和空格，并且不能以-开头。 - 非弱密码（如果不能确定哪些是弱密码，请在[开启SSL双向认证对话框]中输入密码，系统会自动判定是否为弱密码）。 按照提示信息输入证书拥有者的信息，包含姓名、单位、组织名称、所在城市、所在省、所在国家，示例如下。 ``` [root@ecs-kafka ~]# keytool -genkey -keystore server.keystore.jks -alias localhost -validity 3650 -keyalg RSA Enter keystore password: Re-enter new password: What is your first and last name? [Unknown]: Tom What is the name of your organizational unit? [Unknown]: test What is the name of your organization? [Unknown]: test01 What is the name of your City or Locality? [Unknown]: nj What is the name of your State or Province? [Unknown]: js What is the two-letter country code for this unit? [Unknown]: xx Is CN=Tom, OU=test, O=test01, L=nj, ST=js, C=xx correct? [no]: y ``` 2. 执行以下命令，创建CA的认证证书。 ``` openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650 ``` 按照提示信息输入PEM密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求：4\~1024个字符。 按照提示信息输入证书拥有者的信息，包含所在国家、所在省、所在城市、组织名称、单位、姓名、邮箱，示例如下。 ``` [root@ecs-kafka ~]# openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650 Generating a RSA private key ............................................................................................................................................+++++ ..........+++++ writing new private key to 'ca-key' Enter PEM pass phrase: Verifying - Enter PEM pass phrase: ----- You are about to be asked to enter information that will be incorporated into your certificate request. What you are about to enter is what is called a Distinguished Name or a DN. There are quite a few fields but you can leave some blank For some fields there will be a default value, If you enter '.', the field will be left blank. ----- Country Name (2 letter code) [XX]:xx State or Province Name (full name) []:js Locality Name (eg, city) [Default City]:nj Organization Name (eg, company) [Default Company Ltd]:test01 Organizational Unit Name (eg, section) []:test Common Name (eg, your name or your server's hostname) []:Tom Email Address []:xx [root@ecs-kafka ~]# ``` 3. 执行以下命令，从[1]中生成的server.keystore.jks中导出证书，并命名为"server.crt"。 ``` keytool -keystore server.keystore.jks -alias localhost -certreq -file server.crt ``` 按照提示信息输入[1]中记录的keystore密码。 4. 执行以下命令，使用CA的私钥对server.crt进行签名，并为签名后的证书命名为"server-signed.crt"。 ``` openssl x509 -req -CA ca-cert -CAkey ca-key -in server.crt -out server-signed.crt -days 3650 -CAcreateserial ``` 按照提示信息输入[2]中记录的PEM密码。 5. 执行以下命令，将CA证书和"server-signed.crt"导入密钥仓库。 ``` keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert keytool -keystore server.keystore.jks -alias localhost -import -file server-signed.crt ``` 按照提示信息输入[1]中记录的keystore密码。 出现如下提示信息时，输入"y"。 ``` Trust this certificate？ ``` 6. 执行以下命令，使客户端信任服务端证书。 ``` keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert ``` 按照提示信息输入client.truststore.jks的密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求： - 8\~32个字符。 - 至少包含以下字符中的3种：大写字母、小写字母、数字、特殊字符\`\~!@#$%\^\&\*()-_=+\\\|\[{}\]:'",\<.\>/?和空格，并且不能以-开头。 - 非弱密码（如果不能确定哪些是弱密码，请在[开启SSL双向认证对话框]中输入密码，系统会自动判定是否为弱密码）。 出现如下提示信息时，输入"y"。 ``` Trust this certificate？ ``` 7. 导出"client.truststore.jks"和"server.keystore.jks"证书到本地PC机。 "server.keystore.jks"在后续步骤"开启SSL双向认证"时，用来替换"Keystore文件"。"client.truststore.jks"需要存放到客户端中，并记录存放路径，此路径用于客户端连接Kafka实例时，配置文件中"ssl.truststore.location"参数的值。 #### 步骤二：制作服务端验证客户端的证书 1. 登录Linux系统的服务器，执行以下命令，生成client.keystore.jks证书的孵化器。 ``` keytool -genkey -keystore client.keystore.jks -alias localhost -validity 3650 -keyalg RSA ``` 按照提示信息输入keystore密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求： - 8\~32个字符。 - 至少包含以下字符中的3种：大写字母、小写字母、数字、特殊字符\`\~!@#$%\^\&\*()-_=+\\\|\[{}\]:'",\<.\>/?和空格，并且不能以-开头。 - 非弱密码（如果不能确定哪些是弱密码，请在[开启SSL双向认证对话框]中输入密码，系统会自动判定是否为弱密码）。 按照提示信息输入证书拥有者的信息，包含姓名、单位、组织名称、所在城市、所在省、所在国家，示例如下。 ``` [root@ecs-kafka ~]# keytool -genkey -keystore client.keystore.jks -alias localhost -validity 3650 -keyalg RSA Enter keystore password: Re-enter new password: What is your first and last name? [Unknown]: Tom What is the name of your organizational unit? [Unknown]: test What is the name of your organization? [Unknown]: test01 What is the name of your City or Locality? [Unknown]: nj What is the name of your State or Province? [Unknown]: js What is the two-letter country code for this unit? [Unknown]: xx Is CN=Tom, OU=test, O=test01, L=nj, ST=js, C=xx correct? [no]: y ``` 2. 执行以下命令，创建CA的认证证书。 ``` openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650 ``` 按照提示信息输入PEM密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求：4\~1024个字符。 按照提示信息输入证书拥有者的信息，包含所在国家、所在省、所在城市、组织名称、单位、姓名、邮箱，示例如下。 ``` [root@ecs-kafka ~]# openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650 Generating a RSA private key ............................................................................................................................................+++++ ..........+++++ writing new private key to 'ca-key' Enter PEM pass phrase: Verifying - Enter PEM pass phrase: ----- You are about to be asked to enter information that will be incorporated into your certificate request. What you are about to enter is what is called a Distinguished Name or a DN. There are quite a few fields but you can leave some blank For some fields there will be a default value, If you enter '.', the field will be left blank. ----- Country Name (2 letter code) [XX]:xx State or Province Name (full name) []:js Locality Name (eg, city) [Default City]:nj Organization Name (eg, company) [Default Company Ltd]:test01 Organizational Unit Name (eg, section) []:test Common Name (eg, your name or your server's hostname) []:Tom Email Address []:xx [root@ecs-kafka ~]# ``` 3. 执行以下命令，从[1]中生成的client.keystore.jks中导出证书，并命名为"client.crt"。 ``` keytool -keystore client.keystore.jks -alias localhost -certreq -file client.crt ``` 按照提示信息输入[1]中记录的keystore密码。 4. 执行以下命令，使用CA的私钥对client.crt进行签名，并为签名后的证书命名为"client-signed.crt"。 ``` openssl x509 -req -CA ca-cert -CAkey ca-key -in client.crt -out client-signed.crt -days 3650 -CAcreateserial ``` 按照提示信息输入[2]中记录的PEM密码。 5. 执行以下命令，将CA证书和"client-signed.crt"导入密钥仓库。 ``` keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert keytool -keystore client.keystore.jks -alias localhost -import -file client-signed.crt ``` 按照提示信息输入[1]中记录的keystore密码。 出现如下提示信息时，输入"y"。 ``` Trust this certificate？ ``` 6. 执行以下命令，使服务端信任客户端证书。 ``` keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert ``` 按照提示信息输入server.truststore.jks的密码，并记录此密码，此密码在后续步骤中会使用。 密码需要满足如下要求： - 8\~32个字符。 - 至少包含以下字符中的3种：大写字母、小写字母、数字、特殊字符\`\~!@#$%\^\&\*()-_=+\\\|\[{}\]:'",\<.\>/?和空格，并且不能以-开头。 - 非弱密码（如果不能确定哪些是弱密码，请在[开启SSL双向认证对话框]中输入密码，系统会自动判定是否为弱密码）。 出现如下提示信息时，输入"y"。 ``` Trust this certificate？ ``` 7. 导出"server.truststore.jks"和"client.keystore.jks"证书到本地PC机。 "server.truststore.jks"在后续步骤"开启SSL双向认证"时，用来替换"Truststore文件"。"client.keystore.jks"需要存放到客户端中，并记录存放路径，此路径用于客户端连接Kafka实例时，配置文件中"ssl.keystore.location"参数的值。 #### 步骤三：开启SSL双向认证 1. 登录[Kafka控制台](https://console.huaweicloud.com/dms/?engine=kafka&locale=zh-cn#/queue/manager/newKafkaList)。 2. 在管理控制台左上角单击![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0143929918.png)，选择Kafka实例所在的区域。 3. 单击待开启SSL双向认证的实例名称，进入实例详情页面。 4. 在"连接信息 \> SSL双向认证"后，单击![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0000001191767177.png)，开启SSL双向认证。 5. 在弹出的"SSL双向认证"对话框中，参考[表1]，设置开启SSL双向认证的参数。 **开启SSL双向认证将重启实例，请谨慎操作。** 图2开启SSL双向认证 ![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0000001886347210.png "点击放大") 表1开启SSL双向认证参数说明 | 参数名称 | 说明 | |:---|:---| | Key密码 | 输入[server.keystore.jks的密码] | | Keystore密码 | 输入[server.keystore.jks的密码] | | Keystore文件 | 导入"server.keystore.jks"证书 | | Truststore密码 | 输入[server.truststore.jks的密码] | | Truststore文件 | 导入"server.truststore.jks"证书 | 6. 单击"确定"，完成开启SSL双向认证。 #### 步骤四：修改客户端配置文件 开启SSL双向认证后，需要在客户端的"consumer.properties"和"producer.properties"文件中，分别修改服务端证书配置，并增加客户端证书配置。 ``` security.protocol=SSL ssl.truststore.location=/opt/kafka_2.12-2.7.2/config/client.truststore.jks ssl.truststore.password=axxxb ssl.endpoint.identification.algorithm= #增加以下的客户端证书配置 ssl.keystore.location=/var/private/ssl/kafka/client.keystore.jks ssl.keystore.password=txxx3 ssl.key.password=txxx3 ``` 表2配置文件参数说明 | 参数名称 | 说明 | |:---|:---| | security.protocol | 证书协议类型，开启SSL双向认证时，必须设置为**SSL**。 | | ssl.truststore.location | [client.truststore.jks]证书的存放路径。 | | ssl.truststore.password | [client.truststore.jks的密码]。 | | ssl.endpoint.identification.algorithm | 证书域名校验开关，为空则表示关闭。这里需要**保持关闭状态，必须设置为空**。 | | ssl.keystore.location | [client.keystore.jks]证书的存放路径。 | | ssl.keystore.password | [client.keystore.jks的密码]。 | | ssl.key.password | [client.keystore.jks的密码]。 | #### 步骤五：验证证书是否生效 参考[使用客户端连接Kafka（密文接入）](https://support.huaweicloud.com/usermanual-kafka/kafka-ug-180801001.html)，生产和消费消息。如果成功，表示替换的证书已生效。 #### 关闭SSL双向认证 1. 登录[Kafka控制台](https://console.huaweicloud.com/dms/?engine=kafka&locale=zh-cn#/queue/manager/newKafkaList)。 2. 在管理控制台左上角单击![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0143929918.png)，选择Kafka实例所在的区域。 3. 单击待关闭SSL双向认证的实例名称，进入实例详情页面。 4. 在"连接信息 \> SSL双向认证"后，单击![](https://support.huaweicloud.com/usermanual-kafka/zh-cn_image_0000001930152461.png)，关闭SSL双向认证。 ![](https://support.huaweicloud.com/usermanual-kafka/public_sys-resources/caution_3.0-zh-cn.png) 关闭SSL双向认证将重启实例，请谨慎操作。 5. 关闭SSL双向认证后，需要在客户端的"consumer.properties"和"producer.properties"文件中，分别修改服务端证书协议类型，并删除客户端证书配置。 ``` security.protocol=SASL_SSL ssl.truststore.location=/opt/kafka_2.12-2.7.2/config/client.truststore.jks ssl.truststore.password=axxxb ssl.endpoint.identification.algorithm= #删除以下的客户端证书配置 ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks ssl.keystore.password=txxx3 ssl.key.password=txxx3 ``` security.protocol配置证书协议类型，关闭SSL双向认证时，必须设置为SASL_SSL。"ssl.truststore.location"、"ssl.truststore.password"和"ssl.endpoint.identification.algorithm"无需修改。