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

流程

1. 使用 Red Hat Single Sign-On Operator 安装 Red Hat Single Sign-On 服务器，如 Red Hat Single Sign-On 文档中的 Server Installation and configuration 所述。
2. 等待 Red Hat Single Sign-On 实例正在运行。

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

   NS=sso
   oc get ingress keycloak -n $NS

   在本例中，我们假定 Red Hat Single Sign-On 服务器在 sso 命名空间中运行。

4. 获取 admin 用户的密码。

   oc get -n $NS pod keycloak-0 -o yaml | less

   密码存储为 secret，因此获取 Red Hat Single Sign-On 实例的配置 YAML 文件，以识别 secret 的名称(secretKeyRef.name)。

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

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

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

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

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

   现在，您可以使用管理控制台将示例域上传到 Red Hat Single Sign-On。

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

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

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

   默认视图显示 Master 域。

9. 在 Red Hat Single Sign-On Admin 控制台中，进入 Clients > kafka > Authorization > Settings，检查 Decision Strategy 是否已设置为 Affirmative。

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

10. 在 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 主题具有部分访问权限。

11. 在 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

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

流程

1. 使用您部署的 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

   该证书是必需的，因为 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

   注意

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

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

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

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

   SSO_HOST=SSO-HOSTNAME

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

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

流程

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

   NS=sso
   oc run -ti --restart=Never --image=registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 kafka-cli -n $NS -- /bin/sh

   注意

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

2. 附加到 pod 容器。

   oc attach -ti kafka-cli -n $NS

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

   通常没有单个证书，而是证书链。您只需要提供最顶层的签发者 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

   注意

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

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

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

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

   KAFKA_HOST_PORT=my-cluster-kafka-bootstrap:9093
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $KAFKA_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/my-cluster-kafka.pem

   获取的 .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

   注意

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

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

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

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

设置客户端和 admin 用户配置

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

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

   使用 SASL OAUTHBEARER 机制。这个机制需要客户端 ID 和客户端 secret，这意味着客户端首先连接到 Red Hat Single Sign-On 服务器来获取访问令牌。然后，客户端连接到 Kafka 代理，并使用访问令牌进行身份验证。

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

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

3. 使用 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}')

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

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

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

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

1. 写入主题 my-topic。

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

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

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

2. 写入主题 a_messages。

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

   消息被成功生成到 Kafka。

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

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

   oc logs my-cluster-kafka-0 -f -n $NS

1. 从主题 a_messages 获取消息。

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

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

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

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

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

1. 列出主题。

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

   返回 a_messages 主题。

2. 列出消费者组。

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

   返回 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

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

1. 写入主题 a_messages。

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

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

2. 写入主题 b_messages。

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

   消息被成功生成到 Kafka。

3. 写入主题 x_messages。

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

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

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

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

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

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

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

   主题已创建成功。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

   bin/kafka-console-consumer.sh --bootstrap-server my-cluster-kafka-bootstrap:9093 --topic x_messages \
     --from-beginning --consumer.config /tmp/alice.properties

   消息被成功生成到 Kafka。

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

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

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

   本例中的集群配置为空。