为 Kafka 配置 SASL 和 SSL

2023-10-08 13:49:05

保护 Kafka 通信：SASL 和 SSL 深入指南

简介

随着数据传输的日益普及，确保通信安全变得至关重要。对于像 Apache Kafka 这样的分布式流式处理平台，保障集群中通信的安全至关重要。SASL（安全认证层）和 SSL（安全套接字层）协议是两项关键的安全机制，可用于为 Kafka 通信提供身份验证和加密。在这篇深入的指南中，我们将探索如何在 Kafka 中配置 SASL 和 SSL，以提升通信安全性。

SASL 配置

SASL 是一种身份验证协议，用于在客户端和服务器之间建立安全通道。Kafka 支持多种 SASL 机制，例如 PLAIN、SCRAM-SHA-512 和 GSSAPI。在本节中，我们将重点关注 PLAIN 机制，因为它是最简单的实现。

要在 Kafka 中配置 SASL，需要在 server.properties 文件中添加以下属性：

# Kafka Broker SASL 认证
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="secret";

security.protocol：指定使用的安全协议。
sasl.mechanism：指定使用的 SASL 机制。
sasl.jaas.config：配置 JAAS（Java 身份验证和授权服务）登录模块，用于定义身份验证的用户名和密码。

代码示例：

// 创建 Kafka 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("security.protocol", "SASL_SSL");
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"secret\";");

// 创建 Kafka 生产者
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

SSL 配置

SSL 提供加密通信，以保护客户端和服务器之间传输的数据。要在 Kafka 中配置 SSL，需要在 server.properties 文件中添加以下属性：

# Kafka Broker SSL 认证
security.protocol=SASL_SSL
ssl.protocol=TLSv1.2
ssl.keystore.location=kafka.keystore.jks
ssl.keystore.password=secret
ssl.truststore.location=kafka.truststore.jks
ssl.truststore.password=secret

security.protocol：指定使用的安全协议。
ssl.protocol：指定使用的 SSL 协议。
ssl.keystore.location：指定包含私钥的密钥库的位置。
ssl.keystore.password：指定密钥库的密码。
ssl.truststore.location：指定包含受信任证书的信任库的位置。
ssl.truststore.password：指定信任库的密码。

代码示例：

// 创建 SSL 上下文
SSLContext sslContext = SSLContext.getInstance("TLS");
KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
keyManagerFactory.init(keyStore, password.toCharArray());
TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("SunX509");
trustManagerFactory.init(trustStore);
sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), new SecureRandom());

// 创建 SSL 套接字工厂
SSLSocketFactory socketFactory = sslContext.getSocketFactory();