在 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 节点。要将 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

   超级用户 自动具有 priviledges 管理员。该文件可以包含多个用户，但 Kafka 代理只需要一个额外的用户。Kafka 用户的建议名称为 kafka。

   以下示例显示了 客户端到服务器 身份验证的服务器上下文：

   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

   必须为作为 ZooKeeper 集群一部分的每个节点添加 authProvider. <ID > 属性。一个三节点 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 节点。要将 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 代理节点。要将 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. 编辑 /opt/kafka/config/server.properties Kafka 配置文件，以使用 AclAuthorizer。

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

   Copy to Clipboard Toggle word wrap
2. （重新）启动 Kafka 代理。

流程

1. 使用 --add 选项运行 kafka-acls.sh。

   示例：

   • 允许使用 MyConsumerGroup 消费者组从 myTopic 读取 user1 和 user2 访问权限。

     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

   • 添加 user1 作为带有 MyConsumerGroup 的 myTopic 的消费者。

     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 代理。

3. 使用 zookeeper-security-migration.sh 工具在所有现有 ZooKeeper 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 中有其公告和 bootstrap 地址。

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. 更新基于 Java 的 Kafka 客户端以使用 OAuth 2.0