GSSAPI の構成

SASL/GSSAPI の概要

SASL/GSSAPI は Kerberos を使用する組織向けのメカニズムです(たとえば、Active Directory を使用している場合)。Apache Kafka® 用に新しいサーバーをインストールする必要はありません。Kerberos の管理者に問い合わせて、クラスターにある各 Kafka ブローカー用のプリンシパル、および Kerberos 認証で Kafka に(クライアントやツール経由で)アクセスするすべてのオペレーティングシステムユーザーを確認してください。

Kerberos サーバーをまだ利用していなくても、Linux ベンダーが Kerberos 用のパッケージや、インストールおよび構成方法の簡単なガイドを用意している場合があるので、参照してください( UbuntuRed Hat )。Oracle Java を使用する場合、使用する Java バージョンの JCE ポリシーファイルをダウンロードし、$JAVA_HOME/jre/lib/security にコピーする必要があります。次のコマンドを使用して、ユーザー自身がこれらのプリンシパルを作成する必要があります。

sudo /usr/sbin/kadmin.local -q 'addprinc -randkey kafka/{hostname}@{REALM}'
sudo /usr/sbin/kadmin.local -q "ktadd -k /etc/security/keytabs/{keytabname}.keytab kafka/{hostname}@{REALM}"

Kerberos の要件は、すべてのホストが完全修飾ドメイン名(FQDN)で解決できることです。

このページの残りの部分では、Confluent Platform の各コンポーネント用に SASL/GSSAPI を構成する方法を示します。

GSSAPI のログ記録

SASL/GSSAPI のデバッグ出力を有効にするには、sun.security.krb5.debug システムプロパティを true に設定します。以下に例を示します。

export KAFKA_OPTS=-Dsun.security.krb5.debug=true
bin/kafka-server-start etc/kafka/server.properties

ちなみに

これらの手順では、ZIP または TAR アーカイブを使用して Confluent Platform をインストールしているという想定に基づいています。詳細については「オンプレミスのデプロイ」を参照してください。

ブローカー

Kafka クラスター内のすべてのブローカーを、クライアントからのセキュアな接続を受け付けるように構成します。ブローカーに構成の変更内容が反映されるためには、 ローリング再起動 が必要です。

次のセクションに示すとおり、Kafka ブローカーのセキュリティを有効にします。さらに、Confluent Control Center や Auto Data Balancer を使用している場合、次のコンポーネントに対してブローカーを構成します。

JAAS

注釈

個別の JAAS ファイルの使用がサポートされていますが、このアプローチは "推奨されていません"。以下の手順 1 と 2 の代わりに、「構成」で指定されているリスナーの構成を使用します。次の手順 3 はまだ必要であることに注意してください。

  1. まず、各 Kafka ブローカーの構成ディレクトリに、ブローカーの JAAS の構成ファイルを作成します。この例では、kafka_server_jaas.conf という名前です。

  2. 各ブローカーの JAAS ファイルで、それぞれのブローカーに対して一意のプリンシパルとキータブ(シークレットキー)を使用して、KafkaServer セクションを構成します。JAAS ファイルで構成されたキータブが、Kafka ブローカーを起動したオペレーティングシステムユーザーによって読み取ることができることを確認してください。

    // Specifies a unique keytab and principal name for each broker
    KafkaServer {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        keyTab="/etc/security/keytabs/kafka_server.keytab"
        principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
    };
    
  3. ブローカーで ZooKeeper に対して SASL 認証を行う場合は、「ZooKeeper JAAS」および「Zookeeper の構成」を参照してください。

構成

  1. 各ブローカーの server.properties ファイルで GSSAPI メカニズムを有効にします。

    # List of enabled mechanisms, can be more than one
    sasl.enabled.mechanisms=GSSAPI
    
    # Specify one of of the SASL mechanisms
    sasl.mechanism.inter.broker.protocol=GSSAPI
    
  1. ブローカー間通信で、SASL を有効にする場合は、次の行をブローカーのプロパティファイルに追加します(デフォルトは PLAINTEXT )。プロトコルを次のとおり設定します。

    • SASL_SSL: SSL 暗号化が有効な場合(SASL メカニズムが PLAIN である場合、SSL 暗号化の使用が必要)
    • SASL_PLAINTEXT: SSL 暗号化が有効ではない場合
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    security.inter.broker.protocol=SASL_SSL
    
  2. クライアントやブローカー間の SASL 接続をリッスンするポートを Kafka ブローカーに通知します。listeners を構成する必要があります。また、listeners と値が異なる場合は、必要に応じて advertised.listeners を構成する必要があります。リスナーを次のとおり設定します。

    • SASL_SSL: SSL 暗号化が有効な場合(SASL メカニズムが PLAIN である場合、SSL 暗号化の使用が必要)
    • SASL_PLAINTEXT: SSL 暗号化が有効ではない場合
    # With SSL encryption
    listeners=SASL_SSL://kafka1:9093
    advertised.listeners=SASL_SSL://localhost:9093
    
    # Without SSL encryption
    listeners=SASL_PLAINTEXT://kafka1:9093
    advertised.listeners=SASL_PLAINTEXT://localhost:9093
    
  3. 以下に該当する場合は、SASL_SSL ポートと PLAINTEXT ポートの両方を構成します。

    • SASL をブローカー間通信に対して有効にしない
    • クラスターに接続する一部のクライアントで SASL を使用しない

    SSL 暗号化を使う SASL リスナーと、PLAINTEXT リスナーが混在する例を次に示します。

    # With SSL encryption
    listeners=PLAINTEXT://kafka1:9092,SASL_SSL://kafka1:9093
    advertised.listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093
    
    # Without SSL encryption
    listeners=PLAINTEXT://kafka1:9092,SASL_PLAINTEXT://kafka1:9093
    advertised.listeners=PLAINTEXT://localhost:9092,SASL_PLAINTEXT://localhost:9093
    
  4. JAAS を構成するために、個別の JAAS の構成ファイルを使用しない場合、次のとおり、Kafka ブローカーのリスナーに対して JAAS を構成します。

    listener.name.sasl_ssl.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
      useKeyTab=true \
      storeKey=true \
      keyTab="/etc/security/keytabs/kafka_server.keytab" \
      principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
    
  5. GSSAPI を使用するとき、ブローカーの JAAS ファイルで構成されているブローカーのプライマリ名に一致するサービス名を構成します。前の JAAS ファイルの例では principal="kafka/kafka1.hostname.com@EXAMPLE.COM"; が指定されており、プライマリは "kafka" になります。

    sasl.kerberos.service.name=kafka
    

実行

個別の JAAS の構成ファイルを使用する場合は、各 Kafka ブローカーを起動するとき、次のとおり、JVM パラメーターとして JAAS ファイルの名前を渡します。

export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
bin/kafka-server-start etc/kafka/server.properties

ちなみに

これらの手順では、ZIP または TAR アーカイブを使用して Confluent Platform をインストールしているという想定に基づいています。詳細については「オンプレミスのデプロイ」を参照してください。

次に示すのは、コマンドラインから各ブローカーを起動するとき、JVM パラメーターとして渡すことができるいくつかのオプション設定です。

zookeeper.sasl.client

ZooKeeper に対して SASL 認証を有効にするために使用します。

  • 型: Boolean
  • デフォルト: true
  • 使用例: ブローカーを起動するとき、JVM パラメーターとしてパラメーターを渡すには、-Dzookeeper.sasl.client=true を指定します。
zookeeper.sasl.client.username

ZooKeeper に対する SASL 認証でユーザー名を変更する場合は、適切な名前を使用するようにシステムプロパティを設定します。

  • 型: string
  • デフォルト: zookeeper
  • 使用例: ブローカーを起動するとき、JVM パラメーターとしてパラメーターを渡すには、-Dzookeeper.sasl.client.username=zk を指定します。
zookeeper.sasl.clientconfig

JAAS のログインファイルでコンテキストキーを指定します。これは、ZooKeeper に対する SASL 認証でセクション名を変更するために使用します。

  • 型: string
  • デフォルト: Client
  • 使用例: ブローカーを起動するとき、JVM パラメーターとしてパラメーターを渡すには、-Dzookeeper.sasl.clientconfig=ZkClient を指定します。
-Djava.security.krb5.conf
必要に応じて krb5.conf ファイルへのパスを指定できます(詳細については、JDK の Kerberos の要件 を参照)。

クライアント

新しいプロデューサーとコンシューマークライアントは、Kafka バージョン 0.9.0 以上でセキュリティをサポートします。

Kafka Streams API を使用している場合、同等の SSL パラメーターおよび SASL パラメーターを構成する方法を参照してください。

  1. 次のプロパティをクライアントのプロパティファイル client.properties で構成します。

    sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    security.protocol=SASL_SSL
    
  2. ブローカーの JAAS ファイルで構成されている Kafka サーバーのプライマリ名に一致するサービス名を構成します。

    sasl.kerberos.service.name=kafka
    
  3. 各クライアントに対して一意のプリンシパル(通常、クライアントを実行しているユーザーと同じ名前)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useKeyTab=true \
        storeKey=true \
        keyTab="/etc/security/keytabs/kafka_client.keytab" \
        principal="kafkaclient1@EXAMPLE.COM";
    
  4. kafka-console-consumerkafka-console-producer などのコマンドラインユーティリティでは、kinituseTicketCache=true とともに使用できます。

    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
        useTicketCache=true;
    

ZooKeeper

このセクションでは、ブローカーが SASL/GSSAPI を使用して ZooKeeper に対する認証を実行できるように、ZooKeeper を構成する方法について説明します。

詳細については、次の ZooKeeper の SASL 認証を参照してください。

  1. クライアントとサーバー間の相互認証 : Kafka ブローカー(クライアント)と ZooKeeper (サーバー)間
  2. サーバー間の相互認証 : アンサンブル内の ZooKeeper ノード間

JAAS

  1. 各 ZooKeeper ノードに JAAS ファイルを作成します。この例では、zookeeper_jaas.conf です。

  2. 各 ZooKeeper ノードの JAAS ファイルで、各ノードに対して一意のプリンシパルとキータブ(シークレットキー)を使用して、Server セクションを構成します。JAAS ファイルで構成されたキータブが、ZooKeeper ノードを起動したオペレーティングシステムユーザーによって読み取ることができることを確認してください。

    // Specifies a unique keytab and principal name for each ZooKeeper node
    Server {
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        keyTab="/path/to/server/keytab"
        storeKey=true
        useTicketCache=false
        principal="zookeeper/yourzkhostname@EXAMPLE.COM";
    };
    
  3. ブローカーで ZooKeeper に対して SASL 認証を行う場合は、ブローカーの JAAS ファイルClient セクションも構成する必要があります。また、ZooKeeper ノードの ACL をブローカーが設定できるようにします。これにより、ブローカーのみが変更できるようにノードがロックされます。

    // ZooKeeper client authentication
    Client {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/etc/security/keytabs/kafka_server.keytab"
      principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
    };
    

構成

SASL による ZooKeeper 認証を有効にするには、以下を zookeeper.properties に追加します。

authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

注釈

ZooKeeper に格納されたメタデータについては、対応する znode を変更できるのはブローカーのみですが、znode は誰でも読み取り可能になっています。ZooKeeper に格納されたデータは、機密データではありませんが、znode の不適切な操作はクラスターのエラーにつながることがあります。

ZooKeeper 内のデータに ACL を設定するよう Kafka を構成するには(これにより書き込みを制限します)、etc/kafka/server.properties ファイルで ZooKeeper ACL を有効にします。

zookeeper.set.acl=true

注釈

デフォルトでは、ZooKeeper は認可に完全修飾名のプリンシパルを使用します。ZooKeeper の ACL をブローカー構成で zookeeper.set.acl パラメーターを使用して定義している場合、すべての Kafka ブローカーの間で同じプリンシパル(ホスト名を含まない)を使用します。同じプリンシパルを使用しない場合、ZooKeeper サーバーの構成ファイルにある kerberos.removeHostFromPrincipal および kerberos.removeRealmFromPrincipal パラメーターの両方を true に設定する必要があります。この構成により、すべてのブローカーが同じ方法で認可されること、プリンシパルの最初の部分がすべての Kafka ブローカーの間で同じであることが保証されます。

ネットワークのセグメント化を使用して、ZooKeeper へのアクセスを制限することを推奨します(新しいコンシューマーと新しいプロデューサーが使用される場合、ブローカーと一部の管理ツールだけが ZooKeeper にアクセスする必要があります)。

実行

ZooKeeper を起動するとき、次のとおり、JAAS ファイルの名前を JVM パラメーターとして渡します。

export KAFKA_OPTS="-Djava.security.auth.login.config=etc/kafka/zookeeper_jaas.conf"
bin/zookeeper-server-start etc/kafka/zookeeper.properties

Kafka Connect

このセクションでは、Kafka Connect のセキュリティを有効にする方法について説明します。Kafka Connect をセキュアにするには、次の対象でセキュリティを構成する必要があります。

  1. Kafka Connect ワーカー : Kafka Connect API の一部で、ワーカーは実際には内部的に機能する高度なクライアントです。
  2. Kafka Connect コネクター: コネクターにはプロデューサーとコンシューマーを組み込むことができます。そのため、ソースコネクターで使用される Connect プロデューサーやシンクコネクターで使用される Connect コンシューマーのデフォルトの構成をオーバーライドする必要があります。
  3. Kafka Connect REST: Kafka Connect では、 追加のプロパティ を使って SSL を使用するように構成できる REST API が公開されています。

次のセクションに示すとおり、Kafka Connect のセキュリティを構成します。さらに、Confluent Control Center のストリームモニタリングを Kafka Connect に対して使用している場合、次の対象でセキュリティを構成します。

次のすべてのプロパティを connect-distributed.properties で構成します。

  1. SASL/GSSAPI を使用するように Connect ワーカーを構成します。

    sasl.mechanism=GSSAPI
    sasl.kerberos.service.name=kafka
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    security.protocol=SASL_SSL
    
  2. 各ワーカーに対して一意のプリンシパル(通常、ワーカーを実行しているユーザーと同じ名前)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="connect@EXAMPLE.COM";
    
  3. セキュリティを活用するコネクターでは、ワーカーが使用するプロデューサーやコンシューマーのデフォルト構成をオーバーライドする必要もあります。ソースコネクターかシンクコネクターに応じて次のとおり実行します。

    • ソースコネクター: 同じプロパティに producer プレフィックスを追加して構成します。

      producer.sasl.mechanism=GSSAPI
      producer.sasl.kerberos.service.name=kafka
      # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
      producer.security.protocol=SASL_SSL
      producer.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
         useKeyTab=true \
         storeKey=true \
         keyTab="/etc/security/keytabs/kafka_client.keytab" \
         principal="connect@EXAMPLE.COM";
      
    • シンクコネクター: 同じプロパティに consumer プレフィックスを追加して構成します。

      consumer.sasl.mechanism=GSSAPI
      consumer.sasl.kerberos.service.name=kafka
      # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
      consumer.security.protocol=SASL_SSL
      consumer.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
         useKeyTab=true \
         storeKey=true \
         keyTab="/etc/security/keytabs/kafka_client.keytab" \
         principal="connect@EXAMPLE.COM";
      

Confluent Replicator

Confluent Replicator は、Kafka ソースコネクターの一種で、送信元 Kafka クラスターから送信先クラスターにデータを複製します。Replicator に組み込まれたコンシューマーは、送信元クラスターからデータを読み取り、Kafka Connect ワーカーに組み込まれたプロデューサーは送信先クラスターにデータを書き込みます。

Replicator バージョン 4.0 以下では、送信元および送信先の Kafka クラスターで ZooKeeper への接続が必要です。ZooKeeper で認証を有効にする場合、クライアントで ZooKeeper のセキュリティ認証情報を、Connect ワーカーのグローバル JAAS 構成設定 -Djava.security.auth.login.config に構成します。さらに送信元と送信先のクラスターで ZooKeeper のセキュリティ認証情報が一致する必要があります。

Confluent Replicator のセキュリティを構成するには、次に示すとおり Replicator コネクターを構成し、さらに次のコンポーネントを構成する必要があります。

Confluent Replicator を構成して、SASL/GSSAPI を使用するには、次のプロパティを Replicator の JSON 構成ファイルに追加します。

{
  "name":"replicator",
    "config":{
      ....
      "src.kafka.security.protocol" : "SASL_SSL",
      "src.kafka.sasl.mechanism" : "GSSAPI",
      "src.kafka.sasl.kerberos.service.name" : "kafka",
      "src.kafka.sasl.jaas.config" : "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/etc/security/keytabs/kafka_client.keytab\" principal=\"replicator@EXAMPLE.COM\";",
      ....
    }
  }
}

参考

Confluent Replicator の構成例については、SASL のソース認証デモのスクリプト を参照してください。共通のセキュリティ構成のデモについては、Replicator のセキュリティデモ を参照してください。

SASL/GSSAPI 認証を使用して送信先クラスター用に Confluent Replicator を構成するには、Replicator の JSON 構成を次の内容が含まれるように変更します。

{
  "name":"replicator",
    "config":{
      ....
      "dest.kafka.security.protocol" : "SASL_SSL",
      "dest.kafka.sasl.mechanism" : "GSSAPI",
      "dest.kafka.sasl.kerberos.service.name" : "kafka",
      "dest.kafka.sasl.jaas.config" : "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"/etc/security/keytabs/kafka_client.keytab\" principal=\"replicator@EXAMPLE.COM\";",
      ....
    }
  }
}

さらに、Connect ワーカーで次のプロパティを構成できます。

sasl.mechanism=GSSAPI
security.protocol=SASL_SSL
sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/etc/security/keytabs/kafka_client.keytab" principal="replicator@EXAMPLE.COM";
sasl.kerberos.service.name=kafka
producer.sasl.mechanism=GSSAPI
producer.security.protocol=SASL_SSL
producer.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/etc/security/keytabs/kafka_client.keytab" principal="replicator@EXAMPLE.COM";
producer.sasl.kerberos.service.name=kafka

ちなみに

コネクターとして動作する Replicator に対して、Connect ワーカーとは異なるセキュリティを実行する場合、producer.override の手法(「送信元クラスターでの Replicator の実行」を参照)を使用して、送信先でのセキュリティ認証情報をオーラバーライドすることによって、必要なセキュリティを実行できます。この場合、Connect ワーカーでは前に示したプロパティを設定しないでください。たとえば、SASL_SSL/GSSAPI セキュリティに基づいて Replicator を使用する場合、Connect ワーカーで RBAC の OAUTHBEARER 認証が実行されていると、そのワーカーの認証を継続して実行できます。producer.overrides は Replicator の構成を上書きしますが、ワーカーの構成では OAUTHBEARER 構成を引き続き保持することができます。

詳細については、「Kafka Connect のセキュリティの基礎」と Replicator のセキュリティの概要 で、Connect ワーカーの全般的なセキュリティ構成を確認してください。

参考

Confluent Replicator の構成例については、SASL の送信先認証デモのスクリプト を参照してください。共通のセキュリティ構成のデモについては、Replicator のセキュリティデモ を参照してください。

Confluent Control Center

Confluent Control Center は Kafka Streams をステートストアとして使用します。つまり Control Center が支援するクラスターにあるすべての Kafka ブローカーがセキュアである場合、Control Center アプリケーションもセキュアにする必要があります。

注釈

RBAC が有効である場合、Control Center を Kerberos とともに使用できません。Control Center では、OAUTHBEARER 以外の SASL メカニズムをサポートできないためです。

次のセクションに示すとおり、Control Center アプリケーションのセキュリティを有効にします。さらに、以下のコンポーネントのセキュリティを構成します。

  • Confluent Metrics Reporter <sasl_gssapi_metrics-reporter>: モニタリング対象の本稼働環境のクラスターで必須
  • Confluent Monitoring Interceptor: Control Center のストリームモニタリングを使用している場合に必要
  1. Control Center の SASL/GSSAPI とセキュリティプロトコルを etc/confluent-control-center/control-center.properties ファイルで有効にします。

    confluent.controlcenter.streams.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    confluent.controlcenter.streams.security.protocol=SASL_SSL
    
  2. GSSAPI を使用しているので、ブローカーの JAAS ファイルに構成されている Kafka サーバーのプライマリ名に一致するように、Kerberos サービス名を構成します。

    confluent.controlcenter.streams.sasl.kerberos.service.name=kafka
    
  3. 一意のプリンシパル(通常 Control Center を実行しているユーザーと同じ Control Center 名)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    confluent.controlcenter.streams.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
     useKeyTab=true \
     storeKey=true \
     keyTab="/etc/security/keytabs/kafka_client.keytab" \
     principal="controlcenter@EXAMPLE.COM";
    

Confluent Metrics Reporter

このセクションでは、Confluent Control Center と Auto Data Balancer で使用される Confluent Metrics Reporter のセキュリティを有効にするためのブローカーの構成方法について説明します。

Confluent Metrics Reporter の SASL/GSSAPI を構成するには、モニタリング対象となる本稼働環境のクラスター内にあるすべてのブローカーの server.properties ファイルで、次のとおりに構成を変更します。

  1. Confluent Metrics Reporter が有効であることを検証します。

    metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
    confluent.metrics.reporter.bootstrap.servers=kafka1:9093
    
  2. Confluent Metrics Reporter で SASL/GSSAPI メカニズムを有効にします。

    confluent.metrics.reporter.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    confluent.metrics.reporter.security.protocol=SASL_SSL
    
  3. GSSAPI を使用しているので、ブローカーの JAAS ファイルに構成されている Kafka サーバーのプライマリ名に一致するように、Kerberos サービス名を構成します。

    confluent.metrics.reporter.sasl.kerberos.service.name=kafka
    
  4. 一意のプリンシパルとキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    confluent.metrics.reporter.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="client@EXAMPLE.COM";
    

Confluent モニタリングインターセプター

Confluent Monitoring Interceptor を Confluent Control Center のストリームモニタリングで使用します。このセクションでは、Confluent Monitoring Interceptor 向けのセキュリティを次の 3 か所で有効にする方法について説明します。

  1. 一般クライアント
  2. Kafka Connect
  3. Confluent Replicator

重要

Confluent Monitoring Interceptor の典型的なユースケースは、構成が一般に異なる、個別のモニタリングクラスターにモニタリングデータを渡すことです。インターセプターの構成では、モニタリングされているコンポーネントの構成を継承しません。モニタリングされているコンポーネントの構成を使用する場合は、適切なプレフィックスを追加する必要があります。たとえば、オプション confluent.monitoring.interceptor.security.protocol=SSL では、プロデューサーに使用する場合、producer. のプレフィックスを使用して producer.confluent.monitoring.interceptor.security.protocol=SSL とする必要があります。

一般クライアント用のインターセプター

Kafka クライアントで動作する Confluent Control Center のストリームモニタリングについては、各クライアントで Confluent Monitoring Interceptor の SASL/GSSAPI を構成する必要があります。

  1. クライアントでインターセプターが構成されていることを検証します。
  • プロデューサーの場合

    interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    
  • コンシューマーの場合

    interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    
  1. インターセプターに対して SASL のメカニズムとセキュリティプロトコルを構成します。

    confluent.monitoring.interceptor.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    confluent.monitoring.interceptor.security.protocol=SASL_SSL
    
  2. Confluent Monitoring Interceptor 用の Kerberos サービス名 confluent.monitoring.interceptor.sasl.kerberos.service.name をブローカーの Kerberos サービス名 sasl.kerberos.service.name に一致するように構成します。

    confluent.monitoring.interceptor.sasl.kerberos.service.name=kafka
    
  3. 一意のプリンシパルとキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    confluent.monitoring.interceptor.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="client@EXAMPLE.COM";
    

Kafka Connect 用のインターセプター

  1. Kafka Connect で動作する Confluent Control Center のストリームモニタリングについては、Kafka Connect で Confluent Monitoring Interceptor の SASL/GSSAPI を構成する必要があります。コネクターがソースかシンクかに応じて、次のプロパティを connect-distributed.properties に追加することで、Connect ワーカーを構成します。
  • ソースコネクター: producer プレフィックスを使用して、Confluent Monitoring Interceptor の SASL メカニズムを構成します。

    producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    producer.confluent.monitoring.interceptor.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    producer.confluent.monitoring.interceptor.security.protocol=SASL_SSL
    producer.confluent.monitoring.interceptor.sasl.kerberos.service.name=kafka
    
  • シンクコネクター: consumer プレフィックスを使用して、Confluent Monitoring Interceptor の SASL メカニズムを構成します。

    consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    consumer.confluent.monitoring.interceptor.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    consumer.confluent.monitoring.interceptor.security.protocol=SASL_SSL
    consumer.confluent.monitoring.interceptor.sasl.kerberos.service.name=kafka
    
  1. Confluent Monitoring Interceptor 用の Kerberos サービス名 confluent.monitoring.interceptor.sasl.kerberos.service.name をブローカーの Kerberos サービス名 sasl.kerberos.service.name に一致するように構成します。ソースコネクターとシンクコネクターに対して、適切な構成パラメーターのプレフィックスを使用します。
  • ソースコネクター : producer プレフィックスを持つ Kerberos サービス名を使用するように、Confluent Monitoring Interceptor を構成します。

    producer.confluent.monitoring.interceptor.sasl.kerberos.service.name=kafka
    
  • シンクコネクター : consumer プレフィックスを持つ Kerberos サービス名を使用するように、Confluent Monitoring Interceptor を構成します。

    consumer.confluent.monitoring.interceptor.sasl.kerberos.service.name=kafka
    
  1. 一意のプリンシパル(通常、Connect ワーカーを実行しているユーザーと同じ名前)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。ソースコネクターとシンクコネクターに対して、適切な構成パラメーターのプレフィックスを使用します。
  • ソースコネクター: producer プレフィックスを使用して、Confluent Monitoring Interceptor の JAAS 構成を構成します。

    producer.confluent.monitoring.interceptor.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="connect@EXAMPLE.COM";
    
  • シンクコネクター: consumer プレフィックスを使用して、Confluent Monitoring Interceptor の JAAS 構成を指定します。

    consumer.confluent.monitoring.interceptor.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="connect@EXAMPLE.COM";
    

Replicator 用のインターセプター

Replicator で動作する Confluent Control Center のストリームモニタリングについては、Replicator の JSON 構成ファイルで Confluent Monitoring Interceptor の SASL を構成する必要があります。追加する構成プロパティのサブセットの一例を次に示します。

{
  "name":"replicator",
    "config":{
      ....
      "src.consumer.group.id": "replicator",
      "src.consumer.interceptor.classes": "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor",
      "src.consumer.confluent.monitoring.interceptor.sasl.mechanism": "GSSAPI",
      "src.consumer.confluent.monitoring.interceptor.security.protocol": "SASL_SSL",
      "src.consumer.confluent.monitoring.interceptor.sasl.kerberos.service.name": "kafka",
      "src.consumer.confluent.monitoring.interceptor.sasl.jaas.config" : "com.sun.security.auth.module.Krb5LoginModule required \nuseKeyTab=true \nstoreKey=true \nkeyTab=\"/etc/security/keytabs/kafka_client.keytab\" \nprincipal=\"replicator@EXAMPLE.COM\";",
      ....
    }
  }
}

Schema Registry

Schema Registry はスキーマを永続化するために Kafka を使用します。つまり、Kafka クラスターにデータを書き込むためのクライアントとして機能します。したがって、Kafka ブローカーでセキュリティが構成されている場合、Schema Registry もセキュリティを使用するように構成する必要があります。網羅的なリストについては、「Schema Registry の構成オプション」も参照してください。

  1. SASL 認証用に追加する schema-registry.properties 構成パラメーターのサブセットの一例を次に示します。

    kafkastore.bootstrap.servers=kafka1:9093
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    kafkastore.security.protocol=SASL_SSL
    kafkastore.sasl.mechanism=GSSAPI
    
  2. GSSAPI を使用しているので、ブローカーの JAAS ファイルで構成されている Kafka サーバーのプライマリ名に一致するサービス名を構成します。

    kafkastore.sasl.kerberos.service.name=kafka
    
  3. 一意のプリンシパル(通常 Schema Registry を実行しているユーザーと同じ名前)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    kafkastore.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="schemaregistry@EXAMPLE.COM";
    

REST Proxy

Confluent REST Proxy を SASL でセキュアにするには、REST Proxy と Kafka クラスター間でセキュリティを構成する必要があります。

すべての構成オプションの網羅的なリストについては、「SASL の認証」を参照してください。

  1. SASL/GSSAPI 認証用に追加する kafka-rest.properties 構成プロパティのサブセットの一例を次に示します。

    client.bootstrap.servers=kafka1:9093
    client.sasl.mechanism=GSSAPI
    # Configure SASL_SSL if SSL encryption is enabled, otherwise configure SASL_PLAINTEXT
    client.security.protocol=SASL_SSL
    
  2. ブローカーの JAAS ファイルで構成されている Kafka サーバーのプライマリ名に一致するサービス名を構成します。

    client.sasl.kerberos.service.name=kafka
    
  3. 一意のプリンシパル(通常 REST Proxy を実行しているユーザーと同じ名前)、およびキータブ(シークレットキー)を使用して、JAAS の構成のプロパティを構成します。

    client.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true \
       storeKey=true \
       keyTab="/etc/security/keytabs/kafka_client.keytab" \
       principal="restproxy@EXAMPLE.COM";
    

Confluent Rebalancer

SASL の Confluent Rebalancer を保護するには、rebalance-metrics-client.properties ファイル(または自分で選択したファイル名)で、Kafka クラスターの メトリクス構成オプション を指定します。

confluent.rebalancer.metrics.security.protocol=SASL_SSL
confluent.rebalancer.metrics.sasl.mechanism=GSSAPI
confluent.rebalancer.metrics.sasl.kerberos.service.name=kafka
confluent.rebalancer.metrics.ssl.truststore.location=<path>/kafka.client.truststore.jks
confluent.rebalancer.metrics.ssl.truststore.password=<truststore-password>
confluent.rebalancer.metrics.sasl.jaas.config= com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/Users/moshe/kerberos/kafka.user.keytab" principal="kafka@KAFKA.SECURE";
confluent.rebalancer.metrics.ssl.keystore.location=<path>/kafka.server.keystore.com.jks
confluent.rebalancer.metrics.ssl.keystore.password=<keystore-password>
confluent.rebalancer.metrics.ssl.key.password=<key-password>

コマンドラインの confluent-rebalancerrebalance-metrics-client.properties にこの構成を渡します。以下に例を示します。

confluent-rebalancer execute --bootstrap-server <localhost>:9092 --config-file rebalance-metrics-client.properties --command-config rebalance-admin-client.properties --throttle 100000 --verbose

--config-file オプションでは、メトリクスクラスターの接続を指定し、--command-config オプションでは、バランス調整されるクラスターへの管理クライアントの接続を指定することに注意してください。管理クライアント(rebalance-admin-client.properties)の接続を指定するとき、セキュアな接続を確保するには、rebalance-metrics-client.properties で使用したものと同じセキュリティ構成を使用します。ただし、このキーに対する confluent.rebalancer.metrics. プレフィックスを含める必要はありません。

さらに、バランス調整が進められているクラスターと異なるメトリクスクラスターを識別する必要がある場合、--metrics-bootstrap-server オプションを使用できます。デフォルトでは、--bootstrap-server オプションで指定されたクラスターからメトリクスが取得されます。

ここに示すように、JAAS の構成ファイルの場所を JVM パラメーター(confluent.rebalancer.metrics.sasl.jaas.config)としても指定できます。

export REBALANCER_OPTS="-Djava.security.auth.login.config=<path-to-jaas.conf>"

SASL/GSSAPI のトラブルシューティング

このセクションでは、SASL/GSSAPI の構成時に想定される一般的なエラーに対処するための基本的なトラブルシューティングのヒントを示します。

Kerberos

Kerberos ホストのホスト名とプリンシパルが正確に一致しない場合は、以下のホスト名エラーがサービスログに出力されることがあります。

org.apache.kafka.common.errors.SaslAuthenticationException: An error: (java.security.PrivilegedActionException: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))]) occurred when evaluating SASL token received from the Kafka Broker. This may be caused by Java's being unable to resolve the Kafka Broker's hostname correctly. You may want to try to adding '-Dsun.net.spi.nameservice.provider.1=dns,sun' to your client's JVMFLAGS environment. Users must configure FQDN of kafka brokers when authenticating using SASL and `socketChannel.socket().getInetAddress().getHostName()` must match the hostname in `principal/hostname@realm` Kafka Client will go to AUTHENTICATION_FAILED state.
Caused by: javax.security.sasl.SaslException: GSS initiate failed [Caused by GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))]
      at com.sun.security.sasl.gsskerb.GssKrb5Client.evaluateChallenge(GssKrb5Client.java:211)
  ...
Caused by: GSSException: No valid credentials provided (Mechanism level: Server not found in Kerberos database (7))
      ... 26 more

このようなホスト名エラーを回避するには:

  • プロパティファイル内で sasl.kerberos.service.name に指定された値が、対応するキータブファイル内のプライマリ名と一致することを確認します。たとえば、principal="kafka_broker/kafka1.hostname.com@EXAMPLE.COM" であれば、プライマリは kafka_broker であり、sasl.kerberos.service.name=kafka_broker と構成する必要があります。

  • kinitklistkdestroy の各コマンドで、キータブを使用して Kerberos の認証情報を初期化、リスト表示、削除する動作が想定どおりに行われることを確認します。その例を次に示します。

    # To display the current contents of the cache
    klist
    # To acquire new credentials
    kinit -k -t ./filename.keytab kafka_broker/kafka1.hostname.com@EXAMPLE.COM
    # To verify that the credentials were initialized and updated
    klist
    # To remove any newly-initialized credentials (if necessary)
    kdestroy
    
  • クライアントとサーバーの間で Kerberos 環境が正しくセットアップされていることを確認します。Java SASL API を使用してクライアントとサーバー ホストの間の通信を検証する単純な Kerberos ベースの SaslTestClient および SaslServer アプリケーションについては、「Exercise 4: Using the Oracle Java SASL API」を参照してください。

LDAP

Kerberos を使用して LDAP に対して認証できるように、Kafka を構成する際に LDAP 構成のテストおよびトラブルシューティングを行うには、「LDAP クライアント認証のテストとトラブルシューティング」を参照してください。