第4章 Kafka へのセキュアなアクセスの管理

1. Kafka クライアントは承認サーバーからアクセストークンを要求します。これにはクライアント ID とシークレットを使用し、任意で更新トークンも使用します。
2. 承認サーバーによって新しいアクセストークンが生成されます。
3. Kafka クライアントは SASL OAUTHBEARER メカニズムを使用してアクセストークンを渡し、Kafka ブローカーの認証を行います。
4. Kafka ブローカーは、独自のクライアント ID およびシークレットを使用して、承認サーバーでトークンイントロスペクションエンドポイントを呼び出し、アクセストークンを検証します。
5. トークンが有効な場合は、Kafka クライアントセッションが確立されます。

1. Kafka クライアントは、トークンエンドポイントから承認サーバーの認証を行います。これにはクライアント ID とシークレットが使用され、任意で更新トークンも使用されます。
2. 承認サーバーによって新しいアクセストークンが生成されます。
3. Kafka クライアントは SASL OAUTHBEARER メカニズムを使用してアクセストークンを渡し、Kafka ブローカーの認証を行います。
4. Kafka ブローカーは、JWT トークン署名チェックおよびローカルトークンイントロスペクションを使用して、ローカルでアクセストークンを検証します。

1. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用して有効期限の長いアクセストークンを渡し、Kafka ブローカーの認証を行います。
2. Kafka ブローカーは、独自のクライアント ID およびシークレットを使用して、承認サーバーでトークンイントロスペクションエンドポイントを呼び出し、アクセストークンを検証します。
3. トークンが有効な場合は、Kafka クライアントセッションが確立されます。

1. Kafka クライアントは、SASL OAUTHBEARER メカニズムを使用して有効期限の長いアクセストークンを渡し、Kafka ブローカーの認証を行います。
2. Kafka ブローカーは、JWT トークン署名チェックおよびローカルトークンイントロスペクションを使用して、ローカルでアクセストークンを検証します。

手順

1. Red Hat Single Sign-On を OpenShift クラスターにデプロイします。

   OpenShift Web コンソールでデプロイメントの進捗を確認します。

2. Red Hat Single Sign-On の Admin Console にログインし、AMQ Streams の OAuth 2.0 ポリシーを作成します。

   ログインの詳細は、Red Hat Single Sign-On のデプロイ時に提供されます。

3. レルムを作成し、有効にします。

   既存のマスターレルムを使用できます。

4. 必要に応じて、レルムのセッションおよびトークンのタイムアウトを調整します。
5. kafka-broker というクライアントを作成します。

6. Settings タブで以下を設定します。

   • Access Type を Confidential に設定します。
   • Standard Flow Enabled を OFF に設定し、このクライアントからの Web ログインを無効にします。
   • Service Accounts Enabled を ON に設定し、このクライアントが独自の名前で認証できるようにします。
7. 続行する前に Save クリックします。
8. Credentials タブにある、AMQ Streams の Kafka クラスター設定で使用するシークレットを書き留めておきます。

9. Kafka ブローカーに接続するすべてのアプリケーションクライアントに対して、このクライアント作成手順を繰り返し行います。

   新しいクライアントごとに定義を作成します。

   設定では、名前をクライアント ID として使用します。

手順

1. エディターで、Kafka リソースの Kafka ブローカー設定 (Kafka.spec.kafka) を更新します。

   oc edit kafka my-cluster

2. Kafka ブローカーの listeners 設定を行います。

   各タイプのリスナーは独立しているため、同じ設定にする必要はありません。

   以下は、外部リスナーに設定された設定オプションの例になります。

   例 1: 高速なローカル JWT トークン検証の設定

   #...
   - name: external
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth 1
       validIssuerUri: <https://<auth-server-address>/auth/realms/external> 2
       jwksEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/certs> 3
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5
       tlsTrustedCertificates: 6
       - secretName: oauth-server-cert
         certificate: ca.crt
       disableTlsHostnameVerification: true 7
       jwksExpirySeconds: 360 8
       jwksRefreshSeconds: 300 9
       jwksMinRefreshPauseSeconds: 1 10
       enableECDSA: "true" 11

   1

   oauth に設定されたリスナータイプ。

   2

   認証に使用されるトークン発行者の URI。

   3

   ローカルの JWT 検証に使用される JWKS 証明書エンドポイントの URI。

   4

   トークンの実際のユーザー名が含まれるトークン要求 (またはキー)。ユーザー名は、ユーザーの識別に使用される principal です。userNameClaim の値は、使用される認証フローと承認サーバーによって異なります。

   5

   (任意設定): セッションの有効期限がアクセストークンと同じ期間になるよう強制する Kafka の再認証メカニズムを有効にします。指定された値がアクセストークンの有効期限が切れるまでの残り時間未満の場合、クライアントは実際にトークンの有効期限が切れる前に再認証する必要があります。デフォルトでは、アクセストークンの期限が切れてもセッションは期限切れにならず、クライアントは再認証を試行しません。

   6

   (任意設定): 承認サーバーへの TLS 接続用の信用できる証明書。

   7

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   8

   JWKS 証明書が期限切れになる前に有効であるとみなされる期間。デフォルトは 360 秒 です。デフォルトよりも長い時間を指定する場合は、無効になった証明書へのアクセスが許可されるリスクを考慮してください。

   9

   JWKS 証明書を更新する間隔。この間隔は、有効期間よりも 60 秒以上短くする必要があります。デフォルトは 300 秒 です。

   10

   JWKS 公開鍵の更新が連続して試行される間隔の最小一時停止時間 (秒単位)。不明な署名キーが検出されると、JWKS キーの更新は、最後に更新を試みてから少なくとも指定された期間は一時停止し、通常の定期スケジュール以外でスケジュールされます。キーの更新はエクスポネンシャルバックオフ (exponential backoff) のルールに従い、jwksRefreshSeconds に到達するまで、一時停止を増やして失敗した更新の再試行を行います。デフォルト値は 1 です。

   11

   (任意設定): ECDSA を使用して承認サーバーで JWT トークンを署名する場合は、これを有効にする必要があります。BouncyCastle 暗号ライブラリーを使用して追加の暗号プロバイダーがインストールされます。デフォルトは false です。

   例 2: イントロスペクションエンドポイントを使用したトークンの検証の設定

   - name: external
     port: 9094
     type: loadbalancer
     tls: true
     authentication:
       type: oauth
       validIssuerUri: <https://<auth-server-address>/auth/realms/external>
       introspectionEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token/introspect> 1
       clientId: kafka-broker 2
       clientSecret: 3
         secretName: my-cluster-oauth
         key: clientSecret
       userNameClaim: preferred_username 4
       maxSecondsWithoutReauthentication: 3600 5

   1

   トークンイントロスペクションエンドポイントの URI。

   2

   クライアントを識別するためのクライアント ID。

   3

   認証にはクライアントシークレットとクライアント ID が使用されます。

   4

   トークンの実際のユーザー名が含まれるトークン要求 (またはキー)。ユーザー名は、ユーザーの識別に使用される principal です。userNameClaim の値は、使用される承認サーバーによって異なります。

   5

   (任意設定): セッションの有効期限がアクセストークンと同じ期間になるよう強制する Kafka の再認証メカニズムを有効にします。指定された値がアクセストークンの有効期限が切れるまでの残り時間未満の場合、クライアントは実際にトークンの有効期限が切れる前に再認証する必要があります。デフォルトでは、アクセストークンの期限が切れてもセッションは期限切れにならず、クライアントは再認証を試行しません。

   OAuth 2.0 認証の適用方法や、承認サーバーのタイプによっては、追加 (任意) の設定を使用できます。

     # ...
     authentication:
       type: oauth
       # ...
       checkIssuer: false 1
       checkAudience: true 2
       fallbackUserNameClaim: client_id 3
       fallbackUserNamePrefix: client-account- 4
       validTokenType: bearer 5
       userInfoEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/userinfo 6
       enableOauthBearer: false 7
       enablePlain: true 8
       tokenEndpointUri: https://OAUTH-SERVER-ADDRESS/auth/realms/external/protocol/openid-connect/token 9
       customClaimCheck: "@.custom == 'custom-value'" 10

   1

   承認サーバーが iss クレームを提供しない場合は、発行者チェックを行うことができません。このような場合、checkIssuer を false に設定し、validIssuerUri を指定しないようにします。デフォルトは true です。

   2

   承認サーバーによって aud (オーディエンス) クレームが提供され、オーディエンスチェックを強制する場合は、checkAudience を true に設定します。オーディエンスチェックによって、トークンの目的の受信者が特定されます。そのため、Kafka ブローカーによって、aud クレームに clientId のないトークンは拒否されます。デフォルトは false です。

   3

   承認サーバーは、通常ユーザーとクライアントの両方を識別する単一の属性を提供しない場合があります。クライアントが独自の名前で認証される場合、サーバーによって クライアント ID が提供されることがあります。更新トークンまたはアクセストークンを取得するために、ユーザー名およびパスワードを使用してユーザーが認証される場合、サーバーによってクライアント ID の他に ユーザー名 が提供されることがあります。プライマリーユーザー ID 属性が使用できない場合は、このフォールバックオプションで、使用するユーザー名クレーム (属性) を指定します。

   4

   fallbackUserNameClaim が適用される場合、ユーザー名クレームの値とフォールバックユーザー名クレームの値が競合しないようにする必要もあることがあります。producer というクライアントが存在し、producer という通常ユーザーも存在する場合について考えてみましょう。この 2 つを区別するには、このプロパティーを使用してクライアントのユーザー ID に接頭辞を追加します。

   5

   (introspectionEndpointUri を使用する場合のみ該当): 使用している認証サーバーによっては、イントロスペクションエンドポイントによってトークンタイプ属性が返されるかどうかは分からず、異なる値が含まれることがあります。イントロスペクションエンドポイントからの応答に含まれなければならない有効なトークンタイプ値を指定できます。

   6

   (introspectionEndpointUri を使用する場合のみ該当): イントロスペクションエンドポイントの応答に識別可能な情報が含まれないように、承認サーバーが設定または実装されることがあります。ユーザー ID を取得するには、userinfo エンドポイントの URI をフォールバックとして設定します。userNameClaim、fallbackUserNameClaim、および fallbackUserNamePrefix 設定が userinfo エンドポイントの応答に適用されます。

   7

   これを false`to disable the OAUTHBEARER mechanism on the listener. At least one of PLAIN or OAUTHBEARER has to be enabled. Default is `true に設定します。

   8

   リスナーで PLAIN メカニズムを有効にするには、これを true に設定します。これは、すべてのプラットフォームのすべてのクライアントでサポートされます。Kafka クライアントで PLAIN メカニズムを有効にし、username および password を設定する必要があります。このメカニズムは、OAuth アクセストークンを使用するか、OAuth クライアント ID およびシークレット (クライアントクレデンシャル) を使用して、認証するために使用できます。クライアントによって、文字列 $accessToken: で始まる password が設定された場合、パスワードはサーバー上のアクセストークンとして解釈され、username はアカウントのユーザー名として解釈されます。それ以外の場合は、ユーザーはクライアント ID として解釈され、パスワードはクライアントシークレットとして解釈されます。デフォルトは false です。

   9

   これは、前のポイントで説明されているように、enablePlain が true に設定されている場合にクライアントクレデンシャル認証をサポートするように設定する必要があります。

   10

   これを JsonPath フィルタークエリーに設定すると、検証中に追加のカスタムルールを JWT アクセストークンに適用できます。アクセストークンに必要なデータが含まれていないと拒否されます。introspectionEndpointUri を使用する場合、カスタムチェックはイントロスペクションエンドポイントの応答 JSON に適用されます。

3. エディターを保存して終了し、ローリングアップデートの完了を待ちます。

4. ログで更新を確認するか、Pod の状態遷移を監視して確認します。

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   ローリングアップデートによって、ブローカーが OAuth 2.0 認証を使用するように設定されます。

手順

1. OAuth 2.0 サポートのあるクライアントライブラリーを Kafka クライアントの pom.xml ファイルに追加します。

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.7.1.redhat-00003</version>
   </dependency>

2. コールバックのシステムプロパティーを設定します。

   以下に例を示します。

   System.setProperty(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, “https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token”); 1
   System.setProperty(ClientConfig.OAUTH_CLIENT_ID, "<client-name>"); 2
   System.setProperty(ClientConfig.OAUTH_CLIENT_SECRET, "<client-secret>"); 3

   1

   承認サーバーのトークンエンドポイントの URI です。

   2

   クライアント ID。承認サーバーで client を作成するときに使用される名前です。

   3

   承認サーバーで client を作成するときに作成されるクライアントシークレット。

3. Kafka クライアント設定の TLS で暗号化された接続で SASL OAUTHBEARER メカニズムを有効にします。

   以下に例を示します。

   props.put("sasl.jaas.config", "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
   props.put("security.protocol", "SASL_SSL"); 1
   props.put("sasl.mechanism", "OAUTHBEARER");
   props.put("sasl.login.callback.handler.class", "io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler");

   1

   この例では、TLS 接続で SASL_SSL を使用します。暗号化されていない接続では SASL_PLAINTEXT を使用します。

4. Kafka クライアントが Kafka ブローカーにアクセスできることを確認します。

手順

1. クライアントシークレットを作成し、これを環境変数としてコンポーネントにマウントします。

   以下は、Kafka Bridge の Secret を作成する例になります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Secret
   metadata:
    name: my-bridge-oauth
   type: Opaque
   data:
    clientSecret: MGQ1OTRmMzYtZTllZS00MDY2LWI5OGEtMTM5MzM2NjdlZjQw 1

   1

   clientSecret キーは base64 形式である必要があります。

2. Kafka コンポーネントのリソースを作成または編集し、OAuth 2.0 認証が認証プロパティーに設定されるようにします。

   OAuth 2.0 認証では、以下を使用できます。

   • クライアント ID およびシークレット
   • クライアント ID および更新トークン
   • アクセストークン
   • TLS

   KafkaClientAuthenticationOAuth スキーマ参照は、それぞれの例を提供します。

   以下は、クライアント ID、シークレット、および TLS を使用して OAuth 2.0 が Kafka Bridge クライアントに割り当てられる例になります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     # ...
     authentication:
       type: oauth 1
       tokenEndpointUri: https://<auth-server-address>/auth/realms/master/protocol/openid-connect/token 2
       clientId: kafka-bridge
       clientSecret:
         secretName: my-bridge-oauth
         key: clientSecret
       tlsTrustedCertificates: 3
       - secretName: oauth-server-cert
         certificate: tls.crt

   1

   oauth に設定された認証タイプ。

   2

   認証用のトークンエンドポイントの URI。

   3

   承認サーバーへの TLS 接続用の信用できる証明書。

   OAuth 2.0 認証の適用方法や、承認サーバーのタイプによって、使用できる追加の設定オプションがあります。

   # ...
   spec:
     # ...
     authentication:
       # ...
       disableTlsHostnameVerification: true 1
       checkAccessTokenType: false 2
       accessTokenIsJwt: false 3
       scope: any 4

   1

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   2

   承認サーバーによって、JWT トークン内部で typ (タイプ) 要求が返されない場合は、checkAccessTokenType: false を適用するとトークンタイプがチェックされず次に進むことができます。デフォルトは true です。

   3

   不透明なトークンを使用している場合、アクセストークンが JWT トークンとして処理されないように accessTokenIsJwt: false を適用することができます。

   4

   (任意設定): トークンエンドポイントからトークンを要求するための scope。認証サーバーでは、クライアントによるスコープの指定が必要になることがあります。この場合では any になります。

3. Kafka リソースのデプロイメントに変更を適用します。

   oc apply -f your-file

4. ログで更新を確認するか、Pod の状態遷移を監視して確認します。

   oc logs -f ${POD_NAME} -c ${CONTAINER_NAME}
   oc get pod -w

   ローリングアップデートでは、OAuth 2.0 認証を使用して Kafka ブローカーと対話するコンポーネントが設定されます。