Red Hat Summit 红帽全球峰会支持控制台（Console）开发人员开始试用联系人

16.4. 在 Red Hat build of Keycloak 中设置权限

当使用红帽构建的 Keycloak 作为 OAuth 2.0 授权服务器时，Kafka 权限会使用授权权限为用户帐户或服务帐户授予。要授予访问 Kafka 的权限，请在红帽构建的 Keycloak 中创建一个 OAuth 客户端规格，该规范映射红帽构建的 Keycloak 授权服务和 Kafka 的授权模型。

16.4.1. Kafka 和红帽构建的 Keycloak 授权模型
复制链接

Kafka 和红帽构建的 Keycloak 使用不同的授权模型。

Kafka 授权模型

Kafka 的授权模型使用 资源类型 和操作 来描述用户的 ACL。当 Kafka 客户端在代理上执行操作时，代理使用配置的 KeycloakAuthorizer 根据操作和资源类型检查客户端的权限。

每个资源类型都有一组可用于操作的可用权限。例如，主题 资源类型在别处具有 Create 和 Write 权限。

如需资源和权限的完整列表，请参阅 Kafka 文档中的 Kafka 授权模型。

红帽构建的 Keycloak 授权模型

红帽 Keycloak 的授权服务模型构建有四个概念来定义和授予权限：

  • Resources
  • 范围
  • 策略（policy）
  • 权限

有关这些概念的详情，请参考授权服务指南。https://docs.redhat.com/en/documentation/red_hat_build_of_keycloak/

16.4.2. 映射授权模型
复制链接

Kafka 授权模型用作定义由红帽构建的 Keycloak 角色和资源用来控制对 Kafka 的访问的基础。

要为用户帐户或服务帐户授予 Kafka 权限，您首先为 Kafka 集群在 Red Hat build of Keycloak 中创建 OAuth 客户端规格。然后，您可以指定红帽在客户端上的 Keycloak 授权服务规则构建。通常，代表 Kafka 集群的 OAuth 客户端的客户端 ID 是 kafka。由 Apache Kafka 的 Streams 提供 的示例配置文件 使用 kafka 作为 OAuth 客户端 ID。

注意

如果有多个 Kafka 集群，可以对所有集群使用单个 OAuth 客户端(kafka)。这为您提供了一个统一的空间，在其中定义和管理授权规则。但是，您也可以使用不同的 OAuth 客户端 ID （如 my-cluster-kafka 或 cluster-dev-kafka），并在每个客户端配置中为每个集群定义授权规则。

kafka 客户端定义必须在红帽构建的 Keycloak 管理控制台中启用 Authorization Enabled 选项。有关启用授权服务的详情，请参考授权服务指南。https://docs.redhat.com/en/documentation/red_hat_build_of_keycloak/

kafka 客户端的范围中存在所有权限。如果您使用不同的 OAuth 客户端 ID 配置不同的 Kafka 集群，它们各自需要一组单独的权限，即使它们是同一红帽构建的 Keycloak 域的一部分。

当 Kafka 客户端使用 OAUTHBEARER 身份验证时，红帽构建的 Keycloak 授权器(KeycloakAuthorizer)使用当前会话的访问令牌来检索来自红帽构建的 Keycloak 服务器的授权列表。要授予权限，授权者会根据访问令牌所有者的策略和权限评估红帽构建的 Keycloak 授权服务中的授予列表（接收和缓存）。

为 Kafka 权限上传授权范围

初始的 Keycloak 配置构建通常涉及上传授权范围，以创建可在每个 Kafka 资源类型上执行的所有可能操作的列表。此步骤仅在定义任何权限前执行一次。您可以手动添加授权范围，而不是上传它们。

授权范围应包含以下 Kafka 权限，而不考虑资源类型：

  • 创建
  • 写
  • 读
  • 删除
  • Describe
  • 更改
  • DescribeConfigs
  • AlterConfigs
  • ClusterAction
  • IdempotentWrite

如果您确信不需要权限（例如，IdempotentWrite），您可以从授权范围列表中省略它。但是，这个权限不适用于 Kafka 资源上的目标。

注意

不支持 All 权限。

权限检查的资源模式

在执行权限检查时，资源模式用于与目标资源匹配的模式。常规模式格式为 < resource_type>:<pattern_name>。

资源类型镜像 Kafka 授权模型。模式允许两个匹配选项：

  • 完全匹配（当模式不以 *结尾）
  • 前缀匹配（当模式以 *结尾时）

资源模式示例

Topic:my-topic
Topic:orders-*
Group:orders-*
Cluster:*
Copy to Clipboard Toggle word wrap

另外，常规模式格式可以带有 kafka-cluster:<cluster_name> 前缀，后跟一个逗号，其中 <cluster_name > 代表 Kafka 自定义资源中的 metadata.name。

带有集群前缀的资源模式示例

kafka-cluster:my-cluster,Topic:*
kafka-cluster:*,Group:b_*
Copy to Clipboard Toggle word wrap

当缺少 kafka-cluster 前缀时，它被假定为 kafka-cluster:*。

在定义资源时，您可以将它与与资源相关的授权范围列表相关联。设置目标资源类型有意义的操作。

虽然您可以向任何资源添加任何授权范围，但只有资源类型支持的范围才会考虑访问控制。

应用访问权限的策略

策略用于对一个或多个用户帐户或服务帐户的目标权限。目标可以参考：

  • 特定用户或服务帐户
  • 域角色或客户端角色
  • 用户组

一个策略被授予一个唯一名称，可以重复使用以多个资源为目标的多个权限。

授予访问权限的权限

使用细粒度权限将策略、资源和授权范围一起拉取，以授予用户访问权限。

每个权限的名称应该明确定义它授予哪些用户的权限。例如，Dev Team B 可以从以 x 开始的主题读取。

16.4.3. 通用 Kafka 操作的权限
复制链接

以下示例演示了在 Kafka 上执行常见操作所需的用户权限。

创建主题

要创建主题，需要 Create 权限用于特定主题，或 Cluster:kafka-cluster。 

bin/kafka-topics.sh --create --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

列出主题

如果用户对指定主题具有 Describe 权限，则会列出该主题。 

bin/kafka-topics.sh --list \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

显示主题详情

要显示主题的详情，需要主题的 Describe 和 DescribeConfigs 权限。 

bin/kafka-topics.sh --describe --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

将消息生成到主题

要生成消息到主题，需要主题的 Describe 和 Write 权限。

如果主题还没有创建，并且启用了主题 auto-creation，则需要创建主题的权限。 

bin/kafka-console-producer.sh  --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --producer.config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

使用主题的消息

为了消费主题中的消息，需要有主题的 Describe 和 Read 权限。在主题中使用通常依赖于将消费者偏移存储在消费者组中，这需要对消费者组需要额外的 Describe 和 Read 权限。

需要 两个资源 才能匹配。例如： 

Topic:my-topic
Group:my-group-*
Copy to Clipboard Toggle word wrap 
bin/kafka-console-consumer.sh --topic my-topic --group my-group-1 --from-beginning \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --consumer.config /tmp/config.properties
Copy to Clipboard Toggle word wrap

使用幂等制作者将消息生成到主题

除了生成主题的权限外，Cluster:kafka-cluster 资源还需要额外的 IdempotentWrite 权限。

需要 两个资源 才能匹配。例如： 

Topic:my-topic
Cluster:kafka-cluster
Copy to Clipboard Toggle word wrap 
bin/kafka-console-producer.sh  --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --producer.config=/tmp/config.properties --producer-property enable.idempotence=true --request-required-acks -1
Copy to Clipboard Toggle word wrap

列出消费者组

当列出消费者组时，只有用户具有 Describe 权限的组才会返回。另外，如果用户对 Cluster:kafka-cluster 有 Describe 权限，则返回所有消费者组。 

bin/kafka-consumer-groups.sh --list \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

显示消费者组详情

要显示消费者组的详情，在组以及与组关联的主题上需要 Describe 权限。 

bin/kafka-consumer-groups.sh --describe --group my-group-1 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

更改主题配置

要更改主题的配置，需要主题的 Describe 和 Alter 权限。 

bin/kafka-topics.sh --alter --topic my-topic --partitions 2 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

显示 Kafka 代理配置

要使用 kafka-configs.sh 获取代理配置，Cluster:kafka-cluster 需要 DescribeConfigs 权限。 

bin/kafka-configs.sh --entity-type brokers --entity-name 0 --describe --all \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

更改 Kafka 代理配置

要更改 Kafka 代理的配置，Cluster:kafka-cluster 需要 DescribeConfigs 和 AlterConfigs 权限。 

bin/kafka-configs --entity-type brokers --entity-name 0 --alter --add-config log.cleaner.threads=2 \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

删除主题

要删除主题，主题需要 Describe 和 Delete 权限。 

bin/kafka-topics.sh --delete --topic my-topic \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config=/tmp/config.properties
Copy to Clipboard Toggle word wrap

选择引导分区

要为主题分区运行领导选择，Cluster:kafka-cluster 需要 Alter 权限。 

bin/kafka-leader-election.sh --topic my-topic --partition 0 --election-type PREFERRED  /
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --admin.config /tmp/config.properties
Copy to Clipboard Toggle word wrap

重新分配分区

要生成分区重新分配文件，对涉及的主题需要 Describe 权限。 

bin/kafka-reassign-partitions.sh --topics-to-move-json-file /tmp/topics-to-move.json --broker-list "0,1" --generate \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties > /tmp/partition-reassignment.json
Copy to Clipboard Toggle word wrap

要执行分区重新分配，Cluster:kafka-cluster 需要 Describe 和 Alter 权限。另外，有关涉及的主题还需要 Describe 权限。 

bin/kafka-reassign-partitions.sh --reassignment-json-file /tmp/partition-reassignment.json --execute \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties
Copy to Clipboard Toggle word wrap

对于 Cluster:kafka-cluster 以及涉及的每个主题，需要验证分区重新分配、Describe 和 AlterConfigs 权限。 

bin/kafka-reassign-partitions.sh --reassignment-json-file /tmp/partition-reassignment.json --verify \
  --bootstrap-server my-cluster-kafka-bootstrap:9092 --command-config /tmp/config.properties
Copy to Clipboard Toggle word wrap

如果您在红帽构建的 Keycloak 中使用 OAuth 2.0 进行基于令牌的身份验证，您还可以使用红帽构建的 Keycloak 来配置授权规则，以限制客户端访问 Kafka 代理。本例解释了如何将红帽构建的 Keycloak 授权服务与 keycloak 授权服务一起使用。设置红帽构建的 Keycloak 授权服务，以便在 Kafka 客户端上强制访问限制。Red Hat build of Keycloak Authorization Services 使用授权范围、策略和权限来定义并应用对资源的访问控制。

红帽构建的 Keycloak Authorization Services REST 端点提供经过身份验证的用户授予资源的权限列表。在由 Kafka 客户端建立经过身份验证的会话后，从红帽构建的 Keycloak 服务器获取授权（权限）列表作为第一个操作。该列表在后台刷新，以便检测对授予的更改。为每个用户会话在 Kafka 代理上本地执行授权，以提供快速授权决策。

Apache Kafka 的流提供了 示例配置文件。这包括以下用于设置红帽构建的 Keycloak 的示例文件：

kafka-ephemeral-oauth-single-keycloak-authz.yaml
使用红帽构建的 Keycloak 为基于 OAuth 2.0 令牌的授权配置 Kafka 自定义资源示例。您可以使用自定义资源来部署使用 keycloak 授权和基于令牌的 oauth 身份验证的 Kafka 集群。
kafka-authz-realm.json
红帽构建的 Keycloak 域示例，配置了示例组、用户、角色和客户端。您可以将域导入到红帽构建的 Keycloak 实例中，以设置精细的权限来访问 Kafka。

如果要尝试使用红帽构建的 Keycloak 示例，按照所示的顺序执行本节中介绍的任务。

  1. 在 Red Hat build of Keycloak 中设置权限
  2. 使用红帽构建的 Keycloak 授权部署 Kafka 集群
  3. 为 CLI Kafka 客户端会话准备 TLS 连接
  4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问

身份验证

当您配置基于令牌的 oauth 身份验证时，您可以将 jwksEndpointUri 指定为本地 JWT 验证的 URI。配置 keycloak 授权时，您将 tokenEndpointUri 指定为红帽构建的 Keycloak 令牌端点的 URI。两个 URI 的主机名必须相同。

带有组或角色策略的目标权限

在 Red Hat build of Keycloak 中，启用了服务帐户的机密客户端可以使用客户端 ID 和 secret 在自己的名称中向服务器进行身份验证。这对通常充当自有名称的微服务来说很方便，而不适用于特定用户（如网站）的代理。服务帐户可以像常规用户一样分配角色。但是，不能为其分配组。因此，如果要使用服务帐户将权限目标到微服务，则无法使用组策略，而应使用角色策略。相反，如果您只想将某些权限限制为需要使用用户名和密码进行身份验证的普通用户帐户，您可以将其作为使用组策略而不是角色策略的副作用实现。这是本例中以 ClusterManager 开头的权限。执行集群管理通常是使用 CLI 工具交互完成的。在使用生成的访问令牌向 Kafka 代理进行身份验证前，需要用户登录是合理的。在这种情况下，访问令牌代表特定用户，而不是客户端应用程序。

16.4.4.1. 在 Red Hat build of Keycloak 中设置权限
复制链接

设置红帽构建的 Keycloak，然后连接到其管理控制台并添加预配置的域。使用示例 kafka-authz-realm.json 文件导入域。您可以在管理控制台中检查为域定义的授权规则。规则授予 Kafka 集群上资源的访问权限，以使用 Red Hat build of Keycloak 域示例。

先决条件

  • 一个正常运行的 OpenShift 集群。
  • Apache Kafka 示例/security/keycloak-authorization/kafka-authz-realm.json 文件的 Streams，其中包含预配置的域。

流程

  1. 使用 Red Hat build of Keycloak Operator 安装 Red Hat build of Keycloak 服务器，如 Red Hat build of Keycloak 文档所述。https://docs.redhat.com/en/documentation/red_hat_build_of_keycloak/
  2. 等待红帽构建的 Keycloak 实例运行。

  3. 获取可以访问管理控制台的外部主机名。 

    NS=sso
oc get ingress keycloak -n $NS
    Copy to Clipboard Toggle word wrap

    在这个示例中，我们假设红帽构建的 Keycloak 服务器在 sso 命名空间中运行。

  4. 获取 admin 用户的密码。 

    oc get -n $NS pod keycloak-0 -o yaml | less
    Copy to Clipboard Toggle word wrap

    密码存储为 secret，因此获取红帽构建的 Keycloak 实例的配置 YAML 文件，以识别 secret 的名称(secretKeyRef.name)。

  5. 使用 secret 的名称来获取明文密码。 

    SECRET_NAME=credential-keycloak
oc get -n $NS secret $SECRET_NAME -o yaml | grep PASSWORD | awk '{print $2}' | base64 -D
    Copy to Clipboard Toggle word wrap

    在本例中，我们假定 secret 的名称是 credential-keycloak。

  6. 使用用户名 admin 和密码您获取的密码，登录管理控制台。

    使用 https://HOSTNAME 访问 Kubernetes Ingress。

    现在，您可以使用 Admin 控制台将示例域上传到红帽构建的 Keycloak。

  7. 单击 Add Realm 以导入示例域。

  8. 添加 examples/security/keycloak-authorization/kafka-authz-realm.json 文件，然后点 Create。

    您现在具有 kafka-authz 作为管理控制台中的当前域。

    默认视图显示 Master 域。

  9. 在 Red Hat build of Keycloak Admin Console 中，进入 Clients > kafka > Authorization > Settings，检查 Decision Strategy 是否已设置为 Affirmative。

    高效策略意味着，必须至少满足一个策略才能访问 Kafka 集群。

  10. 在 Red Hat build of Keycloak Admin Console 中，进入 Groups,Users,Roles 和 Clients 来查看域配置。

    组
    组 组织用户和设置权限。组可以链接到 LDAP 身份提供程序，并且用于划分用户，如区域或部门。可以通过自定义 LDAP 接口将用户添加到组中，以管理 Kafka 资源的权限。有关使用 KeyCloak 的 LDAP 提供程序的更多信息，请参阅 Keycloak 服务器管理指南。
    用户
    用户定义 单个用户。在本例中，已创建 alice （在 ClusterManager 组中）和 bob （在 ClusterManager-my-cluster中）。用户也可以存储在 LDAP 身份提供程序中。
    角色
    角色 为用户或客户端分配特定权限。角色功能（如标签）授予用户某些权限。虽然角色无法存储在 LDAP 中，但您可以将红帽构建的 Keycloak 角色添加到组中，以组合角色和权限。
    客户端

    客户端 定义 Kafka 交互的配置。

    • kafka 客户端处理代理的 OAuth 2.0 令牌验证，并包含授权策略（需要启用授权）。
    • 命令行工具使用 kafka-cli 客户端来获取访问或刷新令牌。
    • team-a-client 和 team-b-client 代表对特定 Kafka 主题具有部分访问权限的服务。

  11. 在 Red Hat build of Keycloak Admin Console 中，进入 Authorization > Permissions 来查看使用为域定义的资源和策略授予权限。

    例如，kafka 客户端具有以下权限： 

    Dev Team A can write to topics that start with x_ on any cluster
Dev Team B can read from topics that start with x_ on any cluster
Dev Team B can update consumer group offsets that start with x_ on any cluster
ClusterManager of my-cluster Group has full access to cluster config on my-cluster
ClusterManager of my-cluster Group has full access to consumer groups on my-cluster
ClusterManager of my-cluster Group has full access to topics on my-cluster
    Copy to Clipboard Toggle word wrap
    Dev Team A
    Dev Team A realm 角色可以写入任何集群中以 x_ 开头的主题。它组合了名为 Topic:xuildDefaults、Describe 和 Write 范围的资源，以及 Dev Team A 策略。Dev Team A 策略与具有称为 Dev Team A 的 realm 角色的所有用户匹配。
    Dev Team B
    Dev Team B realm 角色可以从任何集群上的 x_ 开头的主题读取。这将合并 Topic:xuild Defaults , Group:x object resources, Describe 和 Read scopes, 和 Dev Team B 策略。Dev Team B 策略与具有称为 Dev Team B 的 realm 角色的所有用户匹配。匹配用户和客户端能够从主题中读取，并为名称以 x_ 开头的主题和消费者组更新消耗的偏移量。

部署配置为连接到红帽构建的 Keycloak 服务器的 Kafka 集群。使用示例 kafka-ephemeral-oauth-single-keycloak-authz.yaml 文件将 Kafka 集群部署为 Kafka 自定义资源。这个示例使用 keycloak 授权和 oauth 身份验证部署单节点 Kafka 集群。

先决条件

  • Red Hat build of Keycloak 授权服务器部署到您的 OpenShift 集群中，并使用 example 域载入。
  • Cluster Operator 部署到 OpenShift 集群。
  • Apache Kafka 示例/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml 自定义资源的 Streams。

流程

  1. 使用您部署的红帽构建的 Keycloak 实例的主机名，为 Kafka 代理准备信任存储证书，以与红帽构建的 Keycloak 服务器通信。 

    SSO_HOST=SSO-HOSTNAME
SSO_HOST_PORT=$SSO_HOST:443
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem
    Copy to Clipboard Toggle word wrap

    该证书是必需的，因为 Kubernetes Ingress 用于进行安全(HTTPS)连接。

    通常没有单个证书，而是证书链。您只需要提供最顶层的签发者 CA，该 CA 在 /tmp/sso.pem 文件中最后列出。您可以手动提取它，也可以使用以下命令：

    在证书链中提取顶级 CA 证书的命令示例

    split -p "-----BEGIN CERTIFICATE-----" sso.pem sso-
for f in $(ls sso-*); do mv $f $f.pem; done
cp $(ls sso-* | sort -r | head -n 1) sso-ca.crt
    Copy to Clipboard Toggle word wrap

    注意

    可信 CA 证书通常从可信来源获取，而不使用 openssl 命令。

  2. 将证书作为机密部署到 OpenShift。 

    oc create secret generic oauth-server-cert --from-file=/tmp/sso-ca.crt -n $NS
    Copy to Clipboard Toggle word wrap

  3. 将主机名设置为环境变量 

    SSO_HOST=SSO-HOSTNAME
    Copy to Clipboard Toggle word wrap

  4. 创建和部署示例 Kafka 集群。 

    cat examples/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml | sed -E 's#\${SSO_HOST}'"#$SSO_HOST#" | oc create -n $NS -f -
    Copy to Clipboard Toggle word wrap

16.4.4.3. 为 CLI Kafka 客户端会话准备 TLS 连接
复制链接

为交互式 CLI 会话创建一个新 pod。使用红帽构建的用于 TLS 连接的 Keycloak 证书设置信任存储。truststore 是连接到红帽构建的 Keycloak 和 Kafka 代理。

先决条件

  • Red Hat build of Keycloak 授权服务器部署到您的 OpenShift 集群中，并使用 example 域载入。

    在 Red Hat build of Keycloak Admin Console 中，检查分配给客户端的角色显示在 Clients > Service Account Roles 中。

  • 配置为连接到红帽构建的 Keycloak 的 Kafka 集群已部署到您的 OpenShift 集群。

流程

  1. 使用 Kafka 镜像运行新的交互式 pod 容器，以连接到正在运行的 Kafka 代理。 

    NS=sso
oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-38-rhel9:2.8.0 kafka-cli -n $NS -- /bin/sh
    Copy to Clipboard Toggle word wrap
    注意

    如果 oc 超时等待镜像下载，后续的尝试可能会导致 AlreadyExists 错误。

  2. 附加到 pod 容器。 

    oc attach -ti kafka-cli -n $NS
    Copy to Clipboard Toggle word wrap

  3. 使用红帽构建的 Keycloak 实例的主机名，使用 TLS 为客户端连接准备证书。 

    SSO_HOST=SSO-HOSTNAME
SSO_HOST_PORT=$SSO_HOST:443
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem
    Copy to Clipboard Toggle word wrap

    通常没有单个证书，而是证书链。您只需要提供最顶层的签发者 CA，该 CA 在 /tmp/sso.pem 文件中最后列出。您可以手动提取它，也可以使用以下命令：

    在证书链中提取顶级 CA 证书的命令示例

    split -p "-----BEGIN CERTIFICATE-----" sso.pem sso-
for f in $(ls sso-*); do mv $f $f.pem; done
cp $(ls sso-* | sort -r | head -n 1) sso-ca.crt
    Copy to Clipboard Toggle word wrap

    注意

    可信 CA 证书通常从可信来源获取，而不使用 openssl 命令。

  4. 为 TLS 连接到 Kafka 代理创建一个信任存储。 

    keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso-ca.crt -noprompt
    Copy to Clipboard Toggle word wrap

  5. 使用 Kafka bootstrap 地址作为 Kafka 代理的主机名和 tls 侦听器端口(9093)来为 Kafka 代理准备证书。 

    KAFKA_HOST_PORT=my-cluster-kafka-bootstrap:9093
STOREPASS=storepass

echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.pem
    Copy to Clipboard Toggle word wrap

    获取的 .pem 文件通常不是单个证书，而是一个证书链。您只需要提供最顶层的签发者 CA，该 CA 在 /tmp/my-cluster-kafka.pem 文件中最后列出。您可以手动提取它，也可以使用以下命令：

    在证书链中提取顶级 CA 证书的命令示例

    split -p "-----BEGIN CERTIFICATE-----" /tmp/my-cluster-kafka.pem kafka-
for f in $(ls kafka-*); do mv $f $f.pem; done
cp $(ls kafka-* | sort -r | head -n 1) my-cluster-kafka-ca.crt
    Copy to Clipboard Toggle word wrap

    注意

    可信 CA 证书通常从可信来源获取，而不使用 openssl 命令。在本例中，我们假设客户端在部署了 Kafka 集群的同一命名空间中的 pod 中运行。如果客户端从 OpenShift 集群外部访问 Kafka 集群，您必须首先确定 bootstrap 地址。在这种情况下，您还可以从 OpenShift secret 直接获取集群证书，且不需要 openssl。如需更多信息，请参阅 第 14 章 设置 Kafka 集群的客户端访问权限。

  6. 将 Kafka 代理的证书添加到信任存储中。 

    keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka-ca.crt -noprompt
    Copy to Clipboard Toggle word wrap

    保持会话处于打开状态，以检查授权访问权限。

使用交互式 CLI 会话，检查通过红帽构建 Keycloak 域的授权规则。使用 Kafka 的示例制作者和消费者客户端应用检查，以使用不同级别访问权限的用户和服务帐户创建主题。

使用 team-a-client 和 team-b-client 客户端来检查授权规则。使用 alice admin 用户对 Kafka 执行额外的管理任务。

本例中使用的 Kafka 镜像包含 Kafka 生成者和消费者二进制文件。

先决条件

设置客户端和 admin 用户配置

  1. 使用 team-a-client 客户端的身份验证属性准备 Kafka 配置文件。 

    SSO_HOST=SSO-HOSTNAME

cat > /tmp/team-a-client.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="team-a-client" \
  oauth.client.secret="team-a-client-secret" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF
    Copy to Clipboard Toggle word wrap

    使用 SASL OAUTHBEARER 机制。此机制需要客户端 ID 和客户端 secret，这意味着客户端首先连接到红帽构建的 Keycloak 服务器来获取访问令牌。然后，客户端连接到 Kafka 代理，并使用访问令牌进行身份验证。

  2. 使用 team-b-client 客户端的身份验证属性准备 Kafka 配置文件。 

    cat > /tmp/team-b-client.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.client.id="team-b-client" \
  oauth.client.secret="team-b-client-secret" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF
    Copy to Clipboard Toggle word wrap

  3. 使用 curl 验证 admin 用户 alice，并执行密码授权身份验证来获取刷新令牌。 

    USERNAME=alice
PASSWORD=alice-password

GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" -H 'Content-Type: application/x-www-form-urlencoded' -d "grant_type=password&username=$USERNAME&password=$PASSWORD&client_id=kafka-cli&scope=offline_access" -s -k)

REFRESH_TOKEN=$(echo $GRANT_RESPONSE | awk -F "refresh_token\":\"" '{printf $2}' | awk -F "\"" '{printf $1}')
    Copy to Clipboard Toggle word wrap

    刷新令牌是一个离线令牌，它长期且不会过期。

  4. 使用 admin 用户 alice 的身份验证属性准备 Kafka 配置文件。 

    cat > /tmp/alice.properties << EOF
security.protocol=SASL_SSL
ssl.truststore.location=/tmp/truststore.p12
ssl.truststore.password=$STOREPASS
ssl.truststore.type=PKCS12
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.refresh.token="$REFRESH_TOKEN" \
  oauth.client.id="kafka-cli" \
  oauth.ssl.truststore.location="/tmp/truststore.p12" \
  oauth.ssl.truststore.password="$STOREPASS" \
  oauth.ssl.truststore.type="PKCS12" \
  oauth.token.endpoint.uri="https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF
    Copy to Clipboard Toggle word wrap

    kafka-cli 公共客户端用于 sasl.jaas.config 中的 oauth.client.id。由于它是公共客户端，因此不需要机密。客户端使用上一步中经过身份验证的刷新令牌进行身份验证。刷新令牌请求 scenes 后面的访问令牌，然后发送到 Kafka 代理进行身份验证。

生成具有授权访问权限的消息

使用 team-a-client 配置检查您可以生成消息到以 a_ 或 x_ 开头的主题。

  1. 写入主题 my-topic。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic my-topic \
  --producer.config=/tmp/team-a-client.properties
First message
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [my-topic] 错误。

    team-a-client 有一个 Dev Team A 角色，授予它对以 a_ 开头的主题执行任何受支持操作的权限，但只能写入以 x_ 开头的主题。名为 my-topic 的主题都匹配这些规则。

  2. 写入主题 a_messages。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --producer.config /tmp/team-a-client.properties
First message
Second message
    Copy to Clipboard Toggle word wrap

    消息被成功生成到 Kafka。

  3. 按 CTRL+C 退出 CLI 应用程序。

  4. 检查 Kafka 容器日志中 的授权 GRANTED 的调试日志。 

    oc logs my-cluster-kafka-0 -f -n $NS
    Copy to Clipboard Toggle word wrap

使用具有授权访问权限的消息

使用 team-a-client 配置来消耗来自主题 a_messages 的消息。

  1. 从主题 a_messages 获取消息。 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties
    Copy to Clipboard Toggle word wrap

    请求返回错误，因为 team-a-client 的 Dev Team A 角色只能访问名称以 a_ 开头的消费者组。

  2. 更新 team-a-client 属性，以指定允许使用的自定义消费者组。 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_1
    Copy to Clipboard Toggle word wrap

    消费者从 a_messages 主题接收所有消息。

使用授权访问权限管理 Kafka

team-a-client 是一个帐户，没有集群级别的访问权限，但可用于一些管理操作。

  1. 列出主题。 

    bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
    Copy to Clipboard Toggle word wrap

    返回 a_messages 主题。

  2. 列出消费者组。 

    bin/kafka-consumer-groups.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
    Copy to Clipboard Toggle word wrap

    返回 a_consumer_group_1 使用者组。

    获取集群配置的详情。 

    bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties \
  --entity-type brokers --describe --entity-default
    Copy to Clipboard Toggle word wrap

    请求返回错误，因为操作需要 team-a-client 没有的集群级别权限。

使用具有不同权限的客户端

使用 team-b-client 配置为以 b_ 开头的主题生成消息。

  1. 写入主题 a_messages。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic a_messages \
  --producer.config /tmp/team-b-client.properties
Message 1
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [a_messages] 错误。

  2. 写入主题 b_messages。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic b_messages \
  --producer.config /tmp/team-b-client.properties
Message 1
Message 2
Message 3
    Copy to Clipboard Toggle word wrap

    消息被成功生成到 Kafka。

  3. 写入主题 x_messages。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-b-client.properties
Message 1
    Copy to Clipboard Toggle word wrap

    返回了 Not authorized access topics: [x_messages] 错误，team-b-client 只能从主题 x_messages 中读取。

  4. 使用 team-a-client 写入主题 x_messages。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-a-client.properties
Message 1
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [x_messages] 错误。team-a-client 可以写入 x_messages 主题，但如果它尚不存在，则没有创建主题的权限。在 team-a-client 可以写入 x_messages 主题之前，admin power 用户 必须使用正确的配置创建它，如分区和副本数。

使用授权 admin 用户管理 Kafka

使用 admin 用户 alice 管理 Kafka。alice 具有管理任何 Kafka 集群上的所有内容的完整访问权限。

  1. 以 alice 身份创建 x_messages 主题。 

    bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
  --topic x_messages --create --replication-factor 1 --partitions 1
    Copy to Clipboard Toggle word wrap

    主题已创建成功。

  2. 以 alice 身份列出所有主题。 

    bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties --list
bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list
bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-b-client.properties --list
    Copy to Clipboard Toggle word wrap

    admin 用户 alice 可以列出所有主题，而 team-a-client 和 team-b-client 只能列出他们有权访问的主题。

    Dev Team A 和 Dev Team B 角色对以 x_ 开头的主题具有 Describe 权限，但它们无法看到其他团队的主题，因为它们没有 Describe 权限。

  3. 使用 team-a-client 生成信息到 x_messages 主题： 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-a-client.properties
Message 1
Message 2
Message 3
    Copy to Clipboard Toggle word wrap

    当 alice 创建 x_messages 主题时，会在 Kafka 中成功生成信息。

  4. 使用 team-b-client 生成消息到 x_messages 主题。 

    bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --producer.config /tmp/team-b-client.properties
Message 4
Message 5
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [x_messages] 错误。

  5. 使用 team-b-client 来消耗来自 x_messages 主题的消息： 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --from-beginning --consumer.config /tmp/team-b-client.properties --group x_consumer_group_b
    Copy to Clipboard Toggle word wrap

    消费者接收来自 x_messages 主题的所有消息。

  6. 使用 team-a-client 来消耗来自 x_messages 主题的消息。 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties --group x_consumer_group_a
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [x_messages] 错误。

  7. 使用 team-a-client 来消耗来自以 a_ 开头的消费者组的消息。 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --from-beginning --consumer.config /tmp/team-a-client.properties --group a_consumer_group_a
    Copy to Clipboard Toggle word wrap

    此请求会返回 Not authorized to access topics: [x_messages] 错误。

    Dev Team A 没有以 x_ 开始的主题的 Read 访问权限。

  8. 使用 alice 生成消息到 x_messages 主题。 

    bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
  --from-beginning --consumer.config /tmp/alice.properties
    Copy to Clipboard Toggle word wrap

    消息被成功生成到 Kafka。

    alice 可以从任何主题读取或写入到任何主题。

  9. 使用 alice 读取集群配置。 

    bin/kafka-configs.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/alice.properties \
  --entity-type brokers --describe --entity-default
    Copy to Clipboard Toggle word wrap

    本例中的集群配置为空。

返回顶部
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务，以及可以信赖的内容，帮助红帽用户创新并实现他们的目标。 了解我们当前的更新.

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情，请参阅红帽博客.

關於紅帽

我们提供强化的解决方案，使企业能够更轻松地跨平台和环境（从核心数据中心到网络边缘）工作。

Theme

© 2025 Red Hat