7.2. 使用监听程序设置对 Kafka 集群的客户端访问

1. 为 Kafka 集群配置外部监听程序，启用 TLS 加密和 mTLS 身份验证，以及 Kafka 简单 授权。
2. 为客户端创建 KafkaUser，使用 mTLS 验证和为 简单 授权定义的访问控制列表(ACL)创建。

流程

1. 使用 Kafka 侦听器配置 Kafka 集群。

   • 定义通过监听程序访问 Kafka 代理所需的身份验证。

   • 在 Kafka 代理上启用授权。

     监听程序配置示例

     apiVersion: kafka.strimzi.io/v1beta2
     kind: Kafka
     metadata:
       name: my-cluster
       namespace: myproject
     spec:
       kafka:
         # ...
         listeners: 

     1

     
         - name: external 

     2

     
           port: 9094 

     3

     
           type: <listener_type> 

     4

     
           tls: true 

     5

     
           authentication:
             type: tls 

     6

     
           configuration: 

     7

     
             #...
         authorization: 

     8

     
           type: simple
           superUsers:
             - super-user-name 

     9

     
       # ...

     1

     有关启用外部监听器的配置选项，请参考 Generic Kafka 侦听器 schema。

     2

     用于标识监听程序的名称。必须在 Kafka 集群中唯一。

     3

     Kafka 内监听器使用的端口号。端口号必须在给定的 Kafka 集群中唯一。允许的端口号为 9092 及更高版本，除了端口 9404 和 9999 除外，后者已用于 Prometheus 和 JMX。根据监听程序类型，端口号可能与连接 Kafka 客户端的端口号不同。

     4

     将外部监听程序类型指定为 路由、loadbalancer、nodeport 或 ingress。内部监听程序被指定为 internal 或 cluster-ip。

     5

     必需。监听器上的 TLS 加密。对于 route 和 ingress 类型监听程序，它必须设置为 true。对于 mTLS 身份验证，也使用 authentication 属性。

     6

     侦听器上的客户端身份验证机制。对于使用 mTLS 的服务器和客户端身份验证，请指定 tls: true 和 authentication.type: tls。

     7

     （可选）根据监听程序类型的要求，您可以指定额外的 监听程序配置。

     8

     授权指定为 simple，它使用 AclAuthorizer Kafka 插件。

     9

     （可选）Super 用户可以访问所有代理，无论 ACL 中定义的任何访问限制如何。

     警告

     OpenShift Route 地址由 Kafka 集群的名称、侦听器的名称以及在其中创建的命名空间的名称组成。例如，my-cluster-kafka-listener1-bootstrap-myproject (CLUSTER-NAME-kafka-LISTENER-NAME-bootstrap-NAMESPACE)。如果您使用 路由 监听程序类型，请注意地址的整个长度不超过 63 个字符的最大值。

2. 创建或更新 Kafka 资源。

   oc apply -f <kafka_configuration_file>

   Kafka 集群使用 mTLS 身份验证配置有 Kafka 代理监听程序。

   为每个 Kafka 代理 pod 创建服务。

   创建服务作为连接到 Kafka 集群的 bootstrap 地址。

   一个服务也作为外部 bootstrap 地址为到 Kafka 集群的外部连接创建，使用 nodeport 监听器。

   在 secret <cluster _name> -cluster-ca-cert 中也创建了用于验证 kafka 代理身份的集群 CA 证书。

   注意

   如果您在使用外部监听程序时扩展 Kafka 集群，可能会触发所有 Kafka 代理的滚动更新。这取决于配置。

3. 检索您用来从 Kafka 资源状态访问 Kafka 集群的 bootstrap 地址。

   oc get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'

   例如：

   oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'

   使用 Kafka 客户端中的 bootstrap 地址连接到 Kafka 集群。

4. 创建或修改代表需要访问 Kafka 集群的客户端的用户。

   • 指定与 Kafka 监听器相同的验证类型。

   • 指定用于 简单 授权的授权 ACL。

     用户配置示例

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaUser
     metadata:
       name: my-user
       labels:
         strimzi.io/cluster: my-cluster 

     1

     
     spec:
       authentication:
         type: tls 

     2

     
       authorization:
         type: simple
         acls: 

     3

     
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operations:
               - Describe
               - Read
           - resource:
               type: group
               name: my-group
               patternType: literal
             operations:
               - Read

     1

     标签必须与 Kafka 集群的标签匹配。

     2

     将身份验证指定为 mutual tls。

     3

     简单授权需要包含的 ACL 规则列表应用到用户。规则根据 用户名 (my-user)定义在 Kafka 资源上允许的操作。

5. 创建或修改 KafkaUser 资源。

   oc apply -f USER-CONFIG-FILE

   创建用户，以及名称与 KafkaUser 资源的名称相同的 secret。secret 包含用于 mTLS 验证的公钥和私钥。

   secret 示例

   apiVersion: v1
   kind: Secret
   metadata:
     name: my-user
     labels:
       strimzi.io/kind: KafkaUser
       strimzi.io/cluster: my-cluster
   type: Opaque
   data:
     ca.crt: <public_key> # Public key of the clients CA
     user.crt: <user_certificate> # Public key of the user
     user.key: <user_private_key> # Private key of the user
     user.p12: <store> # PKCS #12 store for user certificates and keys
     user.password: <password_for_store> # Protects the PKCS #12 store

6. 从 Kafka 集群的 < cluster_name>-cluster-ca-cert secret 中提取集群 CA 证书。

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

7. 从 < user_name&gt; secret 中提取用户 CA 证书。

   oc get secret <user_name> -o jsonpath='{.data.user\.crt}' | base64 -d > user.crt

8. 从 < user_name&gt; secret 中提取用户的私钥。

   oc get secret <user_name> -o jsonpath='{.data.user\.key}' | base64 -d > user.key

9. 使用用于连接 Kafka 集群的 bootstrap 地址主机名和端口配置客户端：

   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<hostname>:<port>");

10. 使用 truststore 凭证配置客户端，以验证 Kafka 集群的身份。

    指定公共集群 CA 证书。

    信任存储配置示例

    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
    props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
    props.put(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "<ca.crt_file_content>");

    SSL 是用于 mTLS 验证的指定安全协议。为 TLS 的 SCRAM-SHA-512 身份验证指定 SASL_SSL。PEM 是 truststore 的文件格式。

11. 使用密钥存储凭证配置您的客户端，以在连接到 Kafka 集群时验证用户。

    指定公共证书和私钥。

    密钥存储配置示例

    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
    props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
    props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "<user.crt_file_content>");
    props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "<user.key_file_content>");

    将密钥存储证书和私钥直接添加到配置中。将 作为单行格式添加。在 BEGIN CERTIFICATE 和 END CERTIFICATE 分隔符之间，以换行符(\n)开头。结束原始证书中的每一行也带有 \n。

    密钥存储配置示例

    props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "-----BEGIN CERTIFICATE----- \n<user_certificate_content_line_1>\n<user_certificate_content_line_n>\n-----END CERTIFICATE---");
    props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "----BEGIN PRIVATE KEY-----\n<user_key_content_line_1>\n<user_key_content_line_n>\n-----END PRIVATE KEY-----");