15.2. Kafka 客户端的安全选项

使用 KafkaUser 资源为 Kafka 客户端配置身份验证机制、授权机制和访问权限。在配置安全性方面，客户端以用户表示。

您可以验证和授权用户对 Kafka 代理的访问。身份验证允许访问，授权限制访问允许的操作。

您还可以创建对 Kafka 代理没有限制的访问权限的 超级用户。

有关配置 KafkaUser 资源以安全地访问 Kafka 代理的更多信息，请参阅第 14.4 节 “使用监听程序设置对 Kafka 集群的客户端访问”。

15.2.1. 为用户处理识别 Kafka 集群

KafkaUser 资源包含一个标签，用于定义相应 Kafka 集群的名称（从其所属的 Kafka 资源的名称派生）。

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

User Operator 使用该标签来标识 KafkaUser 资源并创建新用户，并在以后的用户处理中。

如果标签与 Kafka 集群不匹配，则 User Operator 无法识别 KafkaUser，且不会创建用户。

如果 KafkaUser 资源的状态保持为空，请检查您的标签。

15.2.2. 用户身份验证

使用 KafkaUser 自定义资源为需要访问 Kafka 集群的用户（客户端）配置身份验证凭据。使用 KafkaUser.spec 中的 authentication 属性配置凭证。通过指定 类型，您可以控制生成的凭证。

支持的验证类型：

用于 mTLS 验证的 TLS
使用外部证书的 mTLS 验证的 tls-external
用于 SCRAM -SHA-512 验证的 SCRAM-sha -512

如果指定了 tls 或 scram-sha-512，则 User Operator 会在创建用户时创建身份验证凭据。如果指定了 tls-external，用户仍然使用 mTLS，但不会创建身份验证凭证。当您提供自己的证书时，使用这个选项。如果没有指定身份验证类型，User Operator 不会创建用户或其凭证。

您可以使用 tls-external 使用 User Operator 外部发布的证书与 mTLS 进行身份验证。User Operator 不生成 TLS 证书或 secret。您仍然可以通过 User Operator 管理 ACL 规则和配额，其方式与使用 tls 机制时相同。这意味着，您可以在指定 ACL 规则和配额时使用 CN=USER-NAME 格式。USER-NAME 是 TLS 证书中提供的通用名称。

15.2.2.1. mTLS 身份验证

要使用 mTLS 身份验证，您可以将 KafkaUser 资源中的 type 字段设置为 tls。

启用 mTLS 验证的用户示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  # ...

身份验证类型必须与用于访问 Kafka 集群的 Kafka 侦听器等效的配置匹配。

当用户由 User Operator 创建时，它会创建一个与 KafkaUser 资源名称相同的新 secret。secret 包含 mTLS 的私钥和公钥。公钥包含在用户证书中，该证书由创建时由客户端 CA （证书颁发机构）签名。所有密钥都是 X.509 格式。

注意

如果您使用 Cluster Operator 生成的客户端 CA，则当 Cluster Operator 更新客户端 CA 时，由 User Operator 生成的用户证书也会更新。

用户 secret 以 PEM 和 PKCS #12 的形式提供密钥和证书。

使用用户凭证的 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

配置客户端时，您可以指定以下内容：

公共集群 CA 证书的 Truststore 属性用于验证 Kafka 集群的身份
用户身份验证凭证的 Keystore 属性用于验证客户端

配置取决于文件格式 (PEM 或 PKCS #12)。这个示例使用 PKCS #12 格式存储，以及访问存储中凭证所需的密码。

以 PKCS #12 格式使用 mTLS 的客户端配置示例

bootstrap.servers=<kafka_cluster_name>-kafka-bootstrap:9093 1
security.protocol=SSL 2
ssl.truststore.location=/tmp/ca.p12 3
ssl.truststore.password=<truststore_password> 4
ssl.keystore.location=/tmp/user.p12 5
ssl.keystore.password=<keystore_password> 6

1: 连接到 Kafka 集群的 bootstrap 服务器地址。
2: 使用 TLS 加密时的安全协议选项。
3: truststore 位置包含了 Kafka 集群的公钥证书 (ca.p12)。在创建 Kafka 集群时，Cluster Operator 在 < cluster_name>-cluster-ca-cert secret 中生成集群 CA 证书和密钥。
4: 用于访问信任存储的密码(ca.password)。
5: 密钥存储位置包含 Kafka 用户的公钥证书(user.p12)。
6: 用于访问密钥存储的密码(user.password)。

15.2.2.2. 使用 User Operator 外部发布的证书进行 mTLS 验证

要使用 User Operator 外部发布的证书使用 mTLS 身份验证，您可以将 KafkaUser 资源中的 type 字段设置为 tls-external。不会为用户创建 secret 和凭证。

使用 User Operator 外部发布的带有 mTLS 验证的用户示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls-external
  # ...

15.2.2.3. SCRAM-SHA-512 身份验证

要使用 SCRAM-SHA-512 身份验证机制，您可以将 KafkaUser 资源中的 type 字段设置为 scram-sha-512。

启用 SCRAM-SHA-512 验证的用户示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  # ...

当用户由 User Operator 创建时，它会创建一个与 KafkaUser 资源名称相同的新 secret。secret 包含在 password 键中生成的密码，该密钥使用 base64 编码。要使用密码，必须对其进行解码。

使用用户凭证的 secret 示例

apiVersion: v1
kind: Secret
metadata:
  name: my-user
  labels:
    strimzi.io/kind: KafkaUser
    strimzi.io/cluster: my-cluster
type: Opaque
data:
  password: Z2VuZXJhdGVkcGFzc3dvcmQ= 1
  sasl.jaas.config: b3JnLmFwYWNoZS5rYWZrYS5jb21tb24uc2VjdXJpdHkuc2NyYW0uU2NyYW1Mb2dpbk1vZHVsZSByZXF1aXJlZCB1c2VybmFtZT0ibXktdXNlciIgcGFzc3dvcmQ9ImdlbmVyYXRlZHBhc3N3b3JkIjsK 2

1: 生成的密码 base64 编码。
2: SASL SCRAM-SHA-512 身份验证的 JAAS 配置字符串，采用 base64 编码。

解码生成的密码：

echo "Z2VuZXJhdGVkcGFzc3dvcmQ=" | base64 --decode

15.2.2.3.1. 自定义密码配置

创建用户时，Apache Kafka 的 Streams 会生成一个随机密码。您可以使用自己的密码而不是 Streams for Apache Kafka 生成的密码。要做到这一点，使用密码创建一个 secret，并在 KafkaUser 资源中引用它。

为 SCRAM-SHA-512 验证设置了密码的用户示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
    password:
      valueFrom:
        secretKeyRef:
          name: my-secret 1
          key: my-password 2
  # ...

1: 包含预定义密码的 secret 名称。
2: 存储在机密中的密码的密钥。

15.2.3. 用户授权

使用 KafkaUser 自定义资源为需要访问 Kafka 集群的用户（客户端）配置授权规则。使用 KafkaUser.spec 中的 authorization 属性配置规则。通过指定 类型，您可以控制使用哪些规则。

要使用简单授权，您可以在 KafkaUser.spec.authorization 中将 type 属性设置为 simple。简单的授权使用 Kafka Admin API 管理 Kafka 集群中的 ACL 规则。是否启用 User Operator 中的 ACL 管理，是否取决于您的 Kafka 集群中的授权配置。

对于简单的授权，始终启用 ACL 管理。
对于 OPA 授权，始终禁用 ACL 管理。授权规则在 OPA 服务器中配置。
对于 Red Hat Single Sign-On 授权，您可以在 Red Hat Single Sign-On 中直接管理 ACL 规则。您还可以在配置中将授权委托给简单授权器作为回退选项。启用到简单授权器时，User Operator 也会启用 ACL 规则的管理。
要使用自定义授权插件进行自定义授权，请使用 Kafka 自定义资源的 .spec.kafka.authorization 配置中的 supportsAdminApi 属性来启用或禁用支持。

授权是集群范围的。授权类型必须与 Kafka 自定义资源中的等效配置匹配。

如果没有启用 ACL 管理，如果包含任何 ACL 规则，则 Apache Kafka 的流会拒绝资源。

如果您使用 User Operator 的独立部署，则默认启用 ACL 管理。您可以使用 STRIMZI_ACLS_ADMIN_API_SUPPORTED 环境变量禁用它。

如果没有指定授权，User Operator 不会为用户置备任何访问权限。此类 KafkaUser 仍然可以访问资源取决于所使用的授权器。例如，对于 简单 授权，这由 Kafka 集群中的 allow.everyone.if.no.acl.found 配置决定。

15.2.3.1. ACL 规则

简单 授权使用 ACL 规则来管理 Kafka 代理的访问。

ACL 规则授予您在 acls 属性中指定的用户访问权限。

有关 AclRule 对象的更多信息，请参阅 AclRule 模式参考。

15.2.3.2. 超级用户对 Kafka 代理的访问

如果用户被添加到 Kafka 代理配置中的超级用户列表中，无论 KafkaUser 中的 ACL 中定义的任何授权限制是什么，用户都可以无限访问集群。

有关配置超级用户访问代理的更多信息，请参阅 Kafka 授权。

15.2.3.3. 用户配额

您可以为 KafkaUser 资源配置 spec 以强制配额，以便用户不会超过配置对 Kafka 代理的访问级别。您可以设置基于大小的网络使用量和基于时间的 CPU 使用率阈值。您还可以添加分区修改配额来控制更改分区请求的速率。

带有用户配额的 KafkaUser 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  quotas:
    producerByteRate: 1048576 1
    consumerByteRate: 2097152 2
    requestPercentage: 55 3
    controllerMutationRate: 10 4

1: 用户可推送到 Kafka 代理的数据量的字节每秒配额
2: 用户可以从 Kafka 代理获取的数据量上的字节每秒配额
3: CPU 使用率限制作为客户端组的时间百分比
4: 每秒允许的并发分区创建和删除操作数

有关这些属性的更多信息，请参阅 KafkaUserQuotas 模式参考。