5.2. 为安全访问设置客户端

在 Kafka 代理上设置监听程序来支持安全连接后，下一步是将客户端应用程序配置为使用这些监听程序与 Kafka 集群通信。这包括为每个客户端提供适当的安全设置，以便根据监听器上配置的安全机制与集群进行身份验证。

5.2.1. 配置安全协议
复制链接

配置客户端应用程序使用的安全协议，以匹配 Kafka 代理监听程序中配置的协议。例如，对 TLS 身份验证使用 SSL （安全套接字层），或使用 TLS 加密对 SASL (Simple Authentication and Security Layer)验证 SASL_SSL。在您的客户端配置中添加 truststore 和 keystore，它支持访问 Kafka 集群所需的身份验证机制。

truststore: truststore 包含用于验证 Kafka 代理真实性的可信证书颁发机构(CA)的公共证书。当客户端连接到安全 Kafka 代理时，可能需要验证代理的身份。
keystore: 密钥存储包含客户端的私钥及其公钥。当客户端希望向代理验证其自身时，它会显示自己的证书。

如果使用 TLS 身份验证，您的 Kafka 客户端配置需要一个信任存储和密钥存储来连接到 Kafka 集群。如果您使用 SASL SCRAM-SHA-512，则通过交换用户名和密码凭证而不是数字证书来执行身份验证，因此不需要密钥存储。SCRAM-SHA-512 是一种更加轻量级的机制，但它不像使用基于证书的身份验证一样安全。

注意

如果您有自己的证书基础架构并使用来自第三方 CA 的证书，那么客户端的默认信任存储可能已包含公共 CA 证书，您不需要将它们添加到客户端的信任存储中。如果服务器证书由已包含在默认信任存储中的公共 CA 证书之一签名，客户端会自动信任它。

您可以创建一个 config.properties 文件来指定客户端应用程序使用的身份验证凭证。

在以下示例中，security.protocol 设置为 SSL，以启用客户端和代理之间的 TLS 身份验证和加密。

ssl.truststore.location 和 ssl.truststore.password 属性指定信任存储的位置和密码。ssl.keystore.location 和 ssl.keystore.password 属性指定密钥存储的位置和密码。

使用 PKCS thePublic-Key Cryptography Standards 文件格式。您还可以使用 base64 编码的 PEM (Privacy Enhanced Mail)格式。

TLS 身份验证的客户端配置属性示例

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SSL
ssl.truststore.location = /path/to/ca.p12
ssl.truststore.password = truststore-password
ssl.keystore.location = /path/to/user.p12
ssl.keystore.password = keystore-password
client.id = my-client

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SSL
ssl.truststore.location = /path/to/ca.p12
ssl.truststore.password = truststore-password
ssl.keystore.location = /path/to/user.p12
ssl.keystore.password = keystore-password
client.id = my-client

Copy to Clipboard

Toggle word wrap

在以下示例中，security.protocol 设置为 SASL_SSL，以便在客户端和服务器之间使用 TLS 加密启用 SASL 身份验证。如果您只需要身份验证而不是加密，您可以使用 SASL 协议。进行身份验证的 SASL 机制是 SCRAM-SHA-512。可以使用不同的身份验证机制。sasl.jaas.config 属性指定身份验证凭据。

SCRAM-SHA-512 身份验证的客户端配置示例

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
ssl.truststore.location = path/to/truststore.p12
ssl.truststore.password = truststore_password
ssl.truststore.type = PKCS12
client.id = my-client

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
ssl.truststore.location = path/to/truststore.p12
ssl.truststore.password = truststore_password
ssl.truststore.type = PKCS12
client.id = my-client

Copy to Clipboard

Toggle word wrap

注意

对于不支持 PEM 格式的应用程序，您可以使用 OpenSSL 等工具将 PEM 文件转换为 PKCS the 格式。

5.2.2. 配置允许的 TLS 版本和密码套件
复制链接

您可以组合 SSL 配置和密码套件，以进一步保护客户端应用程序和 Kafka 集群之间的基于 TLS 的通信。在 Kafka 代理配置中指定支持的 TLS 版本和密码套件。如果要限制它们使用的 TLS 版本和密码套件，您还可以将配置添加到客户端。客户端上的配置应该只使用代理上启用的协议和密码套件。

在以下示例中，SSL 使用 security.protocol 进行 Kafka 代理和客户端应用程序之间的通信。您可以将密码套件指定为用逗号分开的列表。ssl.cipher.suites 属性是客户端允许使用的密码套件的逗号分隔列表。

Kafka 代理的 SSL 配置属性示例

security.protocol: "SSL"
ssl.enabled.protocols: "TLSv1.3", "TLSv1.2"
ssl.protocol: "TLSv1.3"
ssl.cipher.suites: "TLS_AES_256_GCM_SHA384"

security.protocol: "SSL"
ssl.enabled.protocols: "TLSv1.3", "TLSv1.2"
ssl.protocol: "TLSv1.3"
ssl.cipher.suites: "TLS_AES_256_GCM_SHA384"

Copy to Clipboard

Toggle word wrap

ssl.enabled.protocols 属性指定可用于集群及其客户端之间安全通信的可用 TLS 版本。在本例中，TLSv1.3 和 TLSv1.2 都已启用。ssl.protocol 属性为所有连接设置默认 TLS 版本，且必须从启用的协议中选择它。默认情况下，客户端使用 TLSv1.3 进行通信。如果客户端只支持 TLSv1.2，它仍然可以连接到代理并使用该支持的版本进行通信。同样，如果配置位于客户端，且代理只支持 TLSv1.2，客户端将使用支持的版本。

Apache Kafka 支持的密码套件取决于您使用的 Kafka 版本和底层环境。检查提供最高安全性级别的最新支持的密码套件。

5.2.3. 使用访问控制列表(ACL)
复制链接

您不必为客户端应用程序中的 ACLS 明确配置任何内容。ACL 由 Kafka 代理在服务器端强制执行。当客户端向服务器发送请求来生成或消耗数据时，服务器会检查 ACL，以确定客户端（用户）是否被授权执行所请求的操作。如果客户端被授权，则会处理请求；否则请求将被拒绝，并返回错误。但是，客户端仍必须经过身份验证，并使用适当的安全协议启用与 Kafka 集群的安全连接。

如果您在 Kafka 代理上使用访问控制列表(ACL)，请确保正确设置了 ACL，以限制对您要控制的主题和操作的访问。如果您使用 Open Policy Agent (OPA)策略来管理访问，在策略中配置授权规则，因此您不需要为 Kafka 代理指定 ACL。OAuth 2.0 提供了一些灵活性：您可以使用 OAuth 2.0 供应商来管理 ACL；或使用 OAuth 2.0 和 Kafka 的简单 授权来管理 ACL。

注意

ACL 适用于大多数类型的请求，不仅限于生成和消耗操作。例如，可将 ACLS 应用到读取操作，如描述主题或写入操作，如创建新主题。

5.2.4. 使用 OAuth 2.0 进行基于令牌的访问
复制链接

使用 OAuth 2.0 开放标准进行 AMQ Streams 授权，通过 OAuth 2.0 供应商强制控制授权。OAuth 2.0 为应用提供了一种安全的方法来访问存储在其他系统中的用户数据。授权服务器可能会向客户端应用程序发出访问令牌，以授予 Kafka 集群的访问权限。

以下步骤描述了设置和使用 OAuth 2.0 进行令牌验证的一般方法：

使用代理和客户端凭证（如客户端 ID 和 secret）配置授权服务器。
从授权服务器获取 OAuth 2.0 凭据。
在 Kafka 代理中使用 OAuth 2.0 凭证配置监听程序，并与授权服务器交互。
将 Oauth 2.0 依赖项添加到客户端库。
使用 OAuth 2.0 凭证配置 Kafka 客户端，并与授权服务器交互。
在运行时获取访问令牌，该令牌使用 OAuth 2.0 供应商验证客户端。

如果您在 Kafka 代理上配置了 OAuth 2.0 的监听程序，您可以将客户端应用程序设置为使用 OAuth 2.0。除了用于访问 Kafka 集群的标准 Kafka 客户端配置外，还必须为 OAuth 2.0 身份验证包含特定的配置。您还必须确保 Kafka 集群和客户端应用程序可以访问您要使用的授权服务器。

指定 SASL （简单身份验证和安全层）安全协议和机制。在生产环境中，推荐进行以下设置：

TLS 加密连接的 SASL_SSL 协议。
用于使用 bearer 令牌的凭据交换的 OAUTHBEARER 机制

JAAS (Java 身份验证和授权服务)模块实施 SASL 机制。机制的配置取决于您使用的身份验证方法。例如，使用凭证交换您添加 OAuth 2.0 访问令牌端点、访问令牌、客户端 ID 和客户端 secret。客户端连接到授权服务器的令牌端点(URL)，以检查令牌是否仍然有效。您还需要一个 truststore，其中包含授权服务器的公钥证书，以便进行身份验证。

OAauth 2.0 的客户端配置属性示例

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = OAUTHBEARER
# ...
sasl.jaas.config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    oauth.token.endpoint.uri = "https://localhost:9443/oauth2/token" \
    oauth.access.token = <access_token> \
    oauth.client.id = "<client_id>" \
    oauth.client.secret = "<client_secret>" \
    oauth.ssl.truststore.location = "/<truststore_location>/oauth-truststore.p12" \
    oauth.ssl.truststore.password = "<truststore_password>" \
    oauth.ssl.truststore.type = "PKCS12" \

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = OAUTHBEARER
# ...
sasl.jaas.config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    oauth.token.endpoint.uri = "https://localhost:9443/oauth2/token" \
    oauth.access.token = <access_token> \
    oauth.client.id = "<client_id>" \
    oauth.client.secret = "<client_secret>" \
    oauth.ssl.truststore.location = "/<truststore_location>/oauth-truststore.p12" \
    oauth.ssl.truststore.password = "<truststore_password>" \
    oauth.ssl.truststore.type = "PKCS12" \

Copy to Clipboard

Toggle word wrap

有关将代理设置为使用 OAuth 2.0 的更多信息，请参阅以下指南：

5.2.5. 使用 Open Policy Agent (OPA)访问策略
复制链接

使用带有 AMQ Streams 的 Open Policy Agent (OPA)策略控制器来评估您的 Kafka 集群的请求，具体取决于访问策略。开放策略代理(OPA)是一种策略引擎，用于管理授权策略。策略集中访问控制，可以动态更新，无需更改客户端应用程序。例如，您可以创建一个策略，仅允许某些用户（客户端）为特定主题生成和使用消息。

AMQ Streams 为 Kafka 授权使用 Open Policy Agent 插件作为授权器。

以下步骤描述了设置和使用 OPA 的一般方法：

设置 OPA 服务器的实例。
定义提供管理对 Kafka 集群访问权限的授权规则的策略。
为 Kafka 代理创建配置，以接受 OPA 授权并与 OPA 服务器交互。
配置 Kafka 客户端，以提供授权访问 Kafka 集群的凭证。

如果您在 Kafka 代理上配置了 OPA 的监听程序，您可以将客户端应用程序设置为使用 OPA。在监听器配置中，您可以指定一个 URL 来连接到 OPA 服务器并授权您的客户端应用程序。除了访问 Kafka 集群的标准 Kafka 客户端配置外，还必须添加凭证以使用 Kafka 代理进行身份验证。代理通过向 OPA 服务器发送请求来评估授权策略，以检查客户端是否有必要的授权来执行请求操作。您不需要信任存储或密钥存储来保护通信，因为策略引擎强制执行授权策略。

OPA 授权的客户端配置属性示例

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
# ...

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
# ...

Copy to Clipboard

Toggle word wrap

注意

红帽不支持 OPA 服务器。

有关将代理设置为使用 OPA 的更多信息，请参阅以下指南：

5.2.6. 在流消息时使用事务
复制链接

通过在代理和制作者客户端应用程序中配置事务属性，您可以确保在单个事务中处理消息。事务向消息流增加可靠性和一致性。

代理始终启用事务。您可以使用以下属性更改默认配置：

事务的 Kafka 代理配置属性示例

transaction.state.log.replication.factor = 3
transaction.state.log.min.isr = 2
transaction.abort.timed.out.transaction.cleanup.interval.ms = 3600000

transaction.state.log.replication.factor = 3
transaction.state.log.min.isr = 2
transaction.abort.timed.out.transaction.cleanup.interval.ms = 3600000

Copy to Clipboard

Toggle word wrap

这是生产环境的典型配置，它为内部 __transaction_state 主题创建 3 个副本。\__transaction_state 主题存储有关事务在进行中的信息。事务日志至少需要 2 个同步副本。清理间隔是检查超时事务和清理对应的事务日志之间的时间。

要为客户端配置添加事务属性，您需要为生产者和消费者设置以下属性：

事务的制作者客户端配置属性示例

transactional.id = unique-transactional-id
enable.idempotence = true
max.in.flight.requests.per.connection = 5
acks = all
transaction.timeout.ms = 30000
delivery.timeout = 25000

transactional.id = unique-transactional-id
enable.idempotence = true
max.in.flight.requests.per.connection = 5
acks = all
transaction.timeout.ms = 30000
delivery.timeout = 25000

Copy to Clipboard

Toggle word wrap

事务 ID 允许 Kafka 代理跟踪事务。它是制作者的唯一标识符，应当与特定的分区一起使用。如果您需要为多组分区执行事务，则需要为每个集合使用不同的事务 ID。启用 idempotence 以避免生成者实例创建重复消息。使用 idempotence 时，消息使用制作者 ID 和序列号进行跟踪。代理收到消息时，它会检查制作者 ID 和序列号。如果收到具有相同制作者 ID 和序列号的消息，代理会丢弃重复的消息。

最大动态请求数设置为 5，以便以发送的顺序处理事务。分区最多可有 5 个动态请求，而不影响消息排序。

通过将 acks 设为 all，生产者会等待来自要写入的主题分区的所有同步副本的确认，然后再将事务视为完成。这样可确保信息被大量写入（提交）到 Kafka 集群，即使代理失败，也不会丢失它们。

事务超时指定客户端在超时前必须完成事务的最长时间。交付超时指定制作者在超时前等待代理确认消息发送的最长时间。要确保在事务周期内发送消息，请将交付超时设置为小于事务超时。在指定重试次数时，请考虑网络延迟和消息吞吐量，并允许临时故障。

事务的消费者客户端配置属性示例

group.id = my-group-id
isolation.level = read_committed
enable.auto.commit = false

group.id = my-group-id
isolation.level = read_committed
enable.auto.commit = false

Copy to Clipboard

Toggle word wrap

read_committed 隔离级别指定消费者只读取成功完成的事务的消息。消费者不会处理属于持续或失败的事务的任何消息。这样可确保消费者只读取属于完整事务的消息。

当使用事务流消息时，务必要将 enable.auto.commit 设置为 false。如果设置为 true，则消费者会定期提交偏移量，而无需考虑事务。这意味着消费者可以在事务完成前提交消息。通过将 enable.auto.commit 设置为 false，使用者仅读取并提交已完全写入并提交至该主题的消息，作为事务的一部分。

5.2. 为安全访问设置客户端

5.2.1. 配置安全协议
复制链接

5.2.2. 配置允许的 TLS 版本和密码套件
复制链接

5.2.3. 使用访问控制列表(ACL)
复制链接

5.2.4. 使用 OAuth 2.0 进行基于令牌的访问
复制链接

5.2.5. 使用 Open Policy Agent (OPA)访问策略
复制链接

5.2.6. 在流消息时使用事务
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

5.2. 为安全访问设置客户端

5.2.1. 配置安全协议复制链接链接已复制到粘贴板!

5.2.2. 配置允许的 TLS 版本和密码套件复制链接链接已复制到粘贴板!

5.2.3. 使用访问控制列表(ACL)复制链接链接已复制到粘贴板!

5.2.4. 使用 OAuth 2.0 进行基于令牌的访问复制链接链接已复制到粘贴板!

5.2.5. 使用 Open Policy Agent (OPA)访问策略复制链接链接已复制到粘贴板!

5.2.6. 在流消息时使用事务复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

5.2.1. 配置安全协议
复制链接

5.2.2. 配置允许的 TLS 版本和密码套件
复制链接

5.2.3. 使用访问控制列表(ACL)
复制链接

5.2.4. 使用 OAuth 2.0 进行基于令牌的访问
复制链接

5.2.5. 使用 Open Policy Agent (OPA)访问策略
复制链接

5.2.6. 在流消息时使用事务
复制链接