Using TLS with KeyStore configure

部署TLS的第一步是为集群中的每台机器生成密钥和证书。您可以使用 Java 工具来完成这项任务。我们首先将为每台 Broker 生成一个临时的 keystore，以便我们能够使用 CA 导出和进行签名。

你必须在上面的命令中指定两个参数：

keystore: 这个 keystore 文件保存了证书信息。这个* keystore *文件包含了证书的私有秘钥信息；因此，你必须安全的保存它。
validity: 证书的有效时间。

创建自定义 CA

在第一步之后，集群中的每个 Broker 将包含一个公私钥对，并且拥有用来识别机器的证书。然而，证书是没有签名的，这意味着攻击者可以制作这样一份证书，来假装是任何机器。

因此，重要的是要防止伪造证书，为此要在为集群中的每台机器进行签名。 证书颁发机构(CA)负责签发证书。 CA 的工作内容类似签发护照的政府部门，即每本护照都有政府的印章(签名)，这使得每本护照难以被伪造。其他国家的政府通过验证这个印章，确保护照是真实的。同样，CA 签署证书，并且通过签名加密证书，保证证书在不能被伪造。因此，只要 CA 是一个可以信任的机构，就能确保客户端能够连上真正的服务器。

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

CA 生成的是一个简单的公私钥对和证书，并打算签署其他证书。

下一步是将生成的CA 添加到客户端的信任秘钥库，以便客户端能够信任这个CA：

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

注意：你可以通过在 Broker 上配置lsRequireTrustedClientCertOnConnect为true来配置客户端需要经过认证才能访问。那么你必须总是为每个 Broker 提供一个可信任的库，并且它拥有所有用来给客户端秘钥签名的 CA 证书。

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

与存储每个机器自身身份的密钥库形成对比，客户端信任的秘钥库是存储客户端应信任的所有证书。将证书导入一个可信赖库，意味着信任由该证书签名的所有证书。从上述类推，信任政府（CA）也意味着信任所有已被发布的护照（证书）此属性称为信任链，在大型BookKeeper集群上部署TLS时特别有用。您可以使用单个CA对群集中的所有证书进行签名，并让所有计算机共享信任该CA相同的信任库。这样，所有机器都可以对所有其他机器进行身份验证。

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

然后用CA签署：

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

keytool -keystore broker.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore broker.keystore.jks -alias localhost -import -file cert-signed

参数的定义如下所示：

keystore: 密钥库的位置
ca-cert：CA 的证书
ca-key：CA 的私钥
ca-password：CA 的密码
cert-file：broker 导出的、未签名的证书
cert-signed：broker 的签名证书

配置代理

Brokers 通过提供有效的 brokerServicePortTls 和 webServicePortTls 来启用 TLS，并且还需要将 tlsEnabledWithKeyStore 设置为 true 以使用 KeyStore 类型配置. 除此之外，还需要提供 KeyStore 路径、KeyStore 密码、TrustStore 路径和 TrustStore 密码。并且由于broker会创建内部client/admin客户端来与其他broker进行通信，用户还需要为他们提供配置，这类似于用户如何配置外部client/admin-client。如果 tlsRequireTrustedClientCertOnConnect 是 true，如果客户端证书不受信任，代理将拒绝连接。

代理端需要以下 TLS 配置：

tlsEnabledWithKeyStore=true
# 密钥存储
tlsKeyStoreType=JKS
tlsKeyStore=/var/private/tls/broker.keystore.jks
tlsKeyStorePassword=brokerpw
# 信任存储
tlsTrustStoreType=JKS
tlsTrustStore=/var/private/tls/broker.truststore.jks
tlsTrustStorePassword=brokerpw
# 内部客户端/管理员客户端配置
brokerClientTlsEnabled=true
brokerClientTlsEnabledWithKeyStore=true
brokerClientTlsTrustStoreType=JKS
brokerClientTlsTrustStore=/var/private/tls/client.truststore.jks
brokerClientTlsTrustStorePassword=clientpw
XPath: /pre[8]/code

注意：必须通过文件系统权限限制对存储文件的访问。

Optional settings that may worth consider:

tlsClientAuthentication=false: Enable/Disable using TLS for authentication. This config when enabled will authenticate the other end of the communication channel. It should be enabled on both brokers and clients for mutual TLS.
tlsCiphers=[TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256], A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS network protocol. By default, it is null. OpenSSL Ciphers
tlsProtocols=[TLSv1.3,TLSv1.2] (list out the TLS protocols that you are going to accept from clients). By default, it is not set.

This is similar to [TLS encryption configuing for client with PEM type](/docs/zh-CN/2.8.1/security-tls-transport#Client configuration). For a a minimal configuration, user need to provide the TrustStore information.

例如：

webServiceUrl=https://broker.example.com:8443/
brokerServiceUrl=pulsar+ssl://broker.example.com:6651/
useKeyStoreTls=true
tlsTrustStoreType=JKS
tlsTrustStorePath=/var/private/tls/client.truststore.jks
tlsTrustStorePassword=clientpw

用于 Java 客户端

import org.apache.pulsar.client.api.PulsarClient;
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar+ssl://broker.example.com:6651/")
    .enableTls(true)
    .useKeyStoreTls(true)
    .tlsTrustStorePath("/var/private/tls/client.truststore.jks")
    .tlsTrustStorePassword("clientpw")
    .allowTlsInsecureConnection(false)
    .build();

for java admin client

This similar to TLS authentication with PEM type

broker authentication config

broker.conf

# Configuration to enable authentication
authenticationEnabled=true
authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls
# this should be the CN for one of client keystore.
superUserRoles=admin
# Enable KeyStore type
tlsEnabledWithKeyStore=true
requireTrustedClientCertOnConnect=true
# key store
tlsKeyStoreType=JKS
tlsKeyStore=/var/private/tls/broker.keystore.jks
tlsKeyStorePassword=brokerpw
# trust store
tlsTrustStoreType=JKS
tlsTrustStore=/var/private/tls/broker.truststore.jks
tlsTrustStorePassword=brokerpw
brokerClientTlsEnabled=true
brokerClientTlsEnabledWithKeyStore=true
brokerClientTlsTrustStoreType=JKS
brokerClientTlsTrustStore=/var/private/tls/client.truststore.jks
brokerClientTlsTrustStorePassword=clientpw
# internal auth config
brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls
brokerClientAuthenticationParameters={"keyStoreType":"JKS","keyStorePath":"/var/private/tls/client.keystore.jks","keyStorePassword":"clientpw"}
# currently websocket not support keystore type
webSocketServiceEnabled=false

Besides the TLS encryption configuring. The main work is configuring the KeyStore, which contains a valid CN as client role, for client.

例如：

对于命令行工具，例如、pulsar-perf 和 [pulsar-client](reference-cli-tools#pulsar- client) 在 Pulsar 安装中使用 conf/client.conf 配置文件。

webServiceUrl=https://broker.example.com:8443/
brokerServiceUrl=pulsar+ssl://broker.example.com:6651/
useKeyStoreTls=true
tlsTrustStoreType=JKS
tlsTrustStorePath=/var/private/tls/client.truststore.jks
tlsTrustStorePassword=clientpw
authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls
authParams={"keyStoreType":"JKS","keyStorePath":"/path/to/keystorefile","keyStorePassword":"keystorepw"}

用于 Java 客户端

import org.apache.pulsar.client.api.PulsarClient;
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar+ssl://broker.example.com:6651/")
    .enableTls(true)
    .useKeyStoreTls(true)
    .tlsTrustStorePath("/var/private/tls/client.truststore.jks")
    .tlsTrustStorePassword("clientpw")
    .allowTlsInsecureConnection(false)
    .authentication(
            "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls",
            "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw")
    .build();

for java admin client

    PulsarAdmin amdin = PulsarAdmin.builder().serviceHttpUrl("https://broker.example.com:8443")
        .useKeyStoreTls(true)
        .tlsTrustStorePath("/var/private/tls/client.truststore.jks")
        .tlsTrustStorePassword("clientpw")
        .allowTlsInsecureConnection(false)
        .authentication(
               "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls",
               "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw")

You can find more details on this in on debugging SSL/TLS connections.