安全配置

Apache Kafka 支持客户端和代理之间建立安全连接。要利用此功能，请遵循 Apache Kafka 文档以及 Kafka 0.9 Confluent 文档中的安全指南。使用 spring.cloud.stream.kafka.binder.configuration 选项为绑定器创建的所有客户端设置安全属性。例如，要将 security.protocol 设置为 SASL_SSL，请设置以下属性：

spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL

所有其他安全属性都可以通过类似的方式设置。使用 Kerberos 时，请遵循参考文档中的说明来创建和引用 JAAS 配置。 Spring Cloud Stream 支持通过使用 JAAS 配置文件和 Spring Boot 属性将 JAAS 配置信息传递给应用程序。

使用 JAAS 配置文件

可以通过使用系统属性为 Spring Cloud Stream 应用程序设置 JAAS 和（可选）krb5 文件的位置。以下示例展示了如何使用 JAAS 配置文件通过 SASL 和 Kerberos 启动 Spring Cloud Stream 应用程序：

 java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \
   --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT

使用 Spring Boot 属性

作为 JAAS 配置文件的替代方案，Spring Cloud Stream 提供了一种机制，可以通过使用 Spring Boot 属性为 Spring Cloud Stream 应用程序设置 JAAS 配置。

以下属性可用于配置 Kafka 客户端的登录上下文：

spring.cloud.stream.kafka.binder.jaas.loginModule: 登录模块名称。通常情况下无需设置。

默认值：com.sun.security.auth.module.Krb5LoginModule。
spring.cloud.stream.kafka.binder.jaas.controlFlag: 登录模块的控制标志。

默认值：required。
spring.cloud.stream.kafka.binder.jaas.options: 包含登录模块选项的键/值对映射。

默认值：空映射。

以下示例展示了如何使用 Spring Boot 配置属性通过 SASL 和 Kerberos 启动 Spring Cloud Stream 应用程序：

 java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \
   --spring.cloud.stream.bindings.input.destination=stream.ticktock \
   --spring.cloud.stream.kafka.binder.autoCreateTopics=false \
   --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \
   --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \
   --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \
   --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \
   --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM

前面的示例表示以下 JAAS 文件的等效内容：

KafkaClient {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_client.keytab"
    principal="kafka-client-1@EXAMPLE.COM";
};

如果所需的 topic 已存在于代理上或将由管理员创建，则可以关闭自动创建，并且只需要发送客户端 JAAS 属性。

不要在同一应用程序中混合使用 JAAS 配置文件和 Spring Boot 属性。如果 -Djava.security.auth.login.config 系统属性已存在，Spring Cloud Stream 将忽略 Spring Boot 属性。

在使用 autoCreateTopics 和 autoAddPartitions 与 Kerberos 时要小心。通常，应用程序可能使用在 Kafka 和 Zookeeper 中没有管理权限的主体。因此，依赖 Spring Cloud Stream 创建/修改 topic 可能会失败。在安全环境中，我们强烈建议使用 Kafka 工具以管理方式创建 topic 和管理 ACL。

多绑定器配置和 JAAS

当连接到多个集群时，如果每个集群都需要单独的 JAAS 配置，则使用属性 sasl.jaas.config 设置 JAAS 配置。当此属性存在于应用程序中时，它将优先于上述其他策略。有关更多详细信息，请参阅此 KIP-85。

例如，如果您的应用程序中有两个集群，每个集群都需要单独的 JAAS 配置，那么以下是您可以使用的模板：

spring.cloud.stream:
    binders:
        kafka1:
          type: kafka
          environment:
             spring:
               cloud:
                 stream:
                  kafka:
                    binder:
                      brokers: localhost:9092
                      configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\";"
        kafka2:
          type: kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      brokers: localhost:9093
                      configuration.sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user1\" password=\"user1-secret\";"
    kafka.binder:
        configuration:
          security.protocol: SASL_PLAINTEXT
          sasl.mechanism: PLAIN

请注意，在上述配置中，两个 Kafka 集群及其各自的 sasl.jaas.config 值都不同。

有关如何设置和运行此类应用程序的更多详细信息，请参阅此示例应用程序。