7.2. 为 OpenShift 外部的客户端设置访问权限

1. 为 Kafka 集群配置外部监听程序，启用 TLS 加密和身份验证，并且 Kafka 简单授权。
2. 为客户端创建 KafkaUser，其使用为 简单授权 定义的 TLS 身份验证和访问控制列表(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:
             preferredNodePortAddressType: InternalDNS 

     7

     
             bootstrap and broker service overrides 

     8

     
             #...
         authorization: 

     9

     
           type: simple
           superUsers:
             - super-user-name 

     10

     
       # ...

     1

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

     2

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

     3

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

     4

     将外部监听程序类型指定为 路由、loadbalancer、nodeport 或 ingress。内部侦听器指定为 内部。

     5

     在监听器上启用 TLS 加密。默认为 false。路由 监听器不需要 TLS 加密。

     6

     指定为 tls 的身份验证。

     7

     （可选，仅适用于 nodeport 监听程序）配置 ，为 AMQ Streams 使用的第一个地址类型指定为节点地址。

     8

     （可选）AMQ Streams 会自动决定向客户端公告的地址。地址由 OpenShift 自动分配。如果您正在运行 AMQ Streams 的基础架构没有提供正确的地址，则可以覆盖 bootstrap 和代理服务 地址。验证在覆盖中不会执行。覆盖配置根据监听器类型而有所不同。例如，您可以覆盖 路由、DNS 名称或 IP 地址的主机以用于 负载均衡器，以及节点端口的 节点端口。

     9

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

     10

     （可选）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 集群使用 TLS 身份验证配置 Kafka 代理监听程序。

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

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

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

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

   注意

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

3. 从 Kafka 资源的状态查找 bootstrap 地址和端口。

   oc get kafka KAFKA-CLUSTER-NAME -o jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}'

   使用 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
             operation: Read
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operation: Describe
           - resource:
               type: group
               name: my-group
               patternType: literal
             operation: Read

     1

     标签必须与 Kafka 集群的标签匹配，才能创建用户。

     2

     指定为 tls 的身份验证。

     3

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

5. 创建或修改 KafkaUser 资源。

   oc apply -f USER-CONFIG-FILE

   创建用户，以及与 KafkaUser 资源名称相同的 Secret。Secret 包含 TLS 客户端身份验证的私钥和公钥。

   例如：

   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-OF-THE-CLIENT-CA
     user.crt: USER-CERTIFICATE-CONTAINING-PUBLIC-KEY-OF-USER
     user.key: PRIVATE-KEY-OF-USER
     user.p12: P12-ARCHIVE-FILE-STORING-CERTIFICATES-AND-KEYS
     user.password: PASSWORD-PROTECTING-P12-ARCHIVE

6. 将公共集群 CA 证书提取到所需证书格式：

   oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

7. 从密码文件中提取密码：

   oc get secret KAFKA-CLUSTER-NAME-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d > ca.password

8. 使用公共集群证书的身份验证详情配置客户端：

   客户端代码示例

   properties.put("security.protocol","SSL"); 

   1

   
   properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,"/path/to/ca.p12"); 

   2

   
   properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,CA-PASSWORD); 

   3

   
   properties.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG,"PKCS12"); 

   4

   1

   启用 TLS 加密（使用或不使用 TLS 客户端身份验证）。

   2

   指定导入证书的 truststore 位置。

   3

   指定用于访问 truststore 的密码。如果 truststore 不需要，可以省略此属性。

   4

   标识 truststore 类型。

   注意

   在通过 TLS 使用 SCRAM-SHA 验证时，使用 security.protocol: SASL_SSL。

9. 将用户 CA 证书从用户 Secret 提取到所需证书格式：

   oc get secret USER-NAME -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

10. 从密码文件中提取密码：

    oc get secret USER-NAME -o jsonpath='{.data.user\.password}' | base64 -d > user.password

11. 使用用户 CA 证书的验证详情配置您的客户端：

    客户端代码示例

    properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,"/path/to/user.p12"); 

    1

    
    properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,"<user.password>"); 

    2

    
    properties.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG,"PKCS12"); 

    3

    1

    指定导入证书的密钥存储位置。

    2

    指定用于访问密钥存储的密码。如果密钥存储不需要，可以省略此属性。公共用户证书在创建时由客户端 CA 签名。

    3

    标识密钥存储类型。

12. 添加用于连接 Kafka 集群的 bootstrap 地址和端口：

    bootstrap.servers: BOOTSTRAP-ADDRESS:PORT