1. Home
  2. Platform
  3. マルチ DC デプロイのアーキテクチャ
  4. Cluster Linking

Cluster Linking のセキュリティ

Looking for Confluent Cloud Cluster Linking docs? You are currently viewing Confluent Platform documentation. If you are looking for Confluent Cloud docs, check out Cluster Linking on Confluent Cloud.

送信元クラスターへの接続に使用されるすべてのセキュリティ構成は、クラスターリンクの作成時にそのクラスターリンクで構成することができます。各リンクは "厳密に 1 つ" のリンク認証情報に関連付けられます。この認証情報は、そのリンクを使用した送信元クラスターへの接続の認証に使用されます。同じクラスターで、さまざまなクラスターリンクが異なるセキュリティ認証情報を使用する場合があります。リンク認証情報には、送信元クラスターでの適切なアクセス許可が付与される必要があります。

認証

以下の例に、送信元クラスターと通信するために、クラスターリンク用の SASL メカニズムとして GSSAPI を使用する SASL_SSL の構成方法を示します。これらの構成は、クラスターリンクに対するプロパティの設定方法 のセクションに記載されているように、config-file を使用して設定できます。

security.protocol=SASL_SSL
ssl.truststore.location=/path/to/truststore.p12
ssl.truststore.password=truststore-password
ssl.truststore.type=PKCS12
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
    useKeyTab=true
    storeKey=true \
    keyTab="/path/to/link.keytab" \
    principal="clusterlink1@EXAMPLE.COM";

このシナリオでは、Cluster Linking 構成には、送信元クラスターに接続するためのクライアント側の SSL および SASL/GSSAPI 構成オプションが含まれている必要があります。

SSL キーとトラストストアの作成の詳細については、「TLS での暗号化と認証」を参照してください。SASL/GSSAPI の詳細については、「GSSAPI の構成」を参照してください。セキュリティが有効の場合、重要なリンク構成を暗号化するために password.encoder.secret を使用してブローカーを構成する必要があります。「パスワード構成の動的なアップデート」も参照してください。

他の SASL メカニズムを使用するようにクラスターリンクを構成するには、そのメカニズム用のクライアント側セキュリティ構成を組み込みます。サポートされている他のメカニズムについては、「JAAS を使用した SASL による認証」を参照してください。セキュリティプロトコルとして SSL を使用する双方向 SSL 認証を使用するには、リンク用のキーストアも構成する必要があります。詳細については、「TLS での暗号化と認証」を参照してください。

注釈

クラスターリンクは、リンクで構成された送信元認証情報を使用して送信元クラスターと通信します。リンクを機能させるには、これらの認証情報が有効である必要があります。

Mutual TLS (mTLS)

Cluster Linking can use mTLS (two-way verification) for some, but not all, data exchanges:

  • Confluent Cloud to Confluent Cloud does not use mTLS; it uses TLS and SASL as described in Cluster Linking on Confluent Cloud.
  • Data coming into Confluent Cloud from OSS Kafka can be configured to use mTLS
  • Data coming into Confluent Cloud from Confluent Platform can be configured to use either mTLS or source-initiated cluster links with TLS+SASL. mTLS cannot be used with source-initiated cluster links.

Follow these guidelines when configuring mTLS for Confluent Platform to Confluent Platform cluster links:

  • To learn more about PEM files and possible configurations, see the Apache Kafka KIP that introduced PEM file support.

  • If you reference a keystore/truststore directly (for example, keystore.jks), the same files must be available in the same location on each of the brokers

  • If the files cannot be copied to the brokers directly, be sure to list them in PEM format

    • If using kafka-cluster-links , add \n to indicate a new lines in the encrypted certificates. The example below is modified; it doesn't include full keys, but should be enough to show the use of \n:

      security.protocol=SSL
ssl.keystore.type=PEM
ssl.truststore.type=PEM
ssl.key.password=clientpass
ssl.endpoint.identification.algorithm=
ssl.keystore.key=-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIIE6TAbBgkqhkiG9w0BBQMwDgQI4d+aGvbJsIcCAggABIIEyB+qVnCcV9BbBz97\njmG1Xm6p2HIbO4tgMTld/Kmfx/4RHECdnhjiP3plH4yxZnaHBbiRHlno1d2xL93n\n4pqQkfhfJhBIqzoO3A==\n-----END ENCRYPTED PRIVATE KEY-----
ssl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----\nMIIETjCCAzagAwIBAgIUX64Yeyvwae61we2rC4gniQUcrdIwDQYJKoZIhvcNAQEL\nBQAwZTEYMBYGA1UEAwwPVGVzdENvbmZsdWVudENBMQwwCgYDVQQLDANFQUExEDAO\nrPhy0+XGDZbC4PWYi3FogFTKKjKzjtO3ZP5nXt6zvF9/nCn8RpljKJH4brIIlhPM\n3Us=\n-----END CERTIFICATE-----
ssl.truststore.certificates=-----BEGIN CERTIFICATE-----\nMIIDRjCCAi4CCQCW7jXMNbE1XzANBgkqhkiG9w0BAQsFADBlMRgwFgYDVQQDDA9U\nZXN0Q29uZmx1ZW50Q0ExDDAKBgNVBAsMA0VBQTEQMA4GA1UECgwHRnJlZW1hbjEP\n4Q/DVCHHUvJwHxd/5Bc08s56FYHFetoB1d4=\n-----END CERTIFICATE----

    • If using confluent kafka link, the command doesn’t interpret \n in the same way. Instead, replace literal \n with a single space.

      ssl.endpoint.identification.algorithm=
security.protocol=SSL
ssl.keystore.type=PEM
ssl.keystore.certificate.chain=-----BEGIN CERTIFICATE-----\
MIIDeTCCAmGgAwIBAgIJAKU/BVTP/yyUMA0GCSqGSIb3DQEBCwUAMHMxCzAJBgNV\
BAYTAlVTMQswCQYDVQQIDAJNVDEQMA4GA1UEBwwHQm96ZW1hbjEPMA0GA1UECgwG\
RmlndXJlMQ8wDQYDVQQLDAZGaWd1cmUxIzAhBgNVBAMMGmthZmthLXJvb3QudGVz\
...
k1PdbjufUewo7KQf8nde1IefAbSARG6Fu4oY2g4=\
-----END CERTIFICATE-----
ssl.keystore.key=-----BEGIN ENCRYPTED PRIVATE KEY-----\
MIIE5DAcBgoqhkiG9w0BDAEBMA4ECIYZ0EPRimP2AgIIAASCBMJLvS+Mtm9HzU4O\
fu3EuTu7LyjH4KEyzSYsuKnhYMDbFxHD1V/dSIr3N8ZqDjz5Xr3TvsN3pVwa5BFh\
Tv8NhhzEWB6jQtf7xo0cGPlL2VVO95D2aAvLBTQegWxqOXrhGIzqffyw/59uyJi9\
...
WsrgdacPOGc4pC+bFXUOySXUBrvI47rPLn3tHHWnGfKEOKA0zMRPHjMUWmooKhKl\
VKz1+zPeZ1s=\
-----END ENCRYPTED PRIVATE KEY-----
ssl.key.password=<REDACTED>
ssl.truststore.type=PEM
ssl.truststore.certificates=-----BEGIN CERTIFICATE-----\
MIIDYjCCAkoCCQCw8cwWkAxRaDANBgkqhkiG9w0BAQsFADBzMQswCQYDVQQGEwJV\
...
CIi0IXotll17POA+hgHeWBxhKg2ULmPxR7UgHgCcnajTNMmbbaBfOSa6xe1qOJvU\
CzFNJ0Yz\
-----END CERTIFICATE-----

      Alternatively, you can use an escaped newline; that is, backslash (), and then an actual newline.

認可（ACL）

ACL が有効であるデプロイでは、送信元クラスターと送信先クラスターの両方に別の ACL を追加する必要があります。ACL の作成の詳細については、「ACL を使用した認可」を参照してください。関連する操作、リソース、および API の完全なリストについては、サブトピックの「操作」を参照してください。

注意

ACL migration, previously available in Confluent Platform 6.0.0 through 6.2.x, was removed due to a security vulnerability. If you are using ACL migration (ACL sync) in your deployments, please disable it by setting acl.sync.enable=false on your cluster links. This feature will be re-introduced in an upcoming Confluent Platform release with the security issue resolved.

送信先クラスターのブローカーに対する ACL

オフセット移行が有効にされていない場合、ブローカーには追加のアクセス許可は必要ありません。

オフセット移行が有効にされている場合、送信先クラスター内のブローカーには追加の ACL が必要です。

操作 リソース API
READ トピック コンシューマーオフセット移行に使用する API
READ グループ コンシューマーオフセット移行に使用する API
ALTER トピック（ミラー） AlterTopicMirrors

送信元クラスターから送信先クラスターへの ACL の移行

Cluster Linking を使用すると、クラスター間を接続し、ブローカーを使用して Kafka クラスター間でデータをレプリケートすることで、トピック共有、クラスター移行、およびハイブリッドアーキテクチャを実現できます。すべてのシナリオ用に、Cluster Linking には送信元から送信先に ACL を移行するオプションが用意されています。クラスター間での ACL の移行はリンクを使用することで実行できます。その際は、どの ACL を移行するかを細かく制御できます。

クラスターリンクを作成するときは、送信先（ミラー）に移行する ACL をリソース、プリンシパル、およびホストに基づいて指定できます。送信先クラスターで必要となる特定の ACL をいくつでも移行できますし、ACL をすべて移行することもできます。

ちなみに

また、 コンシューマーグループをソースから送信先クラスターに移行する こともできます。

注意

Confluent Platform 7.1.0 から提供されている kafka-acls の --link-id オプションは実験的な段階であるため、本稼働環境のデプロイでは使用しないでください。特に、ACL の作成には link-ID を使用しないでください。送信元クラスターで --link-id を使用して ACL を作成した場合は、そのリンク ID によって管理用としてマークされるため、acl.sync.filters に関係なく、送信先で同期されません。現在、Confluent Platform では、kafka-acls で作成されたリンク ID のクラスターの存在を検証する機能は提供されていません。

前提条件

  • Confluent Server Authorizer の構成 や Apache Kafka® の AclAuthorizer などのオーソライザーが構成されている必要があります。
  • 「ACL を使用した認可」と「認可（ACL）」の説明に従って、クラスターとクラスターリンクに ACL 認可が構成されている必要があります。
  • 送信元クラスターに DESCRIBE クラスター ACL（DescribeAcls API）、送信先クラスターに ALTER クラスター ACL（CreateAcls/DeleteAcls APIs）が適切な認可として設定されている必要があります。

定期的な ACL 移行の構成

クラスターリンクに対して ACL 移行を指定できます。

推奨される方法は、ACL を acl.filters.json ファイルに定義し、そのファイル名を引数として CLI コマンド の --acl-filters-json-file フラグに渡すことです。

ACL は、 クラスターリンクを作成する ときか、または 既存の構成のアップデート として構成できます。

acl.sync.ms は、ACL の更新頻度を指定するプロパティです。デフォルトは 5 秒（5000 ミリ秒）です。

また、acl.filters.json は必須パラメーターです。これは、移行する ACL のリストを指定する JSON 文字列です。デフォルトは ”” です。この文字列に値を入力するには、ACL が指定されている JSON ファイルを渡します。

以下に、クラスターリンクを作成するときに ACL 移行とともにクラスターリンクをセットアップする例を示します。この例では、リンク構成（ACL 移行プロパティを含む）は直接コマンドラインで指定せず、link-configs.txt ファイルに定義されます。

.bin/kafka-cluster-links --bootstrap-server destinationCluster:9092 \
 --command-config dest-credentials.txt --create --link example-link \
 --config-file link-configs.txt --acl-filters-json-file acls.filters.json --consumer-group-filters-json-file consumer-groups.json

この構成は引数として link-configs.txt の --config-file に渡すことができます。以下に、link-configs.txt の内容の例を示します。

bootstrap.servers=sourceCluster:9092

ssl.endpoint.identification.algorithm=https

security.protocol=SASL_SSL

sasl.mechanism=PLAIN

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="sourceClusterUserName" password="sourceClusterPassword";

acl.sync.enable=true

acl.sync.ms=1000

consumer.offset.sync.enable=true

consumer.offset.sync.ms=1000

以下のセクションでは、実際の ACL を acls.filters.json に定義する方法の例を示します。

注意

構成ファイルを呼び出すコマンドを実行した後、セキュリティ認証情報を含むファイルを削除して、認証情報がファイルシステムに残らないようにします。上記の kafka-cluster-links コマンドの例では、dest-credentials.txt ファイルと link-configs.txt ファイルを削除します。

例

以下の例では、さまざまなタイプの ACL 移行（トピックやリソースを対象とした移行、および混在する ACL の移行）用の JSON を構成する方法を示します。

ちなみに

  • ホスト名ではなく IP アドレスを使用する必要があります。
  • ホストを指定しない場合、ACL ではデフォルトとしてワイルドカード * を使用してすべてのホストからのアクセスを提供します。

すべての ACL を送信元から送信先クラスターに移行

すべての ACL を送信元から送信先クラスターに移行するには、acl.filters.json に以下を指定します。

{
   "aclFilters": [
     {
       "resourceFilter": {
         "resourceType": "any",
         "patternType": "any"
       },
       "accessFilter": {
         "operation": "any",
         "permissionType": "any"
       }
     }
   ]
 }

JSON の各フィールドには、以下のオプションがあります。

  • resourceType: any、または「ACL を使用した認可」の「リソース」で指定されている値の 1 つを指定できます。
  • patternType: any、literal、prefixed、または match のいずれか
  • name: リソースの名前です。空白のままにした場合、デフォルトとしてワイルドカード * が使用されます。これは、指定された resourceType のすべての名前と一致することを意味します。
  • principal: プリンシパルの名前です。空白のままにした場合、デフォルトとしてワイルドカード * が使用されます。これは、指定された operation と permissionType を持つすべてのプリンシパルと一致することを意味します
  • operation: any、または「操作」の "操作" 列に指定されている値のいずれか 1 つ
  • permissionType: any、allow、または deny のどちらか
  • host: 操作の発生元になることができるホストです。空白のままにした場合、デフォルトとしてワイルドカード * が使用されます。これは、指定された operation と permissionType を持つすべてのホストと一致することを意味します

トピックにすべての ACL 固有を移行

1 つのトピック（この場合、トピック pineapple）に固有のすべての ACL を移行するには、acl.filters.json に以下を行います。

{
  "aclFilters": [
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "literal",
        "name": "pineapple"
      },
      "accessFilter": {
        "operation": "any",
        "permissionType": "any"
      }
    }
  ]
}

プリンシパルに固有のすべての ACL を移行

Alice というプリンシパルに固有のすべての ACL を移行するには、acl.filters.json に以下を指定します。

{
  "aclFilters": [
    {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "operation": "any",
        "permissionType": "any"
      }
    }
  ]
}

ACL の組み合わせを移行

特定の ACL JSON を使用して、より複雑な ACL 移行を行うこともできます。たとえば、以下の ACL のみを送信元クラスターから移行する必要があるとします。

  1. Alice は、ホスト goodHostIPAddress からすべてのトピックに対して書き込むことができます。
  2. Alice は、ホスト goodHostIPAddress のすべてのトピックから読み取ることができます。
  3. Alice は、ホスト goodHostIPAddress ですべてのトピックの作成することができます。
  4. Bob は、ホスト anotherGoodHostIPAddress のプレフィックス coffee が付いたすべてのトピックからの読み取ることができます。
  5. Bob は、anotherGoodHostIPAddress でプレフィックス breakfast.bar が付いたすべてのトピックを作成することができます。
  6. Bob は、anotherGoodHostIPAddress で breakfast.bar という名前ですべてのトピックを作成することができます。
  7. Mallory は、すべてのホストのすべてのトピックからすべての操作を拒否されます。
  8. Admin（管理者）は、ホスト adminHostIPAddress からすべてのトピックですべての操作を行うことができます。
  9. Trent は、ホスト greatHostIPAddress のすべてのトピックから読み取ることができます。
  10. Eve は、すべてのホストのすべてのトピックからの読み取ることが拒否されます。

上記で指定されている ACL を移行するには、acl.filters.json の JSON ファイルに以下の ACL を指定します。

{
  "aclFilters": [
  // filter for 1
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 2
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "write",
        "permissionType": "allow"
      }
    },
    // filter for 3
    {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Alice",
        "host": "goodHostIPAddress",
        "operation": "create",
        "permissionType": "allow"
      }
    },
    // filter for 4
   {
      "resourceFilter": {
        "resourceType": "topic",
        "name": "coffee",
        "patternType": "prefixed"
      },
      "accessFilter": {
        "principal": "User:Bob",
        "host": "anotherGoodHostIPAddress",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 5 and 6
   {
      "resourceFilter": {
        "resourceType": "topic",
        "name": "breakfast.bar",
        // match serves as a catch-all for all the names of
        // a topic this principal is authorized to access
        "patternType": "match"
      },
      "accessFilter": {
        "principal": "User:Bob",
        "host": "anotherGoodHostIPAddress",
        "operation": "create",
        "permissionType": "allow"
      }
    },
    // filter for 7
   {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Mallory",
        "operation": "any",
        "permissionType": "deny"
      }
    },
    // filter for 8
   {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Admin",
        "host":"adminHost",
        "operation": "any",
        "permissionType": "allow"
      }
    },
    // filter for 9
   {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Trent",
        "host":"greatHost",
        "operation": "read",
        "permissionType": "allow"
      }
    },
    // filter for 10
   {
      "resourceFilter": {
        "resourceType": "topic",
        "patternType": "any"
      },
      "accessFilter": {
        "principal": "User:Eve",
        "operation": "read",
        "permissionType": "deny"
      }
    }
  ]
}