RHEL での AMQ Streams の使用

SASL DIGEST-MD5 認証の有効化

1. すべての ZooKeeper ノードで、/opt/kafka/config/zookeeper-jaas.conf JAAS 設定ファイルを作成または編集し、以下のコンテキストを追加します。

   QuorumServer {
          org.apache.zookeeper.server.auth.DigestLoginModule required
          user_<Username>="<Password>";
   };
   
   QuorumLearner {
          org.apache.zookeeper.server.auth.DigestLoginModule required
          username="<Username>"
          password="<Password>";
   };

   Copy to Clipboard Toggle word wrap

   ユーザー名とパスワードは両方の JAAS コンテキストで同一である必要があります。以下に例を示します。

   QuorumServer {
          org.apache.zookeeper.server.auth.DigestLoginModule required
          user_zookeeper="123456";
   };
   
   QuorumLearner {
          org.apache.zookeeper.server.auth.DigestLoginModule required
          username="zookeeper"
          password="123456";
   };

   Copy to Clipboard Toggle word wrap

2. すべての ZooKeeper ノードで、/opt/kafka/config/zookeeper.properties ZooKeeper 設定ファイルを編集し、以下のオプションを設定します。

   quorum.auth.enableSasl=true
   quorum.auth.learnerRequireSasl=true
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner
   quorum.auth.server.loginContext=QuorumServer
   quorum.cnxn.threads.size=20

   Copy to Clipboard Toggle word wrap

3. すべての ZooKeeper ノードを 1 つずつ再起動します。JAAS 設定を ZooKeeper に渡すには、KAFKA_OPTS 環境変数を使用します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/zookeeper-jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   Copy to Clipboard Toggle word wrap

SASL DIGEST-MD5 認証の有効化

1. すべての ZooKeeper ノードで、/opt/kafka/config/zookeeper-jaas.conf JAAS 設定ファイルを作成または編集し、以下のコンテキストを追加します。

   Server {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       user_super="<SuperUserPassword>"
       user<Username1>_="<Password1>" user<USername2>_="<Password2>";
   };

   Copy to Clipboard Toggle word wrap

   super は自動的に管理者特権を持たせます。このファイルには複数のユーザーを含めることができますが、Kafka ブローカーが必要とする追加ユーザーは 1 つだけです。Kafka ユーザーに推奨される名前は kafka です。

   以下の例は、クライアント/サーバー間の認証の Server コンテキストを示しています。

   Server {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       user_super="123456"
       user_kafka="123456";
   };

   Copy to Clipboard Toggle word wrap

2. すべての ZooKeeper ノードで、/opt/kafka/config/zookeeper.properties ZooKeeper 設定ファイルを編集し、以下のオプションを設定します。

   requireClientAuthScheme=sasl
   authProvider.<IdOfBroker1>=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   authProvider.<IdOfBroker2>=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   authProvider.<IdOfBroker3>=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

   Copy to Clipboard Toggle word wrap

   authProvider.<ID> プロパティーは、ZooKeeper クラスターの一部であるすべてのノードに追加する必要があります。3 ノードの ZooKeeper クラスターの設定例は以下のようになります。

   requireClientAuthScheme=sasl
   authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   authProvider.2=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   authProvider.3=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

   Copy to Clipboard Toggle word wrap

3. すべての ZooKeeper ノードを 1 つずつ再起動します。JAAS 設定を ZooKeeper に渡すには、KAFKA_OPTS 環境変数を使用します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/zookeeper-jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   Copy to Clipboard Toggle word wrap

SASL DIGEST-MD5 認証の有効化

1. すべての Kafka ブローカーノードで、/opt/kafka/config/jaas.conf JAAS 設定ファイルを作成または編集し、以下のコンテキストを追加します。

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="<Username>"
       password="<Password>";
   };

   Copy to Clipboard Toggle word wrap

   ユーザー名とパスワードは ZooKeeper で設定されているものと同じである必要があります。

   Client コンテキストの例を以下に示します。

   Client {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       username="kafka"
       password="123456";
   };

   Copy to Clipboard Toggle word wrap

2. すべての Kafka ブローカーノードを 1 つずつ再起動します。JAAS 設定を Kafka ブローカーに渡すには、KAFKA_OPTS 環境変数を使用します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

手順

1. AclAuthorizer を使用するように、Kafka 設定ファイル /opt/kafka/config/server.properties を編集します。

   authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

   Copy to Clipboard Toggle word wrap
2. Kafka ブローカーを (再) 起動します。

手順

1. --add オプションを指定して kafka-acls.sh を実行します。

   例:

   • MyConsumerGroup コンシューマーグループを使用して、user1 および user2 の myTopic からの読み取りを許可します。

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Read --topic myTopic --allow-principal User:user1 --allow-principal User:user2
     
     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Describe --topic myTopic --allow-principal User:user1 --allow-principal User:user2
     
     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Read --operation Describe --group MyConsumerGroup --allow-principal User:user1 --allow-principal User:user2

     Copy to Clipboard Toggle word wrap

   • user1 が IP アドレスホスト 127.0.0.1 から myTopic を読むためのアクセスを拒否します。

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --operation Describe --operation Read --topic myTopic --group MyConsumerGroup --deny-principal User:user1 --deny-host 127.0.0.1

     Copy to Clipboard Toggle word wrap

   • MyConsumerGroup で myTopic のコンシューマーとして user1 を追加します。

     bin/kafka-acls.sh --authorizer-properties zookeeper.connect=zoo1.my-domain.com:2181 --add --consumer --topic myTopic --group MyConsumerGroup --allow-principal User:user1

     Copy to Clipboard Toggle word wrap

手順

1. すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集し、zookeeper.set.acl フィールドを true に設定します。

   zookeeper.set.acl=true

   Copy to Clipboard Toggle word wrap
2. Kafka ブローカーを起動します。

ZooKeeper ACL の有効化

1. すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集し、zookeeper.set.acl フィールドを true に設定します。

   zookeeper.set.acl=true

   Copy to Clipboard Toggle word wrap
2. すべての Kafka ブローカーを 1 つずつ再起動します。

3. zookeeper-security-migration.sh ツールを使用して、既存のすべての znodes ノードに ACL を設定します。

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=<ZooKeeperURL>
   exit

   Copy to Clipboard Toggle word wrap

   以下に例を示します。

   su - kafka
   cd /opt/kafka
   KAFKA_OPTS="-Djava.security.auth.login.config=./config/jaas.conf"; ./bin/zookeeper-security-migration.sh --zookeeper.acl=secure --zookeeper.connect=zoo1.my-domain.com:2181
   exit

   Copy to Clipboard Toggle word wrap

手順

1. クラスター内のすべての Kafka ブローカーの TLS 証明書を生成します。証明書には、Common Name または Subject Alternative Name にアドバタイズされたアドレスおよびブートストラップアドレスが必要です。

2. 以下のように、すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集します。

   • listener.security.protocol.map フィールドを変更して、TLS 暗号化を使用するリスナーに SSL プロトコルを指定します。
   • ssl.keystore.location オプションを、ブローカー証明書を持つ JKS キーストアへのパスに設定します。

   • ssl.keystore.password オプションを、キーストアの保護に使用したパスワードに設定します。

     以下に例を示します。

     listeners=UNENCRYPTED://:9092,ENCRYPTED://:9093,REPLICATION://:9094
     listener.security.protocol.map=UNENCRYPTED:PLAINTEXT,ENCRYPTED:SSL,REPLICATION:PLAINTEXT
     ssl.keystore.location=/path/to/keystore/server-1.jks
     ssl.keystore.password=123456

     Copy to Clipboard Toggle word wrap
3. Kafka ブローカーを (再) 起動します。

手順

1. ユーザー証明書に署名するために使用される認証局の公開鍵が含まれる JKS トラストストアを準備します。

2. 以下のように、すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集します。

   • ssl.truststore.location オプションを、ユーザー証明書の認証局が含まれる JKS トラストストアへのパスに設定します。
   • ssl.truststore.password オプションを、トラストストアの保護に使用したパスワードに設定します。

   • ssl.client.auth オプションを required に設定します。

     以下に例を示します。

     ssl.truststore.location=/path/to/truststore.jks
     ssl.truststore.password=123456
     ssl.client.auth=required

     Copy to Clipboard Toggle word wrap
3. Kafka ブローカーを (再) 起動します。

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集するか、作成します。このファイルには、すべてのユーザーとそのパスワードが含まれている必要があります。このファイルがすべての Kafka ブローカーで同じであることを確認します。

   以下に例を示します。

   KafkaServer {
       org.apache.kafka.common.security.plain.PlainLoginModule required
       user_admin="123456"
       user_user1="123456"
       user_user2="123456";
   };

   Copy to Clipboard Toggle word wrap

2. 以下のように、すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集します。

   • listener.security.protocol.map フィールドを変更して、SASL PLAIN 認証を使用するリスナーの SASL_PLAINTEXT または SASL_SSL プロトコルを指定します。

   • sasl.enabled.mechanisms オプションを PLAIN に設定します。

     以下に例を示します。

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=PLAIN

     Copy to Clipboard Toggle word wrap

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 構成を Kafka ブローカーに渡します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

手順

1. /opt/kafka/config/jaas.conf JAAS 設定ファイルを編集するか、作成します。KafkaServer コンテキストの ScramLoginModule を有効にします。このファイルがすべての Kafka ブローカーで同じであることを確認します。

   以下に例を示します。

   KafkaServer {
       org.apache.kafka.common.security.scram.ScramLoginModule required;
   };

   Copy to Clipboard Toggle word wrap

2. 以下のように、すべてのクラスターノードの /opt/kafka/config/server.properties Kafka 設定ファイルを編集します。

   • listener.security.protocol.map フィールドを変更して、SASL SCRAM 認証を使用するリスナーの SASL_PLAINTEXT または SASL_SSL プロトコルを指定します。

   • sasl.enabled.mechanisms オプションを SCRAM-SHA-256 または SCRAM-SHA-512 に設定します。

     以下に例を示します。

     listeners=INSECURE://:9092,AUTHENTICATED://:9093,REPLICATION://:9094
     listener.security.protocol.map=INSECURE:PLAINTEXT,AUTHENTICATED:SASL_PLAINTEXT,REPLICATION:PLAINTEXT
     sasl.enabled.mechanisms=SCRAM-SHA-512

     Copy to Clipboard Toggle word wrap

3. KAFKA_OPTS 環境変数を使用して Kafka ブローカーを (再) 起動し、JAAS 構成を Kafka ブローカーに渡します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   Copy to Clipboard Toggle word wrap

1. AMQ Streams クラスターおよび Kafka クライアントの OAuth 2.0 承認サーバーを設定します。
2. OAuth 2.0 を使用するよう設定された Kafka ブローカーリスナーで、Kafka クラスターをデプロイまたは更新します。
3. OAuth 2.0 を使用するように Java ベースの Kafka クライアントを更新します。

手順

1. Red Hat Single Sign-On の Admin Console にアクセスするか、Red Hat Single Sign-On の Admin CLI を使用して、OAuth 2.0 認証の設定時に作成した Kafka ブローカークライアントの Authorization Services を有効にします。
2. 承認サービスを使用して、クライアントのリソース、承認スコープ、ポリシー、およびパーミッションを定義します。
3. ロールとグループをユーザーとクライアントに割り当てて、パーミッションをユーザーとクライアントにバインドします。

4. Red Hat Single Sign-On 承認を使用するように Kafka ブローカーを設定します。

   以下を Kafka server.properties 設定ファイルに追加し、Kafka にオーソライザーをインストールします。

   authorizer.class.name=io.strimzi.kafka.oauth.server.authorizer.KeycloakRBACAuthorizer
   principal.builder.class=io.strimzi.kafka.oauth.server.authorizer.JwtKafkaPrincipalBuilder

   Copy to Clipboard Toggle word wrap

5. Kafka ブローカーの設定を追加して、承認サーバーおよび Authorization Services にアクセスします。

   ここでは、server.properties への追加プロパティーとして追加される設定例を示しますが、大文字で始める、または大文字の命名規則を使用して、環境変数として定義することもできます。

   strimzi.authorization.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/auth/realms/REALM-NAME/protocol/openid-connect/token" 

   1

   
   strimzi.authorization.client.id="kafka" 

   2

   Copy to Clipboard Toggle word wrap

   1

   Red Hat Single Sign-On への OAuth 2.0 トークンエンドポイントの URL。実稼働環境では、常に HTTPS を使用してください

   2

   承認サービスが有効になっている Red Hat Single Sign-On の OAuth 2.0 クライアント定義のクライアント ID。通常、kafka が ID として使用されます。

6. (オプション) 特定の Kafka クラスターの構成を追加します。

   以下に例を示します。

   strimzi.authorization.kafka.cluster.name="kafka-cluster" 

   1

   Copy to Clipboard Toggle word wrap

   1

   特定の Kafka クラスターの名前。名前はパーミッションをターゲットにするために使用され、同じ Red Hat シングルサインオンレルム内で複数のクラスターを管理できるようにします。デフォルト値は kafka-cluster です。

7. (オプション) 単純許可に委任します。

   以下に例を示します。

   strimzi.authorization.delegate.to.kafka.acl="false" 

   1

   Copy to Clipboard Toggle word wrap

   1

   Red Hat Single Sign-On Authorization Services ポリシーでアクセスが拒否された場合、Kafka AclAuthorizer に権限を委任します。デフォルトは false です。

8. (オプション) TLS 接続の構成を許可サーバーに追加します。

   以下に例を示します。

   strimzi.authorization.ssl.truststore.location=<path-to-truststore> 

   1

   
   strimzi.authorization.ssl.truststore.password=<my-truststore-password> 

   2

   
   strimzi.authorization.ssl.truststore.type=JKS 

   3

   
   strimzi.authorization.ssl.secure.random.implementation=SHA1PRNG 

   4

   
   strimzi.authorization.ssl.endpoint.identification.algorithm=HTTPS 

   5

   Copy to Clipboard Toggle word wrap

   1

   証明書が含まれるトラストストアへのパス。

   2

   トラストストアのパスワード。

   3

   トラストストアのタイプ。設定されていない場合は、デフォルトの Java キーストアタイプが使用されます。

   4

   乱数ジェネレーターの実装。設定されていない場合は、Java プラットフォーム SDK デフォルトが使用されます。

   5

   ホスト名の検証。空の文字列に設定すると、ホスト名の検証はオフになります。設定されていない場合、デフォルト値は HTTPS で、サーバー証明書のホスト名の検証を強制します。

9. (オプション) 許可サーバーからの許可の更新を構成します。付与更新ジョブは、アクティブなトークンを列挙し、それぞれに最新の付与を要求することで機能します。

   以下に例を示します。

   strimzi.authorization.grants.refresh.period.seconds="120" 

   1

   
   strimzi.authorization.grants.refresh.pool.size="10" 

   2

   Copy to Clipboard Toggle word wrap

   1

   許可サーバーからの許可のリストが更新される頻度を指定します (デフォルトでは 1 分に 1 回)。デバッグの目的で付与の更新をオフにするには、"0" に設定します。

   2

   付与更新ジョブで使用されるスレッドプールのサイズ (並列度) を指定します。デフォルト値は "5" です。

10. クライアントまたは特定のロールを持つユーザーとして Kafka ブローカーにアクセスして、設定したパーミッションを検証し、必要なアクセス権限があり、付与されるべきでないアクセス権限がないことを確認します。