1. Home
2. Platform
3. 管理
4. Metadata Service（MDS）

Kafka ブローカーでの mTLS 認証と RBAC の構成

この構成では、Confluent Metadata Service (MDS) を介して相互 TLS（mTLS）認証と role-based access control (RBAC) を使用して Kafka ブローカーを構成する方法を示します。mTLS は、クライアントと MDS との間のトラフィックが安全であり、双方向からのコンテンツを信頼できることを保証するための双方向認証を提供します。この例では、以下のように構成します。

• Kafka と通信するクライアントは、mTLS（暗号化と認証）を使用できます。
• MDS と通信するクライアントは、TLS（暗号化）と HTTP 基本認証を使用します。

このウォークスルーは、認証と RBAC の他の組み合わせを構成するためのテンプレートとして使用できます。一般的な方法は次のとおりです。

1. 認証用のプロパティを構成します。構成済みの場合は、オーソライザーの構成ステップに進みます。
2. 使用するオーソライザーを決定します（この場合は、RBAC と一元的な ACL のサポートを提供する ConfluentServerAuthorizer）。認可は authorizer.class.name で定義します。
3. MDS を有効化します。
4. ディレクトリサービス（LDAP など）を追加します。

用語

以下は、用語の簡単な紹介です。これらのテクノロジーの詳細については、ブログ投稿「Apache Kafka Security 101」を参照してください。

認証

認証とは、本人確認のためのセキュリティメカニズムです。プラットフォームリソースにアクセスしようとしているプリンシパルを、さまざまなセキュリティプロトコルで認証（識別）できます。

• Kafka ブローカーのコンテキストで、PLAINTEXT とは、認証と暗号化のどちらも行わないことを意味します。データは "平文"（プレーンテキスト、クリアテキスト）で送信されます。

• Secure Sockets Layer（SSL）とその変形である Transport Layer Security（TLS）は、エンティティ間の暗号化された通信を保護するためのプロトコルです。Kafka では（Java と同様に）構成とコードで SSL という用語が使用されます。SSL は、暗号化のみ、または暗号化と双方向認証（mTLS）用に構成できます。この場合、証明書を生成し、認証機関を使用します。mTLS は、SSL を使用した認証です。SSL の詳細については、「TLS/SSL による暗号化」および「TLS/SSL での暗号化と認証」を参照してください。

  プロトコル	提供されるセキュリティ	実装
  SSL	暗号化	TLS
  SSL	暗号化 + 認証	mTLS

• SASL は汎用的な認証用プロトコルであり、同じポート（9092 など）で Kerberos、PLAIN、SCRAM などの複数のメカニズムと実装をサポートできます。SASL の詳細については、「JAAS を使用した SASL による認証」を参照してください。

  プロトコル	提供されるセキュリティ
  SASL（PLAIN）	認証（暗号化なし）
  SASL（認証）、SSL（暗号化）	暗号化 + 認証

  ちなみに

  • SASL と mTLS には別々のポートを使用する必要があります。これらを同じポートで実行することはできません。
  • SASL のメカニズム "PLAIN" は、SSL の "PLAINTEXT" と同じではありません。詳細については、「PLAIN の構成」を参照してください。

認可

認可は、xyz （アクション、リソース）に対するアクセス許可があるかどうかを確認するためのセキュリティメカニズムです。RBAC はクライアント権限を決定し、付与されたロールに基づいてきめ細かいアクセス制御を有効にします。RBAC は ACL と組み合わせて使用できます。

概要と前提条件

ここに示す例とウォークスルーでは、前提条件として以下が必要になります。

• Confluent Platform
• Confluent LDAP Authorizer を構成する対象となる LDAP システムへのアクセス
• セキュリティ証明書とキーの作成 ができること。詳細について説明するセクションで、スクリプトに関する推奨事項を紹介しています。

構成の一環として、ここで説明されているタスクを実行するには、次のことを行います。

• 対応するクライアント証明書を提供するクライアントにのみアクセスが許可されるように、 SSL キーと証明書を作成 します。ウォークスルーのセクションで、スクリプトに関する推奨事項を紹介しています。
• MDS を構成 します。
• Kafka ブローカーのユーザーを作成します。
• MDS トークンサービス用に PEM キーペアの作成 を行います。

構成の概要

以下の例では、単一の etc/kafka/server.properties ファイル、つまり単一のブローカー構成で、mTLS による認証と MDS を介した RBAC による認可の構成を示します。

ちなみに

Kafka 構成ファイルはブローカーごとに必要です。たとえば、3 つのブローカーを作成するには、3 つの構成ファイルを作成します。

mTLS による認証では、次の 3 つのポートを使用します。

• ブローカー間通信用に mTLS を使用するポート
• 外部クライアント用に mTLS を使用するポート
• トークンサービス用のポート（MDS による権限借用を有効化）

MDS HTTPS 接続にはポート 8090 が使用されます。

ちなみに

技術的には、ブローカーを実行するために必要なポートは 1 つだけです。

• Kafka では、クライアント接続とブローカー間通信のために 1 つ以上の開いているポートが必要になります。このポートには mTLS が必要です。
• 運用上の利便性を考えて、例には mTLS を使用する 2 番目のポートが示されています。これにより、異なるホスト名（internal と external など）をリッスンするように構成できるため、トラフィック制御が向上します。
• 3 番目のトークンサービス用ポートも厳密には必須ではありませんが、これを使用すると MDS による権限借用が可能になるため、RBAC のフル機能を使用するためには必要になります。SASL では、クライアントが開始するメカニズムのネゴシエーションが可能です。つまり、1 つの SASL リスナーで複数のメカニズムをサポートでき、どちらを使用するかがクライアントからブローカーに指示されます。

各行の説明は例の最後に示されています。

注釈

ほとんどの構成属性では、値の例が <> 内に示されています。これは、想定される値の型を理解するうえで役立ちます。値の例は、それぞれのセットアップに応じた値に置き換えてください。<> を使用せずに示されている値は、推奨値として使用できます。

############################# Broker Settings ##################################
zookeeper.connect=<host-1>:2181,<host-2>:2181,<host-3>:2181
log.dirs=/var/lib/kafka/data
broker.id=1

############################# Group Coordinator Settings #######################
group.initial.rebalance.delay.ms=0

############################# Log Retention Policy, Log Basics ##################
log.retention.check.interval.ms=300000
log.retention.hours=168
log.segment.bytes=1073741824
num.io.threads=16
num.network.threads=8
num.partitions=1
num.recovery.threads.per.data.dir=2

########################### Socket Server Settings #############################
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400

############################# Internal Topic Settings  #########################
offsets.topic.replication.factor=3
transaction.state.log.min.isr=2
transaction.state.log.replication.factor=3

######################## Metrics Reporting ########################################
metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
confluent.metrics.reporter.bootstrap.servers=<address>-west-2.compute.internal:9092
confluent.metrics.reporter.topic.replicas=3
confluent.support.metrics.enable=true
confluent.support.customer.id=anonymous

######################## LISTENERS ######################################
listeners=INTERNAL://:9092,EXTERNAL://:9093,TOKEN://:9094
advertised.listeners=INTERNAL://<localhost>:9092,\
                     EXTERNAL://<external-hostname>:9093,\
                     TOKEN://<external-hostname>:9094
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL,TOKEN:SASL_SSL

inter.broker.listener.name=INTERNAL

############################ SSL SETTINGS #####################################
ssl.truststore.location=/var/ssl/private/client.truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/var/ssl/private/kafka.keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.client.auth=required
ssl.endpoint.identification.algorithm=HTTPS

##############  SSL settings for metrics reporting ##############
confluent.metrics.reporter.security.protocol=SSL
confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/client.truststore.jks
confluent.metrics.reporter.ssl.truststore.password=<truststore-password>
confluent.metrics.reporter.ssl.keystore.location=/var/ssl/private/kafka.keystore.jks
confluent.metrics.reporter.ssl.keystore.password=<keystore-password>
confluent.metrics.reporter.ssl.key.password=<key-password>

############################# SSL LISTENERS #############################
listener.name.internal.ssl.principal.mapping.rules= \
        RULE:^CN=([a-zA-Z0-9.]*).*$/$1/L ,\
        DEFAULT

listener.name.external.ssl.principal.mapping.rules= \
        RULE:^CN=([a-zA-Z0-9.]*).*$/$1/L ,\
        DEFAULT

############################# TOKEN LISTENER #############################
listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.token.oauthbearer.sasl.jaas.config= \
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        publicKeyPath="<path-to-mds-public-key.pem>";
listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler

############################# Authorization Settings #############################
authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
confluent.authorizer.access.rule.providers=ZK_ACL,CONFLUENT
super.users=User:kafka

#############################  MDS Listener - which port to listen on #############################
confluent.metadata.server.listeners=https://0.0.0.0:8090,http://0.0.0.0:8091
confluent.metadata.server.advertised.listeners=https://<localhost>:8090,\
                                               http://<localhost>:8091

############################# SSL Settings for MDS #############################
confluent.metadata.server.ssl.keystore.location=<path-to-kafka.keystore.jks>
confluent.metadata.server.ssl.keystore.password=<keystore-password>
confluent.metadata.server.ssl.key.password=<key-password>
confluent.metadata.server.ssl.truststore.location=<path-to-client.truststore.jks>
confluent.metadata.server.ssl.truststore.password=<truststore-password>

############################# MDS Token Service Settings - enable token generation #############################
confluent.metadata.server.token.max.lifetime.ms=3600000
confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
confluent.metadata.server.token.signature.algorithm=RS256
confluent.metadata.server.authentication.method=BEARER

############################# Identity Provider Settings(LDAP - local OpenLDAP) #############################
ldap.java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory
ldap.com.sun.jndi.ldap.read.timeout=3000
ldap.java.naming.provider.url=ldap:<ldap-server-address>
# how mds authenticates to ldap server
ldap.java.naming.security.principal=<CN=mds,CN=Demo,DC=confluent,DC=io>
ldap.java.naming.security.credentials=<password>
ldap.java.naming.security.authentication=simple
# ldap search mode (GROUPS is default)
#ldap.search.mode=GROUPS
#ldap.search.mode=USERS
# how to search for users
ldap.user.search.base=<CN=Demo,DC=confluent,DC=io>
# how to search for groups
ldap.group.search.base=<CN=Demo,DC=confluent,DC=io>
# which attribute in ldap record corresponds to user name
ldap.user.name.attribute=sAMAccountName
ldap.user.memberof.attribute.pattern=<CN=(.*),CN=Demo,DC=confluent,DC=io>
ldap.group.object.class=group
ldap.group.name.attribute=sAMAccountName
ldap.group.member.attribute.pattern=<CN=(.*),CN=Demo,DC=confluent,DC=io>

########################### Enable Swagger #############################
confluent.metadata.server.openapi.enable=true

Copy

各構成セクションの説明（行番号を参照）

ブローカー、Kafka リスナー、認証、認可メカニズムの構成

• 行 36 ～ 42: "LISTENERS" では、3 つの Kafka リスナーを定義し、リスナーが使用するセキュリティプロトコルに合わせてブローカー用に HTTPS を構成しています。「リスナーの構成」を参照してください。
• 行 45 ～ 51: "SSL SETTINGS" では、共有してすべてのリスナーに適用できる Kafka の最上位の SSL 設定を定義しています。これより細かくエンティティごとに SSL 設定を定義するには、プレフィックスを付けた構成のセットを複数使用し、それぞれに別々の証明書を生成します。最初の 4 行では、SSL 証明書を生成するために必要な暗号化を定義し、残り 3 行では認証を指定しています。「Kafka の SSL 設定」を参照してください。
• 行 54 ～ 59: "SSL SETTINGS" のサブセクション "SSL settings for metrics reporting" には、メトリクスをモニタリングするための SSL のセットアップ例が示されています。「Kafka の SSL 設定」を参照してください。
• 行 62 ～ 68: "SSL LISTENERS" では、証明書のプリンシパルの識別名（dname ）に正規表現（regex）を適用して、マシン名の先頭の CN （common name: 一般名）のみを抽出しています。「SSL リスナー用プリンシパルのマッピングルール（証明書からのプリンシパルの抽出）」を参照してください。
• 行 71 ～ 76: "TOKEN LISTENER" では、Kafka に別のリスナーを構成しています。このリスナーは、OAUTHBEARER を使用して MDS で権限を借用するためのトークンを受け入れます。「トークンベースの認証の有効化」を参照してください。
• 行 79 ～ 81: "Authorization Settings" では、認可のメカニズムを ConfluentServerAuthorizer として定義しています。これには、RBAC と一元的な ACL のサポートも含まれています。「オーソライザーの構成」を参照してください。

MDS の構成

• 行 84 ～ 86: "MDS Listener" では、HTTP プロトコルと SSL 暗号化（HTTPS）を使用してクライアントと通信する MDS リスナーを定義しています。「MDS 用の HTTPS リスナーの指定」を参照してください。
• 行 89 ～ 93: "SSL Settings for MDS" では、MDS HTTPS リスナーにマップする暗号化と認証を定義しています。「MDS 用の SSL 設定」を参照してください。
• 行 96 ～ 99: MDS は、クライアントの認証情報と引き換えにトークンを提供するだけでなく、クライアントからトークンを受け入れて認証することもできます。「MDS トークンの生成の有効化と構成」を参照してください。"MDS Token Service Settings" では、トークンサービスを有効にして構成し、トークンの有効期間を設定しています。また、MDS が実行されているマシンのパブリック/プライベートキーパスと暗号化アルゴリズムを指定しています。「MDS トークンの生成の有効化と構成」を参照してください。
• 行 102 ～ 121: "Identity Provider Settings" では、 LDAP を RBAC が使用できるように構成しています。「LDAP 設定の構成」を参照してください。

オプションのテストとトラブルシューティング

• 行 124: "Enable Swagger" で、Swagger UI を使用して Kafka への呼び出しをテストする場合は、confluent.metadata.server.openapi.enable を true に設定します。これは、この RBAC を使用した mTLS 構成例に固有の設定ではありません。

  注釈

  本稼働環境では Swagger UI を有効にしないでください。HTTP に制限されているため、HTTPS またはカスタム証明書では機能しません。ただし、初期セットアップやテストには役立つ可能性があります。

実行方法

1. ZooKeeper を起動します。

   sudo <path-to-confluent>/bin/zookeeper-server-start ./etc/kafka/zookeeper.properties
   

   Copy

2. 構成した etc/kafka/server.properties を使用して Kafka を起動します。

   <path-to-confluent>/bin/kafka-server-start <path-to-confluent>/etc/kafka/server.properties
   

   Copy

詳細については、「Confluent Platform の起動」と、 Confluent Platform をインストールして実行する方法 を参照してください。

Schema Registry、Control Center、ksqlDB、Connect などのクライアントは、HTTPS を介して Kafka および MDS と通信できるように、それぞれのプロパティファイルで構成できます。

mTLS および RBAC の構成の詳細

このセクションでは、上の 概要 のセクションに記載されている構成について、詳しく説明します。

Kafka サーバープロパティファイル

このウォークスルーの主な焦点は、./etc/kafka/<server.properties> で定義されている構成です。すべての構成をセットアップした後、プロパティファイルを使用して、Kafka に対する次のコマンドを実行します。

./bin/kafka-server-start ./etc/kafka/<server.properties>

Copy

リスナーの構成

<NAME> ://<PORT> という形式を使用して、Kafka でリッスンするポートを定義します。

デフォルトでは、9092 に 1 つの Kafka リスナーが提供されます。

少なくともこのリスナーが必要であり、インターネットを介したクライアント用にはアドバタイズされたリスナーが必要です。

######################## LISTENERS ######################################
listeners=INTERNAL://:9092,EXTERNAL://:9093,TOKEN://:9094
advertised.listeners=INTERNAL://<localhost>:9092,\
                     EXTERNAL://<external-hostname>:9093,\
                     TOKEN://<external-hostname>:9094
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL,TOKEN:SASL_SSL

inter.broker.listener.name=INTERNAL

Copy

ちなみに

• デフォルトで、Kafka では 9092 が使用されます。
• システム上の任意の空きポートを使用し、必要に応じてカスタム名を付けることもできます。
• 以前は名前とセキュリティプロトコルが一致する必要がありましたが、これは必須ではなくなりました。

listeners - 同じネットワーク上の他のクラウドマシンで使用可能なローカルポート。この例では、わかりやすい名前で 3 つのポートを定義しています。

advertised listeners - 名前を使ってリスナーに設定され、アドバタイズされたポートは、インターネットを介してサーバーとやり取りするクライアント用です。クライアントからやり取りの方法を尋ねられた場合は、ホストアドレス（アドバタイズされたもの）とポートを提供します。

listener.security.protocol.map - リスナー名をセキュリティプロトコルにマップします

• INTERNAL:SSL - SSL 認証
• EXTERNAL:SSL - SSL 認証
• TOKEN:SASL_SSL - SASL 認証と SSL 暗号化

Kafka は起動すると、指定されたポートを取得し、そのポートをリッスンします。プロデューサーやコンシューマーなどの Kafka クライアントは、これらのポートから Kafka に到達できます。

Kafka の SSL 設定

ちなみに

SSL の構成の詳細については、「TLS/SSL による暗号化」および「TLS/SSL での暗号化と認証」を参照してください。

SSL を構成するには、次の手順に従います。

1. 証明書を作成します。
2. プロパティファイルで SSL を構成します。

以下に示されている SSL 設定は、暗号化と認証を定義する Kafka の最上位構成であり、個々のリスナーを指すプレフィックスが付いていないため、各種のリスナーに適用されます。

############################ SSL SETTINGS #####################################
ssl.truststore.location=/var/ssl/private/client.truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/var/ssl/private/client.keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.client.auth=required
ssl.endpoint.identification.algorithm=HTTPS

Copy

リスナーごとに個別の証明書を生成すると SSL を個別に構成できますが、この例では行っていません。

これらの SSL 設定は、MDS 固有のプレフィックスが付いた SSL 設定を構成しない場合、MDS でも共有できます。

リスナーに固有の設定を定義する場合は、プレフィックス listener.name.<name>.<SETTING>=<value> を以下のフォーマットで使用します。たとえば、INTERNAL の値を指定する場合は、次のようになります。

listener.name.internal.ssl.truststore.location=/var/ssl/private/client.truststore.jks
listener.name.internal.ssl.truststore.password=<truststore-password>
listener.name.internal.ssl.keystore.location=/var/ssl/private/client.keystore.jks
listener.name.internal.ssl.keystore.password=<keystore-password>
listener.name.internal.ssl.key.password=<key-password>
listener.name.internal.ssl.client.auth=required
listener.name.internal.ssl.endpoint.identification.algorithm=HTTPS

Copy

このセクションには、メトリクスレポート用の SSL 設定も含まれています。これらは、この RBAC 使用した mTLS 構成例に固有の設定ではありません。ここに含まれているのは、confluent.metrics.reporter のようなプレフィックスを ssl.truststore.xxx に追加することで、さまざまなサービスに固有の SSL 設定を構成できることを示すためです。メトリクスレポートと SSL の詳細については、「Confluent Metrics Reporter」および「SSL の構成」を参照してください。

confluent.metrics.reporter.security.protocol=SSL
confluent.metrics.reporter.ssl.truststore.location=/var/ssl/private/client.truststore.jks
confluent.metrics.reporter.ssl.truststore.password=<truststore-password>
confluent.metrics.reporter.ssl.keystore.location=/var/ssl/private/client.keystore.jks
confluent.metrics.reporter.ssl.keystore.password=<keystore-password>
confluent.metrics.reporter.ssl.key.password=<key-password>

Copy

暗号化

これらの設定（前述の設定のサブセット）では、暗号化を定義しています。これに対して、SSL 証明書を生成する必要があります。

ssl.truststore.location=/var/ssl/private/client.truststore.jks
ssl.truststore.password=<truststore-password>
ssl.keystore.location=/var/ssl/private/client.keystore.jks
ssl.keystore.password=<keystore-password>

Copy

証明書を生成するためのデモスクリプト

証明書を生成するためのデモスクリプトは、こちら から入手できます。スクリプトを自分の環境に合わせるには、デプロイに一致するようにマシンまたは MacBook の名前を変更します。

このスクリプトでは、どの証明書を信頼できるかを定義する "トラストストア" と、自分が誰であるかを定義する "キーストア" が生成されます。2 つのエンティティが相互に通信しているときは常に、ID 確認のためにこれらの証明書が使用されます。

クライアント truststore.jks および client.keystore.jks を生成し、これらのファイルをマシン上の任意の場所（ローカル/オンプレミスまたはクラウドプラットフォーム）に配置して、暗号化セクションで場所を指定します。

証明書とキーを生成するための別のデモスクリプトが、「SSL キーと証明書の作成」で紹介されています。

認証

以下の設定では、認証を定義しています。

ssl.key.password=<key-password>
ssl.client.auth=required
ssl.endpoint.identification.algorithm=HTTPS

Copy

SSL リスナー用プリンシパルのマッピングルール（証明書からのプリンシパルの抽出）

SSL 証明書では、SSL ユーザー名である LDAP 識別名（dname）の値を CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown の形式で定義します。

証明書を生成する際には、dname の CN （common name: 一般名）の値を決定します。

この複雑な名前から ACL またはロールバインディングを作成することは容易ではないため、サポートされているルールを使用して、証明書内の識別名の一部のみを抽出することで、ACL または ロールバインディングを作成できます。

Kafka サーバープロパティファイルの "SSL LISTENERS" では、指定された正規表現（regex）をプリンシパルに適用することで、証明書に保管されている dname からマシン名として最初の CN のみを抽出します（この例では internal と external）。

############################# SSL LISTENERS #############################
listener.name.internal.ssl.principal.mapping.rules= \
        RULE:^CN=([a-zA-Z0-9.]*).*$/$1/L ,\
        DEFAULT

listener.name.external.ssl.principal.mapping.rules= \
        RULE:^CN=([a-zA-Z0-9.]*).*$/$1/L ,\
        DEFAULT

Copy

正規表現の仕組み

デフォルトでは、SSL のユーザー名は CN=writeuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown の形式です。この構成では、X.500 識別名（DN）を短い名前にマッピングするためのルールのリストを利用できます。ルールは順番に評価され、DN と一致する最初のルールが、短い名前へのマッピングに使用されます。リスト内のそれ以降のルールはすべて無視されます。

ssl.principal.mapping.rules のフォーマットは、"RULE:" で始まり、次のフォーマットの式が続く、ルールのリストです。デフォルトのルールは、X.500 証明書 DN の文字列表現を返します。DN がパターンと一致する場合、置換コマンドがその名前に対して実行されます。このコマンドは小文字と大文字のオプションもサポートしており、変換結果を強制的にすべて小文字または大文字に置き換えます。この処理は、ルールの末尾に "/L" または "/U" を追加することによって実行されます。

RULE:pattern/replacement/
RULE:pattern/replacement/[LU]

Copy

たとえば、ssl.principal.mapping.rules の値は次のとおりです。

RULE:^CN=(.*?),OU=ServiceUsers.*$/$1/,
RULE:^CN=(.*?),OU=(.*?),O=(.*?),L=(.*?),ST=(.*?),C=(.*?)$/$1@$2/L,
RULE:^.*[Cc][Nn]=([a-zA-Z0-9.]*).*$/$1/L,
DEFAULT

Copy

これらのルールは DN CN=serviceuser,OU=ServiceUsers,O=Unknown,L=Unknown,ST=Unknown,C=Unknown を serviceuser に、 CN=adminUser,OU=Admin,O=Unknown,L=Unknown,ST=Unknown,C=Unknown を adminuser@admin に変換します。

トークンベースの認証の有効化

Kafka に別のリスナーを構成します。このリスナーは、MDS で権限を借用するためのトークンを受け入れます。この例では、リスナーを "token" と呼んでいますが、任意の名前を付けることができます。

これは、クライアントに代わって MDS で Kafka に対する認証を行うことが目的です。

• Kafka は MDS と通信し、MDS からトークンを取得します。
• Kafka はトークンを "token" リスナーに渡します。
• リスナーはトークンを検証し、認証されたクライアントであることを確認します。

############################# TOKEN LISTENER #############################
listener.name.token.sasl.enabled.mechanisms=OAUTHBEARER
listener.name.token.oauthbearer.sasl.jaas.config= \
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
        publicKeyPath="<path-to-mds-public-key.pem>";
listener.name.token.oauthbearer.sasl.server.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerValidatorCallbackHandler
listener.name.token.oauthbearer.sasl.login.callback.handler.class=io.confluent.kafka.server.plugins.auth.token.TokenBearerServerLoginCallbackHandler

Copy

オーソライザーの構成

Kafka ではオーソライザーが実行されます。

これにより認可メカニズムが実装されます。たとえば、"Bob" など特定のプリンシパルがなんらかの対象にアクセスできるかどうかが決定されます。

############################# Authorization Settings #############################
authorizer.class.name=io.confluent.kafka.security.authorizer.ConfluentServerAuthorizer
confluent.authorizer.access.rule.providers=ZK_ACL,CONFLUENT
super.users=User:kafka

Copy

オーソライザークラス

中心となるプロパティは、使用するオーソライザーを定義する authorizer.class.name です。

• RBAC オーソライザー : ConfluentServerAuthorizer （「Confluent Server Authorizer の構成」を参照）
• ACL オーソライザー : kafka.security.auth.SimpleAclAuthorizer （「ACL を使用した認可」を参照」）
• LDAP オーソライザー : io.confluent.kafka.security.ldap.authorizer.LdapAuthorizer は RBAC オーソライザーより優先され、ACL を使用してユーザーの認可を行うことができます。また、LDAP からグループの解決を行い、ユーザーが承認されたグループのメンバーであるかどうかを確認します（「LDAP Authorizer の構成」を参照）。

ルールプロバイダー

confluent.authorizer.access.rule.providers=ZK_ACL,CONFLUENT

ちなみに

リリース Confluent Platform 5.4 以降、ルールプロバイダーの有効な値として、ACL に代わって ZK_ACL、RBAC に代わって CONFLUENT が使用されます。

この設定では、ACL ルールプロバイダーを指定します。いずれか一方または両方を有効にすることができます。

• CONFLUENT - ロールバインディングを有効にし、RBAC の一部ではなく、MDS 経由でも管理される一元的な ACL を有効にします。Kafka に保管されます。
• ZK_ACL - RBAC 以前の ACL です。「ACL を使用した認可」を参照してください。

スーパーユーザー

どのユーザーがスーパーユーザーであるかを定義します。

super.users=User:kafka

Copy

super.users 構成では、どのユーザーにグローバルなアクセス許可があるかを定義します。ユーザーがスーパーユーザーのアクセス許可を持っている場合、MDS によるそれ以上の確認は必要なく、そのユーザーはどのような操作でも行うことができます。super.users には、ブローカー間通信に使用されるプリンシパルを 1 つ以上含める必要があります。オプションで、ブローカーノードごとに個別のプリンシパルを定義することもできます。

この例では、単一のスーパーユーザー kafka がすべてのブローカー間通信に使用されています。"kafka" というユーザーがブローカーに対して認証を行うたびに、オーソライザーは "kafka" に特定の操作を実行するためのアクセス許可があるかどうかを確認します。ただし、何よりも先に、"kafka" がスーパーユーザーのリストに含まれているかどうかの確認が行われます。これは、Kafka が内部 SSL ポートを介して他のブローカーと通信するために使用している SSL 証明書内のユーザーです。

ここでのユーザー名は、前に説明した certs-create.sh 証明書スクリプトから生成されたものです（「Kafka の SSL 設定」のデモスクリプトを参照）。この例に一致させるには、Kafka の証明書を生成するときに、dname の CN として kafka を定義します（-dname "CN=kafka,OU=QE,O=CONFLUENT,L=PaloAlto,ST=Ca,C=US"）。

スーパーユーザー については、ACL のドキュメントも参照してください。

MDS 構成

以下のセクションでは、Metadata Service (MDS) の構成について説明します。スーパーユーザーなど共有されるものもあり、同じ Java プロセスで実行されますが、一般的に前述の構成は Kafka 用です。同じプロパティファイルで構成しますが、このセクションの内容はすべて、Confluent MDS に関連した記述です。

MDS を使用すると、リソースとプリンシパルにロールバインディングを作成し、後で OAUTHBEARER リスナーへの認証に使用できるトークンを取得できます。

MDS 用の HTTPS リスナーの指定

次のように、MDS 用に SSL 暗号化を使用して HTTP プロトコルを指定します。

#############################  MDS Listener - which port to listen on #############################
confluent.metadata.server.listeners=https://0.0.0.0:8090,http://0.0.0.0:8091
confluent.metadata.server.advertised.listeners=https://<localhost>:8090,\
                                               http://<localhost>:8091

Copy

• confluent.metadata.server.listeners: この例では、MDS はポート 8090 および 8091 でリッスンします。
• confluent.metadata.server.advertised.listeners: アドバタイズされたポートは、インターネットを介して MDS とやり取りするクライアント用です。クライアントからやり取りの方法を確認されると、MDS はホストアドレス（アドバタイズされたもの）とポートを提供します。

MDS 用の SSL 設定

MDS リスナーの 1 つとして HTTPS を指定したため、MDS が実行されているマシンでキーと暗号化を提供する必要があります。SSL 暗号化を使用するには、適切なキーストアとトラストストアを指定する必要があります。

以下では、クライアントが HTTPS を介して MDS と通信する方法に関する SSL を構成しています。

##### SSL Settings for MDS #####
confluent.metadata.server.ssl.keystore.location=/var/ssl/private/client.truststore.jks
confluent.metadata.server.ssl.keystore.password=<keystore-password>
confluent.metadata.server.ssl.key.password=<key-password>
confluent.metadata.server.ssl.truststore.location=/var/ssl/private/kafka.keystore.jks
confluent.metadata.server.ssl.truststore.password=<truststore-password>

Copy

これは多くのレイヤーで構成可能であり、さまざまなコンポーネントに対して複数の証明書を生成することもできます。たとえば、プレフィックスを使用して、MDS と Kafka に別々の証明書、複数の内部リスナーと外部リスナーに別々の証明書を生成するといった処理が可能です。MDS 用に別の証明書を生成し、"mds" の付いた名前にすることができます（confluent.metadata.server.ssl.truststore.location=/var/ssl/private/mds.keystore.jks 。ここでは、"kafka" を "mds" に置き換えて区別しています）。この場合、証明書内のユーザーは "kafka" ではなく "mds" となり、スーパーユーザーのリストにも mds を追加することが一般的です。

MDS トークンの生成の有効化と構成

MDS は、ユーザー名およびパスワードと引き換えにトークンを提供できます。また、ユーザーまたはクライアントからトークンを受け取り、そのトークンによってプリンシパルを認証することもできます（初回以降はこれを使用して認証を継続できます）。

トークンは、Kafka で構成された OAUTHBEARER リスナーに対する認証に使用されます。この例では、MDS トークンサービスの有効化および構成を行う方法を示します。

############################# MDS Token Service Settings - enable token generation #############################
confluent.metadata.server.token.max.lifetime.ms=3600000
confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
confluent.metadata.server.token.signature.algorithm=RS256
confluent.metadata.server.authentication.method=BEARER

Copy

• トークンの有効期間を設定します（最大 1 時間、ミリ秒単位） : confluent.metadata.server.token.max.lifetime.ms=3600000
• トークンのパブリック/プライベートキーを指定します（どちらも MDS が実行されているマシンに保存されます）。
  • トークンを暗号化するためのキー（トークンを発行する場合）: confluent.metadata.server.token.key.path=<path-to-token-key-pair.pem>
• 暗号化アルゴリズムを定義します : confluent.metadata.server.token.signature.algorithm=RS256
• Confluent Platform 5.3 以降は、MDS 側でのトークン認証のサポートを指定する必要があります : confluent.metadata.server.authentication.method=BEARER

MDS トークンと OAUTHBEARER のウォークスルー

• MDS はプライベートキーを使用してトークンを暗号化します。Confluent Control Center などのクライアントがそのトークンを取得して OAUTHBEARER リスナーに渡すと、トークンはパブリックキーで復号化されます。
• Swagger インターフェイスで認証をテストするには、スクリプトを使用してキーを生成します（HTTPS のユーザー名/パスワードによるログインとトークン交換の両方）。
• ユーザーがユーザー名とパスワードの認証情報（認証）を使用して Control Center にログインすると、Control Center はユーザー名でトークンを取得した後、そのトークンを使用して Kafka と通信し、ユーザーが表示を認可されている情報のみを表示します。

MDS 側では次のような処理が行われます。

1. Control Center を呼び出します（ユーザー : fry、パスワード: future）
2. Control Center が MDS を呼び出し、認証を行います（ユーザー : fry、パスワード: future）
3. MDS は暗号化されたトークンをユーザーに返します（ユーザー : fry、パスワード: future）
4. Control Center は、ユーザー（ユーザー : fry、パスワード: future）のトークンおよびトピックのリストに対するリクエストを使用して Kafka を呼び出します

Kafka 側では次のような処理が行われます。

• 認証 : トークンを復号化し、プリンシパルを抽出します（Kafka では、トークンを復号化できる場合、そのトークンを信頼します。これは、復号化できたトークンの暗号化は MDS によって行われ、暗号化の実行中に MDS によって認証情報が検証済みであると信頼できるためです）
• 認可: 通常どおり、プリンシパルにアクセス許可があるかどうかを確認します

LDAP 設定の構成

LDAP は、ディレクトリサービスの API とコントラクトを定義するプロトコルです。

階層に基づくディレクトリにレコードを保存でき、各レコードには多数の属性、レコードの構造、グループの作成方法（識別名、一般名）があります。

LDAP 構成は、以下のような多くの要因によって大きく異なります。

• 個々のインストール
• 使用している LDAP のディストリビューションまたは実装（Microsoft Active Directory、Apache Directory Service、Open LDAP など）
• ユーザーごとに定義されている属性
• LDAP 構成（使用する階層、ユーザーとグループの保存方法など）

LDAP 構成の基本セットを以下に示します。

############################# Identity Provider Settings(LDAP - local OpenLDAP) #############################
ldap.java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory
ldap.com.sun.jndi.ldap.read.timeout=3000
ldap.java.naming.provider.url=ldap://<ldap-host>
# how mds authenticates to ldap server
ldap.java.naming.security.principal=CN=mds,CN=Demo,DC=confluent,DC=io
ldap.java.naming.security.credentials=<password>
ldap.java.naming.security.authentication=simple
# ldap search mode (GROUPS is default)
#ldap.search.mode=GROUPS
#ldap.search.mode=USERS
# how to search for users
ldap.user.search.base=CN=Demo,DC=confluent,DC=io
# how to search for groups
ldap.group.search.base=CN=Demo,DC=confluent,DC=io
# which attribute in ldap record corresponds to user name
ldap.user.name.attribute=sAMAccountName
ldap.user.memberof.attribute.pattern=CN=(.*),CN=Demo,DC=confluent,DC=io
ldap.group.object.class=group
ldap.group.name.attribute=sAMAccountName
ldap.group.member.attribute.pattern=CN=(.*),CN=Demo,DC=confluent,DC=io

Copy

ちなみに

上の例で、sAMAccountName は Microsoft Active Directory に固有の設定です。上の構成は、ご使用の環境の LDAP に固有の設定で変更してください。

これをテストするには、以下のデモのいずれかを実行します。

• RBAC のデモ
• Confluent Platform のデモ（cp-demo）

または、こちらの Open LDAP Docker イメージを使用してください。これは、RBAC Docker デモで使用されているものと同じ LDAP です。

LDAP の構成の詳細については、「Confluent LDAP Authorizer」および「LDAP Authorizer の構成」を参照してください。