4.5.4.4. 使用 CLI Kafka 客户端会话检查对 Kafka 的授权访问权限

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

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

   Copy to Clipboard Toggle word wrap

   使用 SASL OAUTHBEARER 机制。这种机制需要客户端 ID 和客户端 secret，这意味着客户端首先连接到红帽单点登录服务器来获取访问令牌。然后，客户端连接到 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

   Copy to Clipboard Toggle word wrap

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}')

   Copy to Clipboard Toggle word wrap

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

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

   cat > /tmp/alice.properties << EOF
   security.protocol=SASL_SSL
   ssl.truststore.location=/tmp/truststore.p12
   ssl.truststore.password=$STOREPASS
   ssl.truststore.type=PKCS12
   sasl.mechanism=OAUTHBEARER
   sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.refresh.token="$REFRESH_TOKEN" \
     oauth.client.id="kafka-cli" \
     oauth.ssl.truststore.location="/tmp/truststore.p12" \
     oauth.ssl.truststore.password="$STOREPASS" \
     oauth.ssl.truststore.type="PKCS12" \
     oauth.token.endpoint.uri="https://$SSO_HOST/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。由于它是公共客户端，因此不需要机密。客户端使用上一步中验证的刷新令牌进行身份验证。刷新令牌在后台请求访问令牌，然后发送到 Kafka 代理进行身份验证。

1. 写入到 主题 my 主题。

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

   Copy to Clipboard Toggle word wrap

   此请求 返回未授权访问主题：[my-topic] 错误。

   team-a-client 具有开发团队 A 角色，允许它对以 a_ 开头的主题执行任何受支持的操作，但只能写入 x_ 开头的主题。名为 my-topic 的主题都不匹配这些规则。

2. 写入主题 a_messages。

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

   Copy to Clipboard Toggle word wrap

   成功生成消息到 Kafka。

3. 按 CTRL+C 退出 CLI 应用。
4. 检查 Kafka 容器日志中请求的 Authorization GRANTED 的调试日志。

1. 从主题 a_messages 获取消息.

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

   Copy to Clipboard Toggle word wrap

   请求返回一个错误，因为 team-a-client 的 Dev 团队 A 角色仅有权访问名称以 a_ 开头的消费者组。

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

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

   Copy to Clipboard Toggle word wrap

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

1. 列出主题.

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

   Copy to Clipboard Toggle word wrap

   a_messages 主题返回。

2. 列出消费者组。

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

   Copy to Clipboard Toggle word wrap

   a_consumer_group_1 consumer 组返回。

   获取集群配置详情。

   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 没有集群级别权限。

1. 写入主题 a_messages。

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

   Copy to Clipboard Toggle word wrap

   此请求将 返回未授权访问主题：[a_messages] 错误。

2. 写入到主题 b_messages。

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

   Copy to Clipboard Toggle word wrap

   成功生成消息到 Kafka。

3. 写入主题 x_messages。

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

   Copy to Clipboard Toggle word wrap

   未授权访问主题：[x_messages] 错误返回，team-b-client 只能从主题 x_messages 中读取。

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

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

   Copy to Clipboard Toggle word wrap

   此请求 返回未授权访问主题：[x_messages] 错误。team-a-client 可以写入 x_messages 主题，但是如果主题尚不存在，则无权创建该主题。在 team-a-client 可以写入 x_messages 主题之前，管理员 高级用户 必须使用正确的配置创建它，如分区和副本的数量。

1. 创建 x_messages 主题，作为 alice。

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

   Copy to Clipboard Toggle word wrap

   主题创建成功。

2. 将所有主题列为 alice.

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

   Copy to Clipboard Toggle word wrap

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

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

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

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

   Copy to Clipboard Toggle word wrap

   当 alice 创建 x_messages 主题时，会成功生成消息到 Kafka。

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

   bin/kafka-console-producer.sh --broker-list 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

   此请求 返回未授权访问主题：[x_messages] 错误。

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

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

   Copy to Clipboard Toggle word wrap

   使用者从 x_messages 主题接收所有消息。

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

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

   Copy to Clipboard Toggle word wrap

   此请求 返回未授权访问主题：[x_messages] 错误。

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

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

   Copy to Clipboard Toggle word wrap

   此请求 返回未授权访问主题：[x_messages] 错误。

   Dev 团队 A 对 以 x_ 开头的主题没有 读取 访问权限。

8. 使用 alice 生成指向 x_messages 主题的消息。

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

   Copy to Clipboard Toggle word wrap

   成功生成消息到 Kafka。

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

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

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

   Copy to Clipboard Toggle word wrap

   本例的集群配置为空。