14.5. 使用基于 OAuth 2.0 令牌的授权

如果您在 Red Hat Single Sign-On 中使用 OAuth 2.0 进行基于令牌的身份验证，您还可以使用 Red Hat Single Sign-On 来配置授权规则来限制客户端对 Kafka 代理的访问。身份验证建立用户的身份。授权决定该用户的访问权限级别。

AMQ Streams 支持通过 Red Hat Single Sign-On Authorization Services 使用基于 OAuth 2.0 令牌的授权，它允许您集中管理安全策略和权限。

Red Hat Single Sign-On 中定义的安全策略和权限用于授予对 Kafka 代理上资源的访问权限。用户和客户端与允许对 Kafka 代理执行特定操作的策略进行匹配。

Kafka 允许所有用户默认对代理进行完全访问，并提供 AclAuthorizer 和 StandardAuthorizer 插件来配置基于访问控制列表(ACL)的授权。由这些插件管理的 ACL 规则用于根据 用户名 授予或拒绝对资源的访问，这些规则存储在 Kafka 集群本身中。但是，红帽单点登录基于 OAuth 2.0 令牌的授权在您希望实现对 Kafka 代理的访问控制方面具有更大的灵活性。另外，您可以将 Kafka 代理配置为使用 OAuth 2.0 授权和 ACL。

14.5.1. OAuth 2.0 授权机制
复制链接

AMQ Streams 中的 OAuth 2.0 授权使用红帽单点登录服务器授权服务 REST 端点，通过在特定用户上应用定义的安全策略来扩展基于令牌的身份验证，并为该用户提供授予不同资源的权限列表。策略使用角色和组来匹配用户的权限。OAuth 2.0 授权根据从 Red Hat Single Sign-On Authorization Services 用户获得的授予者列表在本地强制实施权限。

14.5.1.1. Kafka 代理自定义授权器
复制链接

AMQ Streams 提供了一个 Red Hat Single Sign-On authorizer (KeycloakAuthorizer)。为了可以使用 Red Hat Single Sign-On 提供的授权服务的 Red Hat Single Sign-On REST 端点，您可以在 Kafka 代理上配置自定义授权器。

授权程序根据需要从授权服务器获取授予权限的列表，并在 Kafka Broker 上本地强制实施授权，为每个客户端请求做出快速授权决策。

14.5.2. 配置 OAuth 2.0 授权支持
复制链接

这个步骤描述了如何使用 Red Hat Single Sign-On Authorization Services 将 Kafka 代理配置为使用 OAuth 2.0 授权服务。

开始前

考虑某些用户所需的访问权限或希望限制某些用户。您可以使用 Red Hat Single Sign-On 组、角色、客户端 和 用户在 Red Hat Single Sign-On 中配置访问权限的组合。

通常，组用于根据机构部门或地理位置匹配用户。和角色用于根据其功能匹配用户。

使用红帽单点登录，您可以在 LDAP 中存储用户和组，而客户端和角色不能以这种方式存储。存储和对用户数据的访问可能是您选择配置授权策略的一个因素。

注意

无论在 Kafka 代理上实现的授权是什么，超级用户始终对 Kafka 代理具有不受限制的访问权限。

先决条件

AMQ Streams 必须通过 Red Hat Single Sign-On 配置为使用 OAuth 2.0 进行基于令牌的身份验证。设置授权时，您可以使用相同的 Red Hat Single Sign-On 服务器端点。
OAuth 2.0 身份验证必须使用 maxSecondsWithoutReauthentication 选项进行配置，才能启用重新身份验证。

流程

访问 Red Hat Single Sign-On Admin 控制台，或使用 Red Hat Single Sign-On Admin CLI 为设置 OAuth 2.0 身份验证时创建的 Kafka 代理客户端启用授权服务。
使用 Authorization Services 为客户端定义资源、授权范围、策略和权限。
通过分配角色和组，将权限绑定到用户和客户端。
通过在编辑器中更新 Kafka 资源的 Kafka 代理配置(Kafka.spec.kafka)，将 Kafka 代理配置为使用 Red Hat Single Sign-On 授权。
```
oc edit kafka my-cluster
```
```
oc edit kafka my-cluster
```
Copy to Clipboard Toggle word wrap
将 Kafka 代理 kafka 配置配置为使用 keycloak 授权，并能够访问授权服务器和授权服务。
例如：
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    authorization:
      type: keycloak 
      tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 
      clientId: kafka 
      delegateToKafkaAcls: false 
      disableTlsHostnameVerification: false 
      superUsers: 
      - CN=fred
      - sam
      - CN=edward
      tlsTrustedCertificates: 
      - secretName: oauth-server-cert
        certificate: ca.crt
      grantsRefreshPeriodSeconds: 60 
      grantsRefreshPoolSize: 5 
      grantsMaxIdleSeconds: 300 
      grantsGcPeriodSeconds: 300 
      grantsAlwaysLatest: false 
      connectTimeoutSeconds: 60 
      readTimeoutSeconds: 60 
      httpRetries: 2 
      enableMetrics: false 
      includeAcceptHeader: false 
    #...
```
```
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    authorization:
      type: keycloak 
```
1
```
      tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 
```
2
```
      clientId: kafka 
```
3
```
      delegateToKafkaAcls: false 
```
4
```
      disableTlsHostnameVerification: false 
```
5
```
      superUsers: 
```
6
```
      - CN=fred
      - sam
      - CN=edward
      tlsTrustedCertificates: 
```
7
```
      - secretName: oauth-server-cert
        certificate: ca.crt
      grantsRefreshPeriodSeconds: 60 
```
8
```
      grantsRefreshPoolSize: 5 
```
9
```
      grantsMaxIdleSeconds: 300 
```
10
```
      grantsGcPeriodSeconds: 300 
```
11
```
      grantsAlwaysLatest: false 
```
12
```
      connectTimeoutSeconds: 60 
```
13
```
      readTimeoutSeconds: 60 
```
14
```
      httpRetries: 2 
```
15
```
      enableMetrics: false 
```
16
```
      includeAcceptHeader: false 
```
17
```
    #...
```
Copy to Clipboard Toggle word wrap
1
type keycloak 启用 Red Hat Single Sign-On 授权。
2
Red Hat Single Sign-On 令牌端点的 URI。对于生产环境，请始终使用 https:// urls。当您配置基于令牌的 oauth 身份验证时，您可以将 jwksEndpointUri 指定为本地 JWT 验证的 URI。tokenEndpointUri URI 的主机名必须相同。
3
在启用了授权服务的 Red Hat Single Sign-On 中 OAuth 2.0 客户端定义的客户端 ID。通常，kafka 被用作 ID。
4
（可选）如果 Red Hat Single Sign-On Authorization Services 策略无法访问，对 Kafka AclAuthorizer 和 StandardAuthorizer 的 Delegate 授权。默认为 false。
5
（可选）禁用 TLS 主机名验证。默认为 false。
6
（可选）设计的超级用户。
7
（可选）与授权服务器进行 TLS 连接的可信证书。
8
（可选）两个连续之间的时间授予刷新运行。这是活动会话的最长时间，用于检测 Red Hat Single Sign-On 上用户的任何权限更改。默认值为 60。
9
（可选）用于刷新（并行）活跃会话授予的线程数量。默认值为 5。
10
（可选）缓存中闲置授权可以被驱除的时间（以秒为单位）。默认值为 300。
11
（可选）连续运行从缓存中清除过时的作业之间的时间（以秒为单位）。默认值为 300。
12
（可选）控制是否为新会话获取最新的授权。启用后，从 Red Hat Single Sign-On 检索并缓存该用户。默认值为 false。
13
（可选）连接到 Red Hat Single Sign-On 令牌端点时的连接超时（以秒为单位）。默认值为 60。
14
（可选）连接到 Red Hat Single Sign-On 令牌端点时读取超时（以秒为单位）。默认值为 60。
15
（可选）重试的最大次数（不会暂停）授权服务器的失败 HTTP 请求。默认值为 0， 表示不会执行重试。要有效地使用这个选项，请考虑减少 connectTimeoutSeconds 和 readTimeoutSeconds 选项的超时时间。但请注意，重试可能会阻止当前 worker 线程可用于其他请求，如果太多请求停滞，则可能会导致 Kafka 代理无响应。
16
（可选）启用或禁用 OAuth 指标。默认值为 false。
17
（可选）有些授权服务器在客户端发送 Accept: application/json 标头时遇到问题。通过设置 includeAcceptHeader: false，不会发送标头。默认为 true。
保存并退出编辑器，然后等待滚动更新完成。
检查日志中的更新，或通过监视 pod 状态转换：
```
oc logs -f ${POD_NAME} -c kafka
oc get pod -w
```
```
oc logs -f ${POD_NAME} -c kafka
oc get pod -w
```
Copy to Clipboard Toggle word wrap
滚动更新将代理配置为使用 OAuth 2.0 授权。
通过以客户端或具有特定角色的用户访问 Kafka 代理来验证配置的权限，确保它们具有必要的访问权限，或者没有应该具有访问权限。

14.5.3. 在 Red Hat Single Sign-On Authorization Services 中管理策略和权限
复制链接

本节论述了 Red Hat Single Sign-On Authorization Services 和 Kafka 使用的授权模型，并定义每个模型中的重要概念。

要授予访问 Kafka 的权限，您可以通过在 Red Hat Single Sign-On 中创建 OAuth 客户端规格 将 Red Hat Single Sign-On Authorization Services 对象映射到 Kafka 资源。使用 Red Hat Single Sign-On Authorization Services 规则为用户帐户或服务帐户授予 Kafka 权限。

示例显示了常见 Kafka 操作所需的不同用户权限，如创建和列出主题。

14.5.3.1. Kafka 和 Red Hat Single Sign-On 授权模型概述
复制链接

Kafka 和 Red Hat Single Sign-On Authorization Services 使用不同的授权模型。

Kafka 授权模型

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

Kafka 使用五个资源类型来控制访问： Topic,Group,Cluster,TransactionalId, 和 DelegationToken。每个资源类型都有一组可用权限。

Topic

创建
写
读
删除
Describe
DescribeConfigs
更改
AlterConfigs

组

读
Describe
删除

集群

创建
Describe
更改
DescribeConfigs
AlterConfigs
IdempotentWrite
ClusterAction

transactionalId

Describe
写

DelegationToken

Describe

Red Hat Single Sign-On Authorization Services 模型

Red Hat Single Sign-On Authorization Services 模型有四个概念来定义和授予权限：resources（资源）, authorization scopes（授权范围）, policies（政策）, 和 permissions（权限）。

Resources

资源是一组资源定义，用于将资源与允许的操作匹配。资源可能是单个主题，例如，也可以是名称以同一前缀开头的所有主题。资源定义与一组可用的授权范围相关联，它代表了资源上可用的一组所有操作。通常，实际上只允许这些操作的子集。

授权范围

授权范围是一组对特定资源定义的所有可用操作。当您定义新资源时，您可以添加来自所有范围集的范围。

策略（policy）

策略是一种授权规则，它使用条件与帐户列表匹配。策略可以匹配：

基于客户端 ID 或角色 的服务帐户
基于用户名、组或角色 的用户帐户。

权限

权限向一组用户授予对特定资源定义的授权范围的子集。

14.5.3.2. 将 Red Hat Single Sign-On Authorization Services 映射到 Kafka 授权模型
复制链接

Kafka 授权模型用作定义 Red Hat Single Sign-On 角色和资源的基础，以控制对 Kafka 的访问。

要为用户帐户或服务帐户授予 Kafka 权限，您首先在 Red Hat Single Sign-On 中为 Kafka 代理创建 OAuth 客户端规格。然后，在客户端上指定 Red Hat Single Sign-On Authorization Services 规则。通常，代表代理的 OAuth 客户端的客户端 id 是 kafka。AMQ Streams 提供的示例配置文件使用 kafka 作为 OAuth 客户端 ID。

注意

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

kafka 客户端定义必须在 Red Hat Single Sign-On Admin 控制台中启用 Authorization Enabled 选项。

kafka 客户端的范围中存在所有权限。如果您使用不同的 OAuth 客户端 ID 配置不同的 Kafka 集群，它们需要单独的权限集，即使它们是同一 Red Hat Single Sign-On 域的一部分。

当 Kafka 客户端使用 OAUTHBEARER 身份验证时，Red Hat Single Sign-On 授权器(KeycloakAuthorizer)使用当前会话的访问令牌从 Red Hat Single Sign-On 服务器检索授权列表。要获得授权，授权者会评估 Red Hat Single Sign-On Authorization Services 策略和权限。

Kafka 权限的授权范围

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

授权范围必须包含所有可能的 Kafka 权限，无论资源类型是什么：

创建
写
读
删除
Describe
更改
DescribeConfig
AlterConfig
ClusterAction
IdempotentWrite

注意

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

权限检查的资源模式

在执行权限检查时，资源模式用于与目标资源匹配的模式。常规模式格式为 RESOURCE-TYPE:PATTERN-NAME。

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

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

资源模式示例

Topic:my-topic
Topic:orders-*
Group:orders-*
Cluster:*

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_*

kafka-cluster:my-cluster,Topic:*
kafka-cluster:*,Group:b_*

Copy to Clipboard

Toggle word wrap

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

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

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

应用访问权限的策略

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

特定用户或服务帐户
域角色或客户端角色
用户组
与客户端 IP 地址匹配的 JavaScript 规则

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

授予访问权限的权限

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

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

14.5.3.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

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

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

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

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-*

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

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

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

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

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

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

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

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

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

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

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

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

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

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

14.5.4. Trying Red Hat Single Sign-On Authorization Services
复制链接

本例解释了如何在 keycloak 授权中使用 Red Hat Single Sign-On Authorization Services。使用 Red Hat Single Sign-On Authorization Services 对 Kafka 客户端强制实施访问限制。Red Hat Single Sign-On Authorization Services 使用授权范围、策略和权限来定义并应用对资源的访问控制。

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

AMQ Streams 提供示例配置文件。这包括设置 Red Hat Single Sign-On 的以下示例文件：

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

如果要尝试使用 Red Hat Single Sign-On 的示例，请使用这些文件执行本节中概述的任务，如显示的顺序。

身份验证

当您配置基于令牌的 oauth 身份验证时，您可以将 jwksEndpointUri 指定为本地 JWT 验证的 URI。配置 keycloak 授权时，您可以将 tokenEndpointUri 指定为 Red Hat Single Sign-On 令牌端点的 URI。两个 URI 的主机名必须相同。

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

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

14.5.4.1. 访问 Red Hat Single Sign-On 管理控制台
复制链接

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

先决条件

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

流程

使用 Red Hat Single Sign-On Operator 安装 Red Hat Single Sign-On 服务器，如 Red Hat Single Sign-On 文档中的 Server Installation and configuration 所述。
等待 Red Hat Single Sign-On 实例正在运行。
获取可以访问管理控制台的外部主机名。
```
NS=sso
oc get ingress keycloak -n $NS
```
```
NS=sso
oc get ingress keycloak -n $NS
```
Copy to Clipboard Toggle word wrap
在本例中，我们假定 Red Hat Single Sign-On 服务器在 sso 命名空间中运行。
获取 admin 用户的密码。
```
oc get -n $NS pod keycloak-0 -o yaml | less
```
```
oc get -n $NS pod keycloak-0 -o yaml | less
```
Copy to Clipboard Toggle word wrap
密码存储为 secret，因此获取 Red Hat Single Sign-On 实例的配置 YAML 文件，以识别 secret 的名称(secretKeyRef.name)。

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

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

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。

使用用户名 admin 和密码您获取的密码，登录管理控制台。
使用 https://HOSTNAME 访问 Kubernetes Ingress。
现在，您可以使用管理控制台将示例域上传到 Red Hat Single Sign-On。
单击 Add Realm 以导入示例域。
添加 examples/security/keycloak-authorization/kafka-authz-realm.json 文件，然后点 Create。
您现在具有 kafka-authz 作为管理控制台中的当前域。
默认视图显示 Master 域。
在 Red Hat Single Sign-On Admin 控制台中，进入 Clients > kafka > Authorization > Settings，检查 Decision Strategy 是否已设置为 Affirmative。
高效策略意味着，必须至少满足一个策略才能访问 Kafka 集群。
在 Red Hat Single Sign-On Admin Console 中，进入 Groups, Users, Roles 和 Clients 来查看 realm 配置。
组
组 用于创建用户组并设置用户权限。组是分配了名称的用户集合。它们用于将用户划分成地理、组织或部门单元。组可以链接到 LDAP 身份提供程序。您可以通过自定义 LDAP 服务器 admin 用户界面使用户成为组的成员，例如，授予对 Kafka 资源的权限。
用户
用户 用于创建用户。在本例中，定义了 alice 和 bob。alice 是 ClusterManager 组的成员，bob 是 ClusterManager-my-cluster 组的成员。用户可以存储在 LDAP 身份提供程序中。
角色
Roles 代表用户或客户端具有特定权限。角色是一种类似于组的概念。它们通常 用于标记 具有组织角色的用户，并具有必要的权限。角色不能存储在 LDAP 身份提供程序中。如果 LDAP 是要求，您可以使用组，并将 Red Hat Single Sign-On 角色添加到组中，以便在为用户分配他们获得对应的角色时。
客户端
客户端可以 具有特定配置。在本例中，配置了 kafka, kafka-cli, team-a-client, 和 team-b-client 客户端。
Kafka 代理使用 kafka 客户端来执行访问令牌验证所需的 OAuth 2.0 通信。此客户端还包含用于在 Kafka 代理上执行授权的授权服务资源定义、策略和授权范围。授权配置在 Authorization 选项卡的 kafka 客户端中定义，在从 Settings 标签页上切换 Authorization 时，它就会可见。
kafka-cli 客户端是一个公共客户端，在使用用户名和密码进行身份验证时由 Kafka 命令行工具使用，以获取访问令牌或刷新令牌。
team-a-client 和 team-b-client 客户端是代表服务的机密客户端，对某些 Kafka 主题具有部分访问权限。

在 Red Hat Single Sign-On Admin 控制台中，进入 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

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_ 开头的主题和消费者组更新消耗的偏移量。

14.5.4.2. 使用 Red Hat Single Sign-On 授权部署 Kafka 集群
复制链接

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

先决条件

Red Hat Single Sign-On 授权服务器部署到 OpenShift 集群，并使用 example 域加载。
Cluster Operator 部署到 OpenShift 集群。
AMQ Streams 示例/security/keycloak-authorization/kafka-ephemeral-oauth-single-keycloak-authz.yaml 自定义资源。

流程

使用您部署的 Red Hat Single Sign-On 实例的主机名，为 Kafka 代理准备信任存储证书，以便与 Red Hat Single Sign-On 服务器通信。

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

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

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 命令。

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

oc create secret generic oauth-server-cert --from-file=/tmp/sso-ca.crt -n $NS

oc create secret generic oauth-server-cert --from-file=/tmp/sso-ca.crt -n $NS

Copy to Clipboard

Toggle word wrap

将主机名设置为环境变量
```
SSO_HOST=SSO-HOSTNAME
```
```
SSO_HOST=SSO-HOSTNAME
```
Copy to Clipboard Toggle word wrap

创建和部署示例 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 -

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

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

为交互式 CLI 会话创建一个新 pod。使用红帽单点登录证书为 TLS 连接设置信任存储。truststore 是连接到 Red Hat Single Sign-On 和 Kafka 代理。

先决条件

Red Hat Single Sign-On 授权服务器部署到 OpenShift 集群，并使用 example 域加载。
在 Red Hat Single Sign-On Admin 控制台中，检查分配给客户端的角色是否显示在 Clients > Service Account Roles 中。
配置为连接到 Red Hat Single Sign-On 的 Kafka 集群已部署到 OpenShift 集群。

流程

使用 AMQ Streams Kafka 镜像运行新的交互式 pod 容器，以连接到正在运行的 Kafka 代理。
```
NS=sso
oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-36-rhel8:2.6.0 kafka-cli -n $NS -- /bin/sh
```
```
NS=sso
oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-36-rhel8:2.6.0 kafka-cli -n $NS -- /bin/sh
```
Copy to Clipboard Toggle word wrap
注意
如果 oc 超时等待镜像下载，后续的尝试可能会导致 AlreadyExists 错误。
附加到 pod 容器。
```
oc attach -ti kafka-cli -n $NS
```
```
oc attach -ti kafka-cli -n $NS
```
Copy to Clipboard Toggle word wrap

使用 Red Hat Single Sign-On 实例的主机名，为使用 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

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

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 命令。

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

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso-ca.crt -noprompt

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias sso -storepass $STOREPASS -import -file /tmp/sso-ca.crt -noprompt

Copy to Clipboard

Toggle word wrap

使用 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
```
```
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
```
```
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。如需更多信息，请参阅第 13 章 设置 Kafka 集群的客户端访问权限。

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

keytool -keystore /tmp/truststore.p12 -storetype pkcs12 -alias my-cluster-kafka -storepass $STOREPASS -import -file /tmp/my-cluster-kafka-ca.crt -noprompt

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

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

14.5.4.4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问
复制链接

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

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

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

先决条件

ZooKeeper 和 Kafka 在 OpenShift 集群中运行，以便能够发送和接收信息。
启动交互式 CLI Kafka 客户端会话。
Apache Kafka 下载。

设置客户端和 admin 用户配置

使用 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/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

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/auth/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，这意味着客户端首先连接到 Red Hat Single Sign-On 服务器来获取访问令牌。然后，客户端连接到 Kafka 代理，并使用访问令牌进行身份验证。

使用 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/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

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/auth/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

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

USERNAME=alice
PASSWORD=alice-password

GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/auth/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}')

USERNAME=alice
PASSWORD=alice-password

GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/auth/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

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

使用 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/auth/realms/kafka-authz/protocol/openid-connect/token" ;
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler
EOF

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/auth/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_ 开头的主题。

写入主题 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
```
```
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 的主题都匹配这些规则。

写入主题 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

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。

按 CTRL+C 退出 CLI 应用程序。
检查 Kafka 容器日志中 的授权 GRANTED 的调试日志。
```
oc logs my-cluster-kafka-0 -f -n $NS
```
```
oc logs my-cluster-kafka-0 -f -n $NS
```
Copy to Clipboard Toggle word wrap

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

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

从主题 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

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_ 开头的消费者组。

更新 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

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 是一个帐户，没有集群级别的访问权限，但可用于一些管理操作。

列出主题。

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-a-client.properties --list

Copy to Clipboard

Toggle word wrap

返回 a_messages 主题。

列出消费者组。

bin/kafka-consumer-groups.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --command-config /tmp/team-a-client.properties --list

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

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_ 开头的主题生成消息。

写入主题 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

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] 错误。

写入主题 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

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。

写入主题 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

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 中读取。

使用 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
```
```
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 集群上的所有内容的完整访问权限。

以 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

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

主题已创建成功。

以 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

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 权限。

使用 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

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 中成功生成信息。

使用 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

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] 错误。

使用 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

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 主题的所有消息。

使用 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

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] 错误。

使用 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

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 访问权限。

使用 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

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 可以从任何主题读取或写入到任何主题。

使用 alice 读取集群配置。

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

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

本例中的集群配置为空。

返回顶部

14.5. 使用基于 OAuth 2.0 令牌的授权

14.5.1. OAuth 2.0 授权机制
复制链接

14.5.1.1. Kafka 代理自定义授权器
复制链接

14.5.2. 配置 OAuth 2.0 授权支持
复制链接

14.5.3. 在 Red Hat Single Sign-On Authorization Services 中管理策略和权限
复制链接

14.5.3.1. Kafka 和 Red Hat Single Sign-On 授权模型概述
复制链接

Kafka 授权模型

Red Hat Single Sign-On Authorization Services 模型

14.5.3.2. 将 Red Hat Single Sign-On Authorization Services 映射到 Kafka 授权模型
复制链接

14.5.3.3. Kafka 操作所需的权限示例
复制链接

14.5.4. Trying Red Hat Single Sign-On Authorization Services
复制链接

14.5.4.1. 访问 Red Hat Single Sign-On 管理控制台
复制链接

14.5.4.2. 使用 Red Hat Single Sign-On 授权部署 Kafka 集群
复制链接

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

14.5.4.4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

14.5. 使用基于 OAuth 2.0 令牌的授权

14.5.1. OAuth 2.0 授权机制复制链接链接已复制到粘贴板!

14.5.1.1. Kafka 代理自定义授权器复制链接链接已复制到粘贴板!

14.5.2. 配置 OAuth 2.0 授权支持复制链接链接已复制到粘贴板!

14.5.3. 在 Red Hat Single Sign-On Authorization Services 中管理策略和权限复制链接链接已复制到粘贴板!

14.5.3.1. Kafka 和 Red Hat Single Sign-On 授权模型概述复制链接链接已复制到粘贴板!

Kafka 授权模型

Red Hat Single Sign-On Authorization Services 模型

14.5.3.2. 将 Red Hat Single Sign-On Authorization Services 映射到 Kafka 授权模型复制链接链接已复制到粘贴板!

14.5.3.3. Kafka 操作所需的权限示例复制链接链接已复制到粘贴板!

14.5.4. Trying Red Hat Single Sign-On Authorization Services复制链接链接已复制到粘贴板!

14.5.4.1. 访问 Red Hat Single Sign-On 管理控制台复制链接链接已复制到粘贴板!

14.5.4.2. 使用 Red Hat Single Sign-On 授权部署 Kafka 集群复制链接链接已复制到粘贴板!

14.5.4.3. 为 CLI Kafka 客户端会话准备 TLS 连接复制链接链接已复制到粘贴板!

14.5.4.4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

14.5.1. OAuth 2.0 授权机制
复制链接

14.5.1.1. Kafka 代理自定义授权器
复制链接

14.5.2. 配置 OAuth 2.0 授权支持
复制链接

14.5.3. 在 Red Hat Single Sign-On Authorization Services 中管理策略和权限
复制链接

14.5.3.1. Kafka 和 Red Hat Single Sign-On 授权模型概述
复制链接

14.5.3.2. 将 Red Hat Single Sign-On Authorization Services 映射到 Kafka 授权模型
复制链接

14.5.3.3. Kafka 操作所需的权限示例
复制链接

14.5.4. Trying Red Hat Single Sign-On Authorization Services
复制链接

14.5.4.1. 访问 Red Hat Single Sign-On 管理控制台
复制链接

14.5.4.2. 使用 Red Hat Single Sign-On 授权部署 Kafka 集群
复制链接

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

14.5.4.4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问
复制链接