8.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 插件来配置基于 Access Control Lists (ACL)的授权。

ZooKeeper 存储 ACL 规则，该规则根据 用户名 授予或拒绝对资源的访问。但是，红帽单点登录基于 OAuth 2.0 令牌的授权在您希望实现对 Kafka 代理的访问控制方面具有更大的灵活性。另外，您可以将 Kafka 代理配置为使用 OAuth 2.0 授权和 ACL。

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

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

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

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

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

8.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 代理客户端启用授权服务。
使用授权服务为客户端定义资源、授权范围、策略和权限。
通过分配角色和组，将权限绑定到用户和客户端。
通过在编辑器中更新 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 
      connectTimeoutSeconds: 60 
      readTimeoutSeconds: 60 
      httpRetries: 2 
    #...
```
```
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
```
      connectTimeoutSeconds: 60 
```
10
```
      readTimeoutSeconds: 60 
```
11
```
      httpRetries: 2 
```
12
```
    #...
```
Copy to Clipboard Toggle word wrap
1
类型 keycloak 启用 Red Hat Single Sign-On authorization。
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 的授权授权。默认为 false。
5
（可选）禁用 TLS 主机名验证。默认为 false。
6
（可选）指定超级用户。
7
（可选）与授权服务器的 TLS 连接的可信证书。
8
（可选）两个连续授予刷新运行之间的时间。这是活动会话的最大时间，用于检测 Red Hat Single Sign-On 上用户的任何权限更改。默认值为 60。
9
（可选）用于刷新活跃会话的授予的线程数量（并行）。默认值为 5。
10
（可选）连接到 Red Hat Single Sign-On 令牌端点时的连接超时（以秒为单位）。默认值为 60。
11
（可选）连接到 Red Hat Single Sign-On 令牌端点时的读取超时（以秒为单位）。默认值为 60。
12
（可选）到授权服务器失败的 HTTP 请求的次数（无需暂停）的最大次数。默认值为 0， 表示没有执行重试。要有效地使用这个选项，请考虑减少 connectTimeoutSeconds 和 readTimeoutSeconds 选项的超时时间。但请注意，重试可能会阻止当前 worker 线程对其他请求使用，如果太多请求停滞，可能会导致 Kafka 代理无响应。
保存并退出编辑器，然后等待滚动更新完成。
检查日志中的更新，或者通过观察 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 代理来验证配置的权限，确保它们具有必要的访问权限，或者没有应该具有访问权限。

8.5.3. 在 Red Hat Single Sign-On 授权服务中管理策略和权限
复制链接

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

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

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

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

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

Kafka 授权模型

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

Kafka 使用五个资源类型来控制访问： 主题、组、集群、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 或角色 的服务帐户
基于用户名、组或角色 的用户帐户。

权限

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

8.5.3.2. 将 Red Hat Single Sign-On 授权服务映射到 Kafka 授权模型
复制链接

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

要为用户帐户或服务帐户授予 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 身份验证时，红帽单点登录授权器(KeycloakRBACAuthorizer)将使用当前会话的访问令牌从 Red Hat Single Sign-On 服务器检索授予的列表。为了检索授予，授权者评估 Red Hat Single Sign-On Authorization Services 策略和权限。

Kafka 权限的授权范围

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

授权范围必须包含所有可能的 Kafka 权限，而不考虑资源类型：

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

注意

如果您确定您不需要权限（例如，Idempot entWrite），您可以从授权范围列表中省略它。但是，那个权限无法用于 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 开头的主题中读取。

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

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

创建主题

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

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

如果主题尚未创建，并且启用了主题自动创建，则需要创建主题的权限。

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

8.5.4. Try 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 端点提供经过身份验证的用户授予资源的权限列表。授权（权限）列表从 Red Hat Single Sign-On 服务器获取，作为 Kafka 客户端建立身份验证会话的第一个操作。该列表在后台刷新，以便检测到对授权的更改。授予会在 Kafka 代理本地缓存并强制每个用户会话，以提供快速授权决策。

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

kafka-ephemeral-oauth-single-keycloak-authz.yaml: 使用 Red Hat Single Sign-On 为基于 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 代理进行身份验证前，需要用户登录。在这种情况下，访问令牌代表特定用户，而不是客户端应用程序。

8.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 examples/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

开发团队 A: Dev Team A realm 角色可写入任何集群上以 x_ 开头的主题。这会组合一个名为 Topic:x、Describe 和 Write 范围的资源，以及 Dev 团队 A 策略。Dev Team A 策略与具有名为 Dev Team A 的域角色匹配的所有用户匹配。
Dev Team B: Dev Team B realm 角色可以从任何集群中以 x_ 开头的主题中读取。这可组合 Topic:xö、Group:x the 资源、Describe 和 Read 范围，以及 Dev 团队 B 策略。Dev Team B 策略与具有名为 Dev Team B 的域角色匹配的所有用户。匹配用户和客户端能够读取主题，并为名称以 x_ 开头的主题和消费者组更新消耗的偏移量。

8.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 examples/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

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

为交互式 CLI 会话创建一个新 pod。使用 Red Hat Single Sign-On 证书为 TLS 连接设置信任存储。信任存储是连接到 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-34-rhel8:2.4.0 kafka-cli -n $NS -- /bin/sh
```
```
NS=sso
oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-34-rhel8:2.4.0 kafka-cli -n $NS -- /bin/sh
```
Copy to Clipboard Toggle word wrap
注意
如果 oc time out 等待镜像下载，后续的尝试可能会导致 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 命令。

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

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。如需更多信息，请参阅第 7 章 设置 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

使会话保持打开，以检查授权访问。

8.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 下载。

设置客户端和管理员用户配置

使用 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。由于它是公共客户端，因此不需要 secret。客户端使用上一步中身份验证的刷新令牌进行身份验证。刷新令牌请求 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 容器日志，以了解请求的 Authorization 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 to access: [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 主题之前，管理员用户 必须使用正确配置（如分区和副本数）创建它。

使用授权管理员用户管理 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

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

Dev 团队 A 和 Dev 团队 B 角色对以 x_ 开头的主题具有 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

本例中的集群配置为空。

返回顶部

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

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

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

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

8.5.3. 在 Red Hat Single Sign-On 授权服务中管理策略和权限
复制链接

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

Kafka 授权模型

Red Hat Single Sign-On Authorization Services 模型

8.5.3.2. 将 Red Hat Single Sign-On 授权服务映射到 Kafka 授权模型
复制链接

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

8.5.4. Try Red Hat Single Sign-On Authorization Services
复制链接

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

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

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

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

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

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

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

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

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

8.5.3. 在 Red Hat Single Sign-On 授权服务中管理策略和权限复制链接链接已复制到粘贴板!

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

Kafka 授权模型

Red Hat Single Sign-On Authorization Services 模型

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

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

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

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

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

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

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

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

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

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

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

8.5.3. 在 Red Hat Single Sign-On 授权服务中管理策略和权限
复制链接

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

8.5.3.2. 将 Red Hat Single Sign-On 授权服务映射到 Kafka 授权模型
复制链接

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

8.5.4. Try Red Hat Single Sign-On Authorization Services
复制链接

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

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

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

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