16.4. Red Hat build of Keycloak での権限のセットアップ

手順

1. Red Hat build of Keycloak ドキュメントの スタートガイド の説明に従って、Red Hat build of Keycloak Operator を使用して Red Hat build of Keycloak サーバーをインストールします。
2. Red Hat build of Keycloak インスタンスが実行されるまで待ちます。

3. 管理コンソールにアクセスできるように外部ホスト名を取得します。

   NS=sso
   oc get ingress keycloak -n $NS

   Copy to Clipboard

   この例では、Red Hat build of Keycloak サーバーが sso namespace で実行されていると想定しています。

4. admin ユーザーのパスワードを取得します。

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

   Copy to Clipboard

   パスワードはシークレットとして保存されるため、Red Hat build of Keycloak インスタンスの設定 YAML ファイルを取得して、シークレットの名前 (secretKeyRef.name) を特定します。

5. シークレットの名前を使用して、クリアテキストのパスワードを取得します。

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

   Copy to Clipboard

   この例では、シークレットの名前が credential-keycloak であることを前提としています。

6. ユーザー名 admin と取得したパスワードを使用して、管理コンソールにログインします。

   https://HOSTNAME を使用して Kubernetes Ingress にアクセスします。

   管理コンソールを使用して、サンプルレルムを Red Hat build of Keycloak にアップロードできるようになりました。

7. Add Realm をクリックして、サンプルレルムをインポートします。

8. examples/security/keycloak-authorization/kafka-authz-realm.json ファイルを追加してから Create をクリックします。

   これで、管理コンソールの現在のレルムとして kafka-authz が含まれるようになりました。

   デフォルトビューには、Master レルムが表示されます。

9. Red Hat build of Keycloak 管理コンソールで、Clients > kafka > Authorization > Settings に移動し、Decision Strategy が Affirmative に設定されていることを確認します。

   肯定的な (Affirmative) ポリシーとは、クライアントが Kafka クラスターにアクセスするためには少なくとも 1 つのポリシーが満たされている必要があることを意味します。

10. Red Hat build of Keycloak 管理コンソールで、Groups、Users、Roles、Clients に移動して、レルム設定を表示します。

    グループ

    Groups はユーザーを整理し、権限を設定します。グループは LDAP アイデンティティープロバイダーにリンクして、リージョンや部門などによってユーザーを区分するために使用できます。カスタム LDAP インターフェイスを介してユーザーをグループに追加し、Kafka リソースの権限を管理できます。KeyCloak の LDAP プロバイダーの使用に関する詳細は、Keycloak サーバー管理 ガイドを参照してください。

    ユーザー

    Users は個々のユーザーを定義します。この例では、alice (ClusterManager グループ内) と bob (ClusterManager-my-cluster 内) を作成します。ユーザーは LDAP アイデンティティープロバイダーに保存することもできます。

    ロール

    Roles は、ユーザーまたはクライアントに特定の権限を割り当てます。ロールはタグのように機能し、ユーザーに特定の権限を付与します。ロールは LDAP に保存できませんが、Red Hat build of Keycloak ロールをグループに追加すると、ロールと権限の両方を組み合わせることができます。

    クライアント

    Clients は Kafka のやり取りの設定を定義します。

    • kafka クライアントは、ブローカーの OAuth 2.0 トークン検証を処理し、認可ポリシー (認可の有効化が必要) を含んでいます。
    • kafka-cli クライアントは、アクセストークンまたはリフレッシュトークンを取得するためにコマンドラインツールによって使用されます。
    • team-a-client と team-b-client は、特定の Kafka トピックへの部分的なアクセス権を持つサービスを表します。

11. Red Hat build of Keycloak 管理コンソールで、Authorization > Permissions に移動して、レルムに定義されたリソースとポリシーを使用する付与された権限を確認します。

    たとえば、kafka クライアントには次の権限があります。

    Dev Team A can write to topics that start with x_ on any cluster
    Dev Team B can read from topics that start with x_ on any cluster
    Dev Team B can update consumer group offsets that start with x_ on any cluster
    ClusterManager of my-cluster Group has full access to cluster config on my-cluster
    ClusterManager of my-cluster Group has full access to consumer groups on my-cluster
    ClusterManager of my-cluster Group has full access to topics on my-cluster

    Copy to Clipboard

    Dev Team A

    Dev チーム A レルムロールは、任意のクラスターで x_ で始まるトピックに書き込みできます。これは、Topic:x_* というリソース、Describe と Write のスコープ、そして Dev Team A のポリシーを組み合わせたものです。Dev Team A ポリシーは、Dev Team A というレルムロールを持つすべてのユーザーにマッチします。

    Dev Team B

    Dev チーム B レルムロールは、任意のクラスターで x_ で始まるトピックから読み取ることができます。これは、Topic:x_*、 Group:x_* のリソース、Describe と Read のスコープ、および Dev Team B のポリシーを組み合わせたものです。Dev Team B ポリシーは、Dev Team B というレルムロールを持つすべてのユーザーにマッチします。一致するユーザーおよびクライアントはトピックから読み取りでき、名前が x_ で始まるトピックおよびコンシューマーグループの消費されたオフセットを更新できます。

手順

1. デプロイした Red Hat build of Keycloak インスタンスのホスト名を使用して、Kafka ブローカーが Red Hat build of Keycloak サーバーと通信するためのトラストストア証明書を準備します。

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem

   Copy to Clipboard

   Kubernetes Ingress はセキュアな (HTTPS) 接続を確立するために使用されるため、証明書が必要です。

   通常、単一の証明書ではなく、証明書チェーンがあります。指定する必要があるのは、/tmp/sso.pem ファイルの最後にリストされている最上位の発行者 CA だけです。手動で抽出するか、次のコマンドを使用して抽出できます。

   証明書チェーンの最上位の CA 証明書を抽出するコマンドの例

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

   Copy to Clipboard

   注記

   信頼できる CA 証明書は通常、openssl コマンドを使用するのではなく、信頼できるソースから取得します。

2. シークレットとして OpenShift に証明書をデプロイします。

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

   Copy to Clipboard

3. ホスト名を環境変数として設定します。

   SSO_HOST=SSO-HOSTNAME

   Copy to Clipboard

4. サンプル Kafka クラスターを作成およびデプロイします。

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

   Copy to Clipboard

手順

1. Kafka イメージを使用して新しいインタラクティブな Pod コンテナーを実行し、実行中の Kafka ブローカーに接続します。

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

   Copy to Clipboard
   注記

   イメージのダウンロードの待機中に oc がタイムアウトする場合、その後の試行によって AlreadyExists エラーが発生することがあります。

2. Pod コンテナーにアタッチします。

   oc attach -ti kafka-cli -n $NS

   Copy to Clipboard

3. Red Hat build of Keycloak のホスト名を使用して、TLS を使用したクライアント接続用の証明書を準備します。

   SSO_HOST=SSO-HOSTNAME
   SSO_HOST_PORT=$SSO_HOST:443
   STOREPASS=storepass
   
   echo "Q" | openssl s_client -showcerts -connect $SSO_HOST_PORT 2>/dev/null | awk ' /BEGIN CERTIFICATE/,/END CERTIFICATE/ { print $0 } ' > /tmp/sso.pem

   Copy to Clipboard

   通常、単一の証明書ではなく、証明書チェーンがあります。指定する必要があるのは、/tmp/sso.pem ファイルの最後にリストされている最上位の発行者 CA だけです。手動で抽出するか、次のコマンドを使用して抽出できます。

   証明書チェーンの最上位の CA 証明書を抽出するコマンドの例

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

   Copy to Clipboard

   注記

   信頼できる CA 証明書は通常、openssl コマンドを使用するのではなく、信頼できるソースから取得します。

4. Kafka ブローカーへの TLS 接続のトラストストアを作成します。

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

   Copy to Clipboard

5. Kafka ブートストラップアドレスを Kafka ブローカーのホスト名および tls リスナーポート (9093) のホスト名として使用し、Kafka ブローカーの証明書を準備します。

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

   Copy to Clipboard

   取得した .pem ファイルは通常、1 つの証明書ではなく、証明書チェーンです。指定する必要があるのは、/tmp/my-cluster-kafka.pem ファイルの最後にリストされている最上位の発行者 CA のみです。手動で抽出するか、次のコマンドを使用して抽出できます。

   証明書チェーンの最上位の CA 証明書を抽出するコマンドの例

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

   Copy to Clipboard

   注記

   信頼できる CA 証明書は通常、openssl コマンドを使用するのではなく、信頼できるソースから取得します。この例では、Kafka クラスターがデプロイされたのと同じ namespace の Pod でクライアントが実行されていると想定しています。クライアントが OpenShift クラスターの外部から Kafka クラスターにアクセスしている場合は、最初にブートストラップアドレスを決定する必要があります。その場合、クラスター証明書を OpenShift シークレットから直接取得することもでき、openssl は必要ありません。詳細は、14章Kafka クラスターへのクライアントアクセスの設定 を参照してください。

6. Kafka ブローカーの証明書をトラストストアに追加します。

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

   Copy to Clipboard

   承認されたアクセスを確認するために、セッションを開いたままにします。

クライアントおよび管理ユーザーの設定

1. team-a-client クライアントの認証プロパティーで Kafka 設定ファイルを準備します。

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

   Copy to Clipboard

   SASL OAUTHBEARER メカニズムを使用します。このメカニズムにはクライアント ID とクライアントシークレットが必要です。つまり、クライアントはまず Red Hat build of Keycloak サーバーに接続してアクセストークンを取得します。その後、クライアントは Kafka ブローカーに接続し、アクセストークンを使用して認証します。

2. team-b-client クライアントの認証プロパティーで Kafka 設定ファイルを準備します。

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

   Copy to Clipboard

3. curl を使用して管理者ユーザー alice を認証し、パスワード付与認証を実行して更新トークンを取得します。

   USERNAME=alice
   PASSWORD=alice-password
   
   GRANT_RESPONSE=$(curl -X POST "https://$SSO_HOST/realms/kafka-authz/protocol/openid-connect/token" -H 'Content-Type: application/x-www-form-urlencoded' -d "grant_type=password&username=$USERNAME&password=$PASSWORD&client_id=kafka-cli&scope=offline_access" -s -k)
   
   REFRESH_TOKEN=$(echo $GRANT_RESPONSE | awk -F "refresh_token\":\"" '{printf $2}' | awk -F "\"" '{printf $1}')

   Copy to Clipboard

   更新トークンは、有効期間がなく、期限切れにならないオフライントークンです。

4. admin ユーザー alice の認証プロパティーで Kafka 設定ファイルを準備します。

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

   Copy to Clipboard

   kafka-cli パブリッククライアントは、sasl.jaas .config の oauth.client. id に使用されます。これはパブリッククライアントであるため、シークレットは必要ありません。クライアントは直前の手順で認証された更新トークンで認証されます。更新トークンは背後でアクセストークンを要求します。これは、認証のために 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

   Copy to Clipboard

   以下のリクエストは、Not authorized to access topics: [my-topic] エラーを返します。

   team-a-client は Dev Team A ロールを持っており、a_ で始まるトピックに対してサポートされているすべてのアクションを実行する権限を与えられていますが、x_ で始まるトピックへの書き込みのみ可能です。my-topic という名前のトピックは、これらのルールのいずれにも一致しません。

2. トピック a_messages に書き込む。

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

   Copy to Clipboard

   メッセージは Kafka に正常に生成されます。

3. CTRL+C を押して CLI アプリケーションを終了します。

4. リクエストについて、Kafka コンテナーログで Authorization GRANTED のデバッグログを確認します。

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

   Copy to Clipboard

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

   team-a-client の Dev Team A ロールは、名前が a_ で始まるコンシューマーグループのみにアクセスできるため、リクエストはエラーを返します。

2. team-a-client プロパティーを更新し、使用が許可されているカスタムコンシューマーグループを指定します。

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

   Copy to Clipboard

   コンシューマーは 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

   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

   a_consumer_group_1 コンシューマーグループが返されます。

   クラスター設定の詳細を取得します。

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

   Copy to Clipboard

   操作には 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

   Copy to Clipboard

   以下のリクエストは、Not authorized to access topics: [a_messages] エラーを返します。

2. トピック b_messages に書き込む。

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

   Copy to Clipboard

   メッセージは Kafka に正常に生成されます。

3. トピック x_messages に書き込む。

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

   Copy to Clipboard

   Not authorized to access topics: [x_messages] エラーが返され、team-b-client はトピック x_messages からのみ読み取りできます。

4. team-a-client を使用してトピック x_messages に書き込みます。

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

   Copy to Clipboard

   このリクエストは、Not authorized to access topics: [x_messages] エラーを返します。team-a-client は x_messages トピックに書き込みできますが、トピックが存在しない場合に作成するパーミッションがありません。team-a-client が x_messages トピックに書き込みできるようにするには、管理者 power user はパーティションやレプリカの数などの適切な設定で作成する必要があります。

1. alice として x_messages トピックを作成します。

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

   Copy to Clipboard

   トピックが正常に作成されました。

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

   管理者ユーザーの alice はすべてのトピックをリスト表示できますが、team-a-client と team-b-client は自分がアクセスできるトピックのみをリスト表示できます。

   Dev Team A ロールと Dev Team B ロールは、どちらも x_ で始まるトピックに対する Describe 権限を持っていますが、他のチームのトピックに対する Describe 権限を持っていないため、他のチームのトピックを見ることができません。

3. team-a-client を使用して、x_messages トピックにメッセージを生成します。

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

   Copy to Clipboard

   alice が x_messages トピックを作成すると、メッセージが正常に Kafka に生成されます。

4. team-b-client を使用して、x_messages トピックにメッセージを生成します。

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

   Copy to Clipboard

   このリクエストは、Not authorized to access topics: [x_messages] エラーを返します。

5. team-b-client を使用して、x_messages トピックからメッセージを消費します。

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

   Copy to Clipboard

   コンシューマーは、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

   このリクエストは、Not authorized to access topics: [x_messages] エラーを返します。

7. team-a-client を使用して、a_ で始まるコンシューマーグループからのメッセージを消費します。

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

   Copy to Clipboard

   このリクエストは、Not authorized to access topics: [x_messages] エラーを返します。

   Dev Team A には、x_ で始まるトピックの Read 権限がありません。

8. alice を使用して、x_messages トピックへのメッセージを生成します。

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

   Copy to Clipboard

   メッセージは 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

   この例のクラスター設定は空です。