6.7. OAuth 2.0 トークンベース認証の使用

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. Kafka クライアントは、clientId をユーザー名として、secret をパスワードとして渡します。
2. Kafka ブローカーは、トークンエンドポイントを使用して clientId および secret を認可サーバーに渡します。
3. 認可サーバーは、新しいアクセストークンまたはエラー (クライアントクレデンシャルが有効でない場合) を返します。

4. Kafka ブローカーは、以下のいずれかの方法でトークンを検証します。

   1. トークンイントロスペクションエンドポイントが指定されている場合、Kafka ブローカーは認可サーバーでエンドポイントを呼び出すことで、アクセストークンを検証します。トークンの検証に成功した場合には、セッションが確立されます。
   2. ローカルトークンのイントロスペクションが使用される場合、要求は認可サーバーに対して行われません。Kafka ブローカーは、JWT トークン署名チェックを使用して、アクセストークンをローカルで検証します。

1. Kafka クライアントはユーザー名とパスワードを渡します。パスワードは、クライアントを実行する前に手動で取得および設定されたアクセストークンの値を提供します。

2. Kafka ブローカーリスナーが認証のトークンエンドポイントで設定されているかどうかに応じて、$accessToken: 文字列の接頭辞の有無にかかわらず、パスワードは渡されます。

   1. トークンエンドポイントが設定されている場合、パスワードの前に $accessToken: を付け、password パラメーターにクライアントシークレットではなくアクセストークンが含まれていることをブローカーに知らせる必要があります。Kafka ブローカーは、ユーザー名をアカウントのユーザー名として解釈します。
   2. トークンエンドポイントが Kafka ブローカーリスナーで設定されていない場合 (no-client-credentials mode を強制)、パスワードは接頭辞なしでアクセストークンを提供する必要があります。Kafka ブローカーは、ユーザー名をアカウントのユーザー名として解釈します。このモードでは、クライアントはクライアント ID およびシークレットを使用せず、password パラメーターは常に raw アクセストークンとして解釈されます。

3. Kafka ブローカーは、以下のいずれかの方法でトークンを検証します。

   1. トークンイントロスペクションエンドポイントが指定されている場合、Kafka ブローカーは認可サーバーでエンドポイントを呼び出すことで、アクセストークンを検証します。トークンの検証に成功した場合には、セッションが確立されます。
   2. ローカルトークンイントロスペクションが使用されている場合には、認可サーバーへの要求はありません。Kafka ブローカーは、JWT トークン署名チェックを使用して、アクセストークンをローカルで検証します。

手順

1. Red Hat Single Sign-On をインストールします。

   ZIP ファイルから、または RPM を使用してインストールできます。

2. Red Hat Single Sign-On 管理コンソールにログインして、Streams for Apache Kafka の 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 タブから、Streams for Apache Kafka のクラスター設定で使用するシークレットをメモします。

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

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

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

手順

1. server.properties ファイルで Kafka ブローカーリスナー設定を設定します。

   たとえば、OAUTHBEARER メカニズムを使用する場合:

   sasl.enabled.mechanisms=OAUTHBEARER
   listeners=CLIENT://0.0.0.0:9092
   listener.security.protocol.map=CLIENT:SASL_PLAINTEXT
   listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
   sasl.mechanism.inter.broker.protocol=OAUTHBEARER
   inter.broker.listener.name=CLIENT
   listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required ;
   listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

2. listener.name.client.oauthbearer.sasl.jaas.config の一部としてブローカーの接続設定を行います。

   この例では、接続設定オプションを示しています。

   例 1: JWKS エンドポイント設定を使用したローカルトークンの検証

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.valid.issuer.uri="https://<oauth_server_address>/auth/realms/<realm_name>" \
     oauth.jwks.endpoint.uri="https://<oauth_server_address>/auth/realms/<realm_name>/protocol/openid-connect/certs" \
     oauth.jwks.refresh.seconds="300" \
     oauth.jwks.refresh.min.pause.seconds="1" \
     oauth.jwks.expiry.seconds="360" \
     oauth.username.claim="preferred_username" \
     oauth.ssl.truststore.location="<path_to_truststore_p12_file>" \
     oauth.ssl.truststore.password="<truststore_password>" \
     oauth.ssl.truststore.type="PKCS12" ;
   listener.name.client.oauthbearer.connections.max.reauth.ms=3600000

   例 2: OAuth 2.0 イントロスペクションエンドポイントを使用した認可サーバーへのトークン検証の委任

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     oauth.introspection.endpoint.uri="https://<oauth_server_address>/auth/realms/<realm_name>/protocol/openid-connect/introspection" \
     # ...

3. 必要な場合は、認可サーバーへのアクセスを設定します。

   この手順は通常、サービスメッシュ などの技術がコンテナーの外部でセキュアなチャネルを設定するために使用される場合を除き、実稼働環境で必要になります。

   1. セキュアな認可サーバーに接続するためのカスタムトラストストアを指定します。認可サーバーへのアクセスには SSL が常に必要になります。

      プロパティーを設定して、トラストストアを設定します。

      以下に例を示します。

      listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        # ...
        oauth.client.id="kafka-broker" \
        oauth.client.secret="kafka-broker-secret" \
        oauth.ssl.truststore.location="<path_to_truststore_p12_file>" \
        oauth.ssl.truststore.password="<truststore_password>" \
        oauth.ssl.truststore.type="PKCS12" ;

   2. 証明書のホスト名がアクセス URL ホスト名と一致しない場合は、証明書のホスト名の検証をオフにできます。

      oauth.ssl.endpoint.identification.algorithm=""

      このチェックは、認可サーバーへのクライアント接続が認証されるようにします。実稼働以外の環境で検証をオフにすることもできます。

4. 選択した認証フローに従って追加のプロパティーを設定します。

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.token.endpoint.uri="https://<oauth_server_address>/auth/realms/<realm_name>/protocol/openid-connect/token" \ 1
     oauth.custom.claim.check="@.custom == 'custom-value'" \ 2
     oauth.scope="<scope>" \ 3
     oauth.check.audience="true" \ 4
     oauth.audience="<audience>" \ 5
     oauth.valid.issuer.uri="https://https://<oauth_server_address>/auth/<realm_name>" \ 6
     oauth.client.id="kafka-broker" \ 7
     oauth.client.secret="kafka-broker-secret" \ 8
     oauth.connect.timeout.seconds=60 \ 9
     oauth.read.timeout.seconds=60 \ 10
     oauth.http.retries=2 \ 11
     oauth.http.retry.pause.millis=300 \ 12
     oauth.groups.claim="$.groups" \ 13
     oauth.groups.claim.delimiter="," \ 14
     oauth.include.accept.header="false" ; 15

   1

   認可サーバーへの OAuth 2.0 トークンエンドポイント URL。実稼働環境の場合は、常に https:// urls を使用してください。KeycloakAuthorizer を使用する場合、またはブローカー間通信に OAuth 2.0 が有効なリスナーを使用する場合に必要です。

   2

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

   3

   (オプション) scope パラメーターがトークンエンドポイントに渡されます。スコープ は、ブローカー間認証用にアクセストークンを取得する場合に使用されます。また、clientId と secret を使用した PLAIN クライアント認証の上にある OAuth 2.0 のクライアント名にも使われています。これは、認可サーバーに応じて、トークンの取得機能とトークンの内容のみに影響します。リスナーによるトークン検証ルールには影響しません。

   4

   (オプション) オーディエンスチェック。認可サーバーが aud (オーディエンス) クレームを提供していて、オーディエンスチェックを実施する場合は、ouath.check.audience を true に設定します。オーディエンスチェックによって、トークンの目的の受信者が特定されます。これにより、Kafka ブローカーは aud 要求に clientId を持たないトークンを拒否します。デフォルトは false です。

   5

   (オプション) トークンエンドポイントに渡される audience パラメーター。オーディエンス は、ブローカー間認証用にアクセストークンを取得する場合に使用されます。また、clientId と secret を使用した PLAIN クライアント認証の上にある OAuth 2.0 のクライアント名にも使われています。これは、認可サーバーに応じて、トークンの取得機能とトークンの内容のみに影響します。リスナーによるトークン検証ルールには影響しません。

   6

   有効な発行者 URI。この発行者が発行するアクセストークンのみが受け入れられます。(常に必要です)

   7

   すべてのブローカーで同一の、Kafka ブローカーの設定されたクライアント ID。これは、kafka-broker として認可サーバーに登録されたクライアントです。イントロスペクションエンドポイントがトークンの検証に使用される場合、または KeycloakAuthorizer が使用される場合に必要です。

   8

   すべてのブローカーで同じ Kafka ブローカーに設定されたシークレット。ブローカーが認証サーバーに対して認証する必要がある場合は、クライアントシークレット、アクセストークン、またはリフレッシュトークンのいずれかを指定する必要があります。

   9

   (オプション) 認可サーバーへの接続時のタイムアウト (秒単位)。デフォルト値は 60 です。

   10

   (オプション): 認可サーバーへの接続時の読み取りタイムアウト (秒単位)。デフォルト値は 60 です。

   11

   認可サーバーへの失敗した HTTP リクエストを再試行する最大回数。デフォルト値は 0 で、再試行は実行されないことを意味します。このオプションを効果的に使用するには、oauth.connect.timeout.seconds オプションと oauth.read.timeout.seconds オプションのタイムアウト時間を短縮することを検討してください。ただし、再試行により現在のワーカースレッドが他のリクエストで利用できなくなる可能性があり、リクエストが多すぎると停止する場合、Kafka ブローカーが応答しなくなる可能性があることに注意してください。

   12

   認可サーバーへの失敗した HTTP リクエストの再試行を行うまでの待機時間。デフォルトでは、この時間はゼロに設定されており、一時停止は適用されません。これは、リクエストの失敗の原因となる問題の多くは、リクエストごとのネットワークの不具合やプロキシーの問題であり、すぐに解決できるためです。ただし、認可サーバーに負荷がかかっている場合、または高トラフィックが発生している場合は、このオプションを 100 ミリ秒以上の値に設定して、サーバーの負荷を軽減し、再試行が成功する可能性を高めることができます。

   13

   JWT トークンまたはイントロスペクションエンドポイントの応答からグループ情報を抽出するために使用される JsonPath クエリー。デフォルトでは設定されません。これは、カスタムオーソライザーがユーザーグループに基づいて認可の決定を行うために使用できます。

   14

   1 つのコンマ区切りの文字列として返されるときにグループ情報を解析するために使用される区切り文字。デフォルト値は ',' (コンマ) です。

   15

   (オプション) oauth.include.accept.header を false に設定して、リクエストから Accept ヘッダーを削除します。ヘッダーを含めることで認可サーバーとの通信時に問題が発生する場合は、この設定を使用できます。

5. OAuth 2.0 認証の適用方法、および使用されている認証サーバーのタイプに応じて、設定を追加します。

   listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
     # ...
     oauth.check.issuer=false \ 1
     oauth.fallback.username.claim="<client_id>" \ 2
     oauth.fallback.username.prefix="<client_account>" \ 3
     oauth.valid.token.type="bearer" \ 4
     oauth.userinfo.endpoint.uri="https://<oauth_server_address>/auth/realms/<realm_name>/protocol/openid-connect/userinfo" ; 5

   1

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

   2

   認可サーバーは、通常ユーザーとクライアントの両方を識別する単一の属性を提供しない場合があります。クライアントが独自の名前で認証される場合、サーバーによって クライアント ID が提供されることがあります。リフレッシュトークンまたはアクセストークンを取得するために、ユーザー名およびパスワードを使用してユーザーが認証される場合、サーバーによってクライアント ID の他に ユーザー名 が提供されることがあります。プライマリーユーザー ID 属性が使用できない場合は、このフォールバックオプションで、使用するユーザー名クレーム (属性) を指定します。必要に応じて、"['client.info'].['client.id']" のような JsonPath 式を使用して、トークン内のネストされた JSON 属性からフォールバックユーザー名を取得できます。

   3

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

   4

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

   5

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

手順

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

   <dependency>
    <groupId>io.strimzi</groupId>
    <artifactId>kafka-oauth-client</artifactId>
    <version>0.15.0.redhat-00007</version>
   </dependency>

2. プロパティーファイルで以下の設定を指定して、クライアントプロパティーを設定します。

   • セキュリティープロトコル
   • SASL メカニズム

   • 使用されているメソッドに応じた JAAS モジュールと認証プロパティー

     たとえば、以下を client.properties ファイルに追加できます。

     クライアントクレデンシャルメカニズムのプロパティー

     security.protocol=SASL_SSL 1
     sasl.mechanism=OAUTHBEARER 2
     ssl.truststore.location=/tmp/truststore.p12 3
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \ 4
       oauth.client.id="<client_id>" \ 5
       oauth.client.secret="<client_secret>" \ 6
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \ 7
       oauth.ssl.truststore.password="$STOREPASS" \ 8
       oauth.ssl.truststore.type="PKCS12" \ 9
       oauth.scope="<scope>" \ 10
       oauth.audience="<audience>" ; 11
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     TLS 暗号化接続用の SASL_SSL セキュリティープロトコル。ローカル開発のみでは、暗号化されていない接続で SASL_PLAINTEXT を使用します。

     2

     OAUTHBEARER または PLAIN として指定された SASL メカニズム。

     3

     Kafka クラスターへのセキュアなアクセスのためのトラストストア設定。

     4

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

     5

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

     6

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

     7

     この場所には、認可サーバーの公開鍵証明書 (truststore.p12) が含まれています。

     8

     トラストストアにアクセスするためのパスワード。

     9

     トラストストアのタイプ。

     10

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

     11

     (オプション) トークンエンドポイントからトークンを要求するための audience。認可サーバーでは、クライアントによるオーディエンスの指定が必要になることがあります。

     パスワード付与メカニズムのプロパティー

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.password.grant.username="<username>" \ 3
       oauth.password.grant.password="<password>" \ 4
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" \
       oauth.scope="<scope>" \
       oauth.audience="<audience>" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

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

     2

     (オプション) 認可サーバーで クライアント を作成するときに作成されるクライアントシークレット。

     3

     パスワードグラント認証のユーザー名。OAuth パスワードグラント設定 (ユーザー名とパスワード) では、OAuth 2.0 パスワードグラント方式が使用されます。パスワードグラントを使用するには、権限が制限されたクライアントのユーザーアカウントを認可サーバー上に作成します。アカウントは、サービスアカウントのように機能する必要があります。認証にユーザーアカウントが必要な環境で使用しますが、先にリフレッシュトークンの使用を検討してください。

     4

     パスワード付与認証のパスワード。

     注記

     SASL PLAIN は、OAuth 2.0 パスワード付与メソッドを使用したユーザー名とパスワードの受け渡し (パスワード付与) をサポートしていません。

     アクセストークンのプロパティー

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.access.token="<access_token>" \ 1
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

     Kafka クライアントの有効期間が長いアクセストークン。

     トークンのプロパティーを更新する

     security.protocol=SASL_SSL
     sasl.mechanism=OAUTHBEARER
     ssl.truststore.location=/tmp/truststore.p12
     ssl.truststore.password=$STOREPASS
     ssl.truststore.type=PKCS12
     sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
       oauth.token.endpoint.uri="<token_endpoint_url>" \
       oauth.client.id="<client_id>" \ 1
       oauth.client.secret="<client_secret>" \ 2
       oauth.refresh.token="<refresh_token>" \ 3
       oauth.ssl.truststore.location="/tmp/oauth-truststore.p12" \
       oauth.ssl.truststore.password="$STOREPASS" \
       oauth.ssl.truststore.type="PKCS12" ;
     sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

     1

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

     2

     (オプション) 認可サーバーで クライアント を作成するときに作成されるクライアントシークレット。

     3

     Kafka クライアントの有効期間が長い更新トークン。

3. OAUTH 2.0 認証のクライアントプロパティーを Java クライアントコードに入力します。

   クライアントプロパティーの入力を示す例

   Properties props = new Properties();
   try (FileReader reader = new FileReader("client.properties", StandardCharsets.UTF_8)) {
     props.load(reader);
   }

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