前言
AutoMQ[1] 是一种基于云重新设计的流处理系统，在保持与 Apache Kafka[2] 100% 兼容的前提下，AutoMQ 通过将存储分离至对象存储，显著提升了系统的成本效益和弹性能力。具体来说，AutoMQ 通过构建在 S3 上的流存储库 S3Stream，将存储卸载至共享云存储 EBS 和 S3，提供低成本、低延时、高可用、高可靠和无限容量的流存储能力。与传统的 Shared Nothing 架构相比，AutoMQ 采用了 Shared Storage 架构，显著降低了存储和运维的复杂性，同时提升了系统的弹性和可靠性。
得益于 AutoMQ 对 Kafka 的 100% 兼容，对于安全认证配置 AutoMQ 与 Kafka 的实现是相同的。通过本文你可以掌握如何通过 SSL 安全的启动 AutoMQ。
使用自签名证书完成 SSL 认证配置
第一节本文将会基于自认证的 SSL 证书来对 AutoMQ 进行 SSL 安全协议配置，如果你对 SSL 这个协议本身并没有太多了解，那么在往下前阅读之前，了解下面这些知识会对你有一定的帮助。
密钥对：通常由一个公钥和一个私钥组成。公钥可以公开分发，用于加密数据或验证数字签名；私钥则必须安全保管，用于解密由公钥加密的数据或创建数字签名。
数字签名：用私钥给信息摘要(通常是哈希值)进行加密生成的产物。
证书签名请求（CSR）：被用于生成数字签名证书，CSR 中包含了申请签名者的公钥以及申请者的身份信息（组织名称、域名等）。
可被信任的签名证书：由公众信任的机构使用他们的私钥对 CSR 进行数字签名的产物。
自签名证书：使用私钥对 CSR 进行数字签名的产物，但是 CSR 中包含的公钥以及进行签名的私钥都是属于同一机构或个人。
信任库和密钥库：在本文中两者对应的都是我们所生成的.jks文件，两者可代指同一个文件，但是实际开发中建议把信任库和密钥库分开配置。
为每个 Broker 生成 SSL 密钥对
部署一个或多个支持 SSL 的 Broker 的第一步是为每台服务器生成一对公钥与私钥。我们将使用 Java 的 keytool 命令来为你需要配置的 Broker 生成密钥对。
keytool -keystore {keystorefile} - alias {localhost} -validity {validity} -genkey -keyalg RSA [-ext SAN=DNS:{FQDN},IP:{IPADDRESS1}]
配置参数含义：
keystrore: 指定密钥库文件的位置和名称。
alias: 为密钥条目指定一个别名，用于在密钥库中唯一标识此密钥。
validity: 设置用它生成的自签名证书的有效期天数。
genkey: 表示要生成一个新的密钥对。
keyalg: 指定密钥对使用的算法。
ext: 补充信息。
如果你想在证书中添加主机名信息（用于后面的主机名验证），你需要使用扩展参数 -ext SAN=DNS:{FQDN},IP:{IPADDRESS1}。
本文示例：
keytool - keystore server.keystore.jks - alias localhost - validity 365 - genkey - keyalg RSA
使用后会生成一个使用 RSA 算法生成的密钥对，并将其存储在名为server.keystore.jks的文件中。密钥对的别名为 localhost，用它签出证书的有效期为365天。如果文件不存在，会生成在指定路径下，且会要求你设置密码以及各种个人信息。
配置主机名验证
启用 "主机名验证 "后，将根据服务器的实际主机名或 IP 地址检查所连接服务器提供的证书中的属性，以确保连接到了正确的服务器。
进行此类检查的主要目的是防止中间人攻击(下图所示)。从 Kafka 2.0.0 版本开始，当我们使用 SSL 安全协议启动 Kafka 服务时，服务器的主机名验证默认已启用。
如果 Client 启用了主机名验证，那么它将将根据以下两个字段之一验证服务器的完全合格域名（FQDN）或 IP 地址。
Common Name (CN)
Subject Alternative Name (SAN)
虽然 Kafka 检查这两个字段，但自2000年以来，使用通用名称（common name/CN）字段进行主机名验证已经废弃，应尽可能避免使用。此外，SAN字段更加灵活，允许在一个证书中声明多个 DNS 和 IP 条目。在上一小节中，我们已经给出了声明的配置方案。
为服务配置主机名验证显然是确保 Kafka 安全通信的最佳实践之一。然而，如果你只是想快速启动配置了 SSL 协议的 Kafka 服务，可以将配置文件中的 ssl.endpoint.identification.algorithm 属性设置为空字符串，以禁用服务器主机名验证。本文也将暂时禁用此属性，以便读者能够快速启动一个启用 SSL 的 Broker，同时也会提供配置主机名验证的命令供参考。
创建自己的CA
完成以上步骤后，你需要配置 SSL 协议的 Kafka 服务器都应该已经拥有了一个密钥对，这是创建证书的基础。为了增强身份验证功能，通常需要将签名请求提交给公众信任机构进行数字签名，这里的公众信任的机构通常被我们称为证书颁发机构（CA）。
证书颁发机构负责签署证书。证书颁发机构就像一个颁发护照的政府（有公信力），政府在每本护照上盖章（数字签名），使护照难以伪造。其他人会对印章进行验证，以确保护照的真实性。密码学保证了 CA 签署的证书在计算上难以伪造。因此，只要 CA 是一个真实可信的机构，人们就能保证连接到的是他们需要连接的服务器。
在本教程的第一部分中，我们将使用自己创建的 CA 完成配置 。而在企业环境中建立生产集群时，通常由公司内部可信的企业 CA 签发证书。关于如何配置由他人签发的证书，我们将在本文的第二部分进行说明。
接下来我们将用 OpenSSL 生成 CA 和签署证书，这个加密库包含主要的加密算法、常用的密钥和证书管理功能，请根据你的机型选择合适的安装方式进行安装，具体安装步骤这里不再赘述。
现在你应该准备的初始文件如下 （不配置主机名验证则可以跳过该配置） ：
ca/ ├── openssl.cnf ├── serial.txt ├── index.txt
下面是本次 Openssl 使用的 opensll.cnf 配置文件，可以根据自身需求调整其参数，serial.txt，index.txt 这两个文件是用于跟踪哪些证书是由该 CA 签发的，如果不需要可以在下列配置文件中注释掉该配置。
由于 Openssl 的 bug，x509 模块不会将请求的扩展字段从证书请求签名复制到最终证书中。制作 CA 需要依靠指定 config 的方式才能确保扩展字段被复制到最终的证书中。
HOME = . RANDFILE = $ENV ::HOME/.rnd #################################################################### [ ca ] default_ca = CA_default # The default ca section[ CA_default ] base_dir = . certificate = $base_dir /cacert.pem # The CA certificate private_key = $base_dir /cakey.pem # The CA private key new_certs_dir = $base_dir # Location for new certs after signing database = $base_dir /index.txt # Database index file serial = $base_dir /serial.txt # The current serial number default_days = 1000 # How long to certify for default_crl_days = 30 # How long before next CRL default_md = sha256 # Use public key default MD preserve = no # Keep passed DN ordering x509_extensions = ca_extensions # The extensions to add to the cert email_in_dn = no # Don't concat the email in the DN copy_extensions = copy # Required to copy SANs from CSR to cert #################################################################### [ req ] default_bits = 4096 default_keyfile = cakey.pem distinguished_name = ca_distinguished_name x509_extensions = ca_extensions string_mask = utf8only #################################################################### [ ca_distinguished_name ] countryName = Country Name (2 letter code) countryName_default = DE stateOrProvinceName = State or Province Name (full name) stateOrProvinceName_default = Test Province localityName = Locality Name (eg, city) localityName_default = Test Town organizationName = Organization Name (eg, company) organizationName_default = Test Company organizationalUnitName = Organizational Unit (eg, division) organizationalUnitName_default = Test Unit commonName = Common Name (e.g. server FQDN or YOUR name) commonName_default = Test Name emailAddress = Email Address emailAddress_default = test @test.com #################################################################### [ ca_extensions ] subjectKeyIdentifier = hash authorityKeyIdentifier = keyid:always, issuer basicConstraints = critical, CA: true keyUsage = keyCertSign, cRLSign #################################################################### [ signing_policy ] countryName = optional stateOrProvinceName = optional localityName = optional organizationName = optional organizationalUnitName = optional commonName = supplied emailAddress = optional #################################################################### [ signing_req ] subjectKeyIdentifier = hash authorityKeyIdentifier = keyid,issuer basicConstraints = CA:FALSE keyUsage = digitalSignature, keyEncipherment
两种生成 CA 指令如下：
各个参数含义：
openssl req执行证书请求操作。
x509使用 x509 模块。X.509是密码学里公钥证书的格式标准。
config指定 OpenSSL 配置文件的路径和名称，不指定则使用默认路径的配置。
nodes不加密私钥，生成的私钥将不会被加密，以便在操作中使用时不需要输入密码。
keyout指定生成密钥的路径和名称。
out指定生成的证书文件的路径和名称。
执行命令后，将会生成一个密钥对cakey.pem以及 X.509 格式的公钥证书cacert.pem。
其中，cakey.pem 中的私钥将用于签署证书，而 cacert.pem 则需要配置在客户端的信任库中，以便客户端能够通过此证书中的公钥验证服务端发来的证书是否由此 CA 签署。这个密钥对应该让它非常安全，如果有人获取了它，他们就可以创建和签署被你的基础架构信任的证书，这意味着他们在连接到任何信任该 CA 的服务时，可以冒充任何人。
创建信任库
接下来我们需要将生成的 CA 公钥证书cacert.pem添加到客户端的信任库，以便客户端可以信任该 CA：
keytool -keystore
client.truststore.jks - alias CARoot -import -file cacert.pem # 这个truststore是在客户端操作的，将我们生成的CA证书放入客户端的信任链里面
新参数说明：
import表示要导入一个证书。
file指定要导入的证书文件的路径和名称。
这条命令代表证书cacert.pem将被导入到客户端信任库 client.truststore.jks 中，并使用别名 CARoot 进行标识。
注意 如果你在 Kafka Brokers 配置中将ssl.client.auth 设置为 "requested" 或 "required" 来配置 Kafka Brokers 以要求客户端身份验证，则必须为 Kafka Brokers 也提供一个 truststore，其中应包含为客户端签署证书的 CA 证书，这里我们客户端的证书也是由我们自己生成的 CA 签署生成，所以这里直接把我们的 CA 证书导入到服务端即可。
keytool -keystore server.truststore.jks - alias CARoot -import -file cacert.pem # 这个truststore是在服务端操作的，将给客户端证书签名的CA证书也导入到服务端的信任链里面，没有设置ssl.client.auth属性的话就没必要做这个操作
当ssl.client.auth也被设置时，我们就建立了 SSL 双向认证，即当客户端和服务端连接时，不仅客户端要验证服务端的证书有效性，服务端也要验证客户端的证书有效性，等双方都认证通过了，才能开始建立安全通信通道进行数据传输。本篇教程均采用单向认证部署，即服务器不会验证客户端的证书。
签署证书
现在我们想要为服务器配置上 CA 签署的证书，首先需要用我们第一步为服务器生成的密钥对创建一个证书签名请求（CSR）。这里继续使用 keytool 工具创建。
keytool -keystore ssl.keystore.jks - alias localhost -certreq -file cert-file # cert-file即我们生成的csr文件
然后用我们创建的 CA 进行证书签署（分为需要与不需要主机认证）：
#不需要主机认证的时候 openssl x509 -req -CA cacert.pem -CAkey cakey.pem - in cert-file -out cert-signed -days 3650 -CAcreateserial -passin pass:123456
参数说明：
x509用于处理 X.509 证书的 OpenSSL 子命令。
req表示将输入视为证书请求。
CA指定 CA 证书文件，用于对证书请求进行签署。
CAkey指定 CA 的私钥文件，用于对证书请求进行签署。
in指定待签署的证书请求文件。
out指定生成的已签署证书的输出文件。
CAcreateserial表示创建一个序列号文件，并将其添加到 CA 证书中，以便跟踪已签署的证书。
passin pass123456: 指定用于解锁私钥的密码，这里密码为“123456”。
#需要主机验证的时候 openssl ca -config openssl.cnf -policy signing_policy -extensions signing_req -out cert-signed - in cert-file
参数说明：
openssl ca调用 OpenSSL 的 CA 功能，用于签署证书签名请求(CSR)并生成证书。
config指定配置文件的路径和名称。
policy指定证书签发的策略。
extensions指定扩展名，如果配置中有设置的话。
执行命令后，将会生成一个服务器证书，并保存为cert-signed。
最后，需要将已签名的证书导入密钥库：
keytool -keystore ssl.keystore.jks - alias localhost -import -file cert-signed
此外，如果集群配置 SSL 的话那么每个节点都将有一个 ssl.keystore.jks 文件，其中包含该节点的密钥对、签名证书。
所有 Client 和 Broker 都可以使用相同的信任库，因为其中不包含任何敏感信息。
配置 Kafka Broker
如果内部 Broker 通讯不需要启用 SSL，那么只需要如下配置：
可选设置
ssl.client.auth=none(可选值为"required"，表示需要客户端验证，且客户端必须提供有效证书；以及"requested"，表示需要客户端验证，但没有证书的客户端仍可连接。不鼓励使用 "requested"，因为它会提供虚假的安全感，配置错误的客户端仍可成功连接。)
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1 （列出要从客户端接受的 SSL 协议。请注意，SSL 已被弃用，取而代之的是 TLS，因此不建议在生产中使用 SSL）
如果要在 Broker 之间通信中启用 SSL，请在 server.properties 文件中添加以下内容：
security.inter.broker.protocol=SSL
配置 Kafka Client
从 Kafka 2.0.0 开始，服务器的主机名验证在客户端连接和 Broker 间连接中都被默认启用，可通过将 SSL.endpoint.identification.algorithm 设置为空字符串来关闭主机名验证。
client-ssl.properties文件
security.protocol=SSL ssl.truststore.location=/root/automq/ssl/ssl.keystore.jks ssl.truststore.password=123456 #关闭主机名验证时设置以下参数 ssl.endpoint.identification.algorithm=
配置文件启动 AutoMQ
bin/kafka-server-start.sh /root/automq/config/kraft/ssl.properties
测试连接
创建Topic
bin/kafka-topics.sh --bootstrap-server 47.253.200.218:9092 --command-config /root/automq/ssl/client-ssl.properties --create --topic test
创建成功
生产者 && 消费者测试
bin/kafka-console-producer.sh --bootstrap-server 47.253.200.218:9092 --topic test --producer.config /root/automq/ssl/client-ssl.properties bin/kafka-console-consumer.sh --bootstrap-server 47.253.200.218:9092 --topic test --consumer.config /root/automq/ssl/client-ssl.properties
收发消息成功
总结
至此，已经完成了SSL相关的所有配置，大体流程如下图所示：
使用云厂商颁发的证书完成 SSL 认证配置
通常，云厂商如阿里云[3]等会提供 SSL 证书服务，支持用户为已有域名申请证书。审核通过后，平台会提供各种格式的证书下载选项，以及根证书的下载方式。本段教程是基于阿里云提供的证书配置。
配置 Kafka Broker
云厂商提供下载的 JKS 文件中包含了所有我们需要的信息，如数字签名证书、密钥对。我们可以通过下载获得如automq.space.jks这种格式的文件，可以直接按如下配置即可：
配置 Kafka Client
client-ali-ssl.properties文件
security.protocol=SSL ssl.truststore.location=/root/automq/ssl/automq.space.jks ssl.truststore.password=mhrx2d7h #关闭主机名验证时设置以下参数 ssl.endpoint.identification.algorithm=
如果我们此时进行连接可能会发现出现如下报错
这通常表示客户端无法验证服务端提供的证书，这是由于客户端信任库中缺少必要的证书，即给服务端签署证书的 CA 证书并不在我们客户端的信任库中，这里我们需要下载云厂商提供的根证书来配置客户端。具体云厂商提供方式不同，这里请读者自行探索，具体可参考阿里云证书相关文档[4]。
为客户端信任库添加根证书文件
这里导入的方式依旧是使用 keytool 导入。
keytool -import -file /root/automq/ssl/DigicertG2ROOT.cer -keystore
client.truststore.jks - alias root-certificate
重写client-ali-ssl.properties文件
security.protocol=SSL ssl.truststore.location=/root/automq/ssl/client.truststore.jks ssl.truststore.password=123456 #关闭主机名验证时设置以下参数 ssl.endpoint.identification.algorithm=
配置文件启动 AutoMQ
bin/kafka-server-start.sh /root/automq/config/kraft/ssl.properties
测试链接
创建Topic
bin/kafka-topics.sh --bootstrap-server 47.253.200.218:9092 --command-config /root/automq/ssl/client-ali-ssl.properties --create --topic test1
创建成功
生产者 && 消费者测试
bin/kafka-console-producer.sh --bootstrap-server 47.253.200.218:9092 --topic test1 --producer.config /root/automq/ssl/client-ali-ssl.properties bin/kafka-console-consumer.sh --bootstrap-server 47.253.200.218:9092 --topic test1 --consumer.config /root/automq/ssl/client-ali-ssl.properties
收发消息成功
总结
使用云厂商提供的证书对比自签名证书少了很多步骤，我们能够直接从云厂商拿到根证书以及签名后的证书并直接给服务器配置使用，需要注意的就是自建的客户端并不是天然的拥有根证书，需要我们从云厂商那下载并导入我们自己的信任库中，不然就会导致 SSL 连接失败。
参考资料
[1] AutoMQ: https://www.automq.com
[2] Apache Kafka: https://kafka.apache.org/
[3] Alibaba Cloud: https://www.alibabacloud.com/en/product/certificates?_p_lc=1&spm=a2c5v.8452849.6791778070.10.1859m0OFm0OFVe
[4] 阿里云证书相关文档: https://www.alibabacloud.com/help/en/ssl-certificate/