Kafka安全配置实战指南

Kafka安全配置实战指南

引言

在生产环境中，Kafka的安全配置是保护数据安全的重要措施。Kafka提供了完善的安全机制，包括认证（Authentication）、授权（Authorization）、加密（Encryption）等功能。本文将详细介绍Kafka的安全配置方案，帮助开发者构建安全的Kafka集群。

认证机制

1.1 SSL/TLS认证

# server.properties – SSL配置

# SSL监听器配置
listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093

# SSL证书配置
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.keystore.type=JKS
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=test1234
ssl.truststore.type=JKS

# SSL协议和加密套件
ssl.protocol=TLS
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.secure.random.implementation=SHA1PRNG

# 客户端认证配置
ssl.client.auth=required

# 信任证书配置
ssl.endpoint.identification.algorithm=HTTPS

# 生成SSL证书
keytool -keystore kafka.server.keystore.jks \\
-alias localhost \\
-validity 365 \\
-keyalg RSA \\
-storepass test1234 \\
-keypass test1234 \\
-dname "CN=localhost"

1.2 SASL认证

# server.properties – SASL配置

# 启用SASL
listeners=PLAINTEXT://localhost:9092,SASL_SSL://localhost:9093

# SASL机制
sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512

# JAAS配置
sasl.mechanism.inter.broker.protocol=PLAIN

# SASL协议配置
security.inter.broker.protocol=SASL_SSL

# JAAS配置文件路径
authorizer.class.name=kafka.security.authorizer.AclAuthorizer

# kafka_server_jaas.conf

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret"
user_admin="admin-secret"
user_user1="user1-secret";
};

1.3 生产者安全配置

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Properties;

public class SecureProducer {

public static Properties createSSLProducerConfig() {
Properties props = new Properties();

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");

// SSL配置
props.put("security.protocol", "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
"/var/private/ssl/kafka.client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
"test1234");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
"/var/private/ssl/kafka.client.keystore.jks");
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,
"test1234");
props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG,
"test1234");
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,
"HTTPS");

return props;
}

public static Properties createSASLProducerConfig() {
Properties props = new Properties();

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");

// SASL配置
props.put("security.protocol", "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG,
"org.apache.kafka.common.security.plain.PlainLoginModule " +
"required username=\\"user1\\" password=\\"user1-secret\\";");

// SSL配置
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
"/var/private/ssl/kafka.client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
"test1234");

return props;
}

public static void demonstrateSecureProducer() {
KafkaProducer<String, String> producer =
new KafkaProducer<>(createSASLProducerConfig());

ProducerRecord<String, String> record =
new ProducerRecord<>("secure-topic", "key", "value");

producer.send(record, (metadata, exception) -> {
if (exception != null) {
System.err.println("发送失败: " + exception);
} else {
System.out.println("发送成功: " + metadata.offset());
}
});

producer.close();
}
}

1.4 消费者安全配置

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Properties;

public class SecureConsumer {

public static Properties createSASLConsumerConfig() {
Properties props = new Properties();

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "secure-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");

// SASL配置
props.put("security.protocol", "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG,
"org.apache.kafka.common.security.plain.PlainLoginModule " +
"required username=\\"user1\\" password=\\"user1-secret\\";");

// SSL配置
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
"/var/private/ssl/kafka.client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
"test1234");

return props;
}

public static void demonstrateSecureConsumer() {
KafkaConsumer<String, String> consumer =
new KafkaConsumer<>(createSASLConsumerConfig());

consumer.subscribe(java.util.Collections.singletonList("secure-topic"));

while (true) {
ConsumerRecords<String, String> records =
consumer.poll(java.time.Duration.ofMillis(1000));

for (ConsumerRecord<String, String> record : records) {
System.out.println(record.value());
}
}
}
}

授权机制

2.1 ACL访问控制

# server.properties – 启用ACL

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

# 超级用户配置
super.users=User:admin

# 允许匿名访问（谨慎使用）
allow.everyone.if.no.acl.found=false

# 添加ACL规则
kafka-acls.sh –bootstrap-server localhost:9093 \\
–add –allow-principal User:user1 \\
–operation Read –operation Write \\
–topic secure-topic

# 添加主题级别的读权限
kafka-acls.sh –bootstrap-server localhost:9093 \\
–add –allow-principal User:user1 \\
–operation Read \\
–topic secure-topic

# 添加消费者组权限
kafka-acls.sh –bootstrap-server localhost:9093 \\
–add –allow-principal User:user1 \\
–operation Read \\
–group secure-group

# 添加集群级别权限
kafka-acls.sh –bootstrap-server localhost:9093 \\
–add –allow-principal User:user1 \\
–operation ClusterAction \\
–cluster

# 查看ACL
kafka-acls.sh –bootstrap-server localhost:9093 \\
–list –principal User:user1

# 移除ACL
kafka-acls.sh –bootstrap-server localhost:9093 \\
–remove –allow-principal User:user1 \\
–operation Read \\
–topic secure-topic

2.2 Java ACL配置

import org.apache.kafka.clients.admin.*;
import java.util.*;

public class ACLManagement {

public static void createACLs() throws Exception {
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");
adminProps.put("security.protocol", "SASL_SSL");
adminProps.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
adminProps.put(SaslConfigs.SASL_JAAS_CONFIG,
"org.apache.kafka.common.security.plain.PlainLoginModule " +
"required username=\\"admin\\" password=\\"admin-secret\\";");

try (AdminClient adminClient = AdminClient.create(adminProps)) {

// 创建资源ACL
List<AclBinding> acls = new ArrayList<>();

// 允许用户读取主题
acls.add(new AccessControlEntry(
"User:user1", "*",
ResourcePattern.WILDCARD_RESOURCE,
AclOperation.READ));

// 允许用户写入主题
acls.add(new AccessControlEntry(
"User:user1", "*",
ResourcePattern.WILDCARD_RESOURCE,
AclOperation.WRITE));

// 创建ACL
AccessControlEntry ace = new AccessControlEntry(
"User:user1", "*",
ResourcePattern.WILDCARD_RESOURCE,
AclOperation.READ);

ResourcePattern resource = new ResourcePattern(
ResourceType.TOPIC, "secure-topic", PatternType.LITERAL);

AclBinding acl = new AclBinding(resource, ace);

CreateAclsResult result = adminClient.createAcls(
Collections.singleton(acl));

result.all().get();
System.out.println("ACL创建成功");
}
}

public static void listACLs() throws Exception {
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");

try (AdminClient adminClient = AdminClient.create(adminProps)) {

DescribeAclsResult result = adminClient.describeAcls(
AclBindingFilter.WILDCARD_RESOURCE_FILTER);

Collection<AclBinding> acls = result.all().get();

System.out.println("=== ACL列表 ===");
for (AclBinding acl : acls) {
System.out.println("Principal: " + acl.principal());
System.out.println("Resource: " + acl.resource());
System.out.println("Operation: " + acl.operation());
System.out.println("Permission: " + acl.permission());
System.out.println();
}
}
}
}

加密配置

3.1 传输加密

# 完整的加密配置示例

# 监听器配置
listeners=SSL://localhost:9093,SASL_SSL://localhost:9094

# SSL配置
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=test1234

# TLS版本
ssl.enabled.protocols=TLSv1.2,TLSv1.3

# 加密套件
ssl.cipher.suites=TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384

# 客户端认证
ssl.client.auth=required

# SASL配置
sasl.enabled.mechanisms=SCRAM-SHA-512

# 内部通信加密
security.inter.broker.protocol=SASL_SSL

3.2 端到端加密

import javax.crypto.Cipher;
import javax.crypto.KeyGenerator;
import javax.crypto.SecretKey;
import javax.crypto.spec.GCMParameterSpec;
import java.security.SecureRandom;
import java.util.Base64;

public class EndToEndEncryption {

private static final int GCM_IV_LENGTH = 12;
private static final int GCM_TAG_LENGTH = 128;

public static SecretKey generateKey() throws Exception {
KeyGenerator keyGen = KeyGenerator.getInstance("AES");
keyGen.init(256);
return keyGen.generateKey();
}

public static String encrypt(String plaintext, SecretKey key)
throws Exception {

byte[] iv = new byte[GCM_IV_LENGTH];
SecureRandom random = new SecureRandom();
random.nextBytes(iv);

Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
GCMParameterSpec parameterSpec =
new GCMParameterSpec(GCM_TAG_LENGTH, iv);

cipher.init(Cipher.ENCRYPT_MODE, key, parameterSpec);

byte[] ciphertext = cipher.doFinal(
plaintext.getBytes("UTF-8"));

byte[] combined = new byte[iv.length + ciphertext.length];
System.arraycopy(iv, 0, combined, 0, iv.length);
System.arraycopy(ciphertext, 0, combined, iv.length, ciphertext.length);

return Base64.getEncoder().encodeToString(combined);
}

public static String decrypt(String encrypted, SecretKey key)
throws Exception {

byte[] combined = Base64.getDecoder().decode(encrypted);

byte[] iv = new byte[GCM_IV_LENGTH];
byte[] ciphertext = new byte[combined.length – GCM_IV_LENGTH];

System.arraycopy(combined, 0, iv, 0, iv.length);
System.arraycopy(combined, iv.length, ciphertext, 0, ciphertext.length);

Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
GCMParameterSpec parameterSpec =
new GCMParameterSpec(GCM_TAG_LENGTH, iv);

cipher.init(Cipher.DECRYPT_MODE, key, parameterSpec);

byte[] plaintext = cipher.doFinal(ciphertext);
return new String(plaintext, "UTF-8");
}
}

安全最佳实践

4.1 生产环境配置

# 生产环境完整安全配置

# 监听器配置
listeners=INTERNAL://internal-host:9092,EXTERNAL://external-host:9093
advertised.listeners=INTERNAL://internal-host:9092,EXTERNAL://external-host:9093
listener.security.protocol.map=INTERNAL:SASL_PLAINTEXT,EXTERNAL:SSL

# SSL配置
ssl.keystore.location=/etc/kafka/secrets/kafka.server.keystore.jks
ssl.keystore.password=${KAFKA_SSL_PASSWORD}
ssl.key.password=${KAFKA_SSL_KEY_PASSWORD}
ssl.truststore.location=/etc/kafka/secrets/kafka.server.truststore.jks
ssl.truststore.password=${KAFKA_SSL_TRUST_PASSWORD}
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.cipher.suites=TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256

# SASL配置
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# ACL配置
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:kafka-admin

# 日志配置
audit.logger=REFINEDKafkaCommitLogAppender

# 安全协议
security.inter.broker.protocol=SASL_SSL

4.2 密钥轮换

# 轮换SSL证书

# 1. 生成新证书
keytool -keystore kafka.server.keystore.jks \\
-alias new-server \\
-validity 365 \\
-genkeypair \\
-keyalg RSA \\
-storepass test1234

# 2. 导出证书签名请求
keytool -keystore kafka.server.keystore.jks \\
-alias new-server \\
-certreq \\
-file new-server.csr

# 3. 使用CA签名
openssl ca -in new-server.csr \\
-out new-server.crt \\
-cert ca-cert.pem \\
-keyfile ca-key.pem

# 4. 导入CA证书和新证书
keytool -keystore kafka.server.keystore.jks \\
-alias CARoot \\
-import -file ca-cert.pem

keytool -keystore kafka.server.keystore.jks \\
-alias new-server \\
-importcert new-server.crt

4.3 SASL密码轮换

# kafka_server_jaas.conf – 支持动态更新

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret"
user_admin="admin-secret"
user_user1="user1-secret"
user_user2="user2-secret";
};

# 使用kafka-configs.sh更新SCRAM凭据

# 添加新用户
kafka-configs.sh –bootstrap-server localhost:9093 \\
–alter \\
–add-config 'SCRAM-SHA-512=[password=strong-password]' \\
–entity-type users \\
–entity-name new-user

# 查看用户配置
kafka-configs.sh –bootstrap-server localhost:9093 \\
–describe \\
–entity-type users \\
–entity-name new-user

监控与审计

5.1 安全审计日志

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.AclBindingFilter;
import java.util.*;

public class SecurityAudit {

public static void enableAuditLogging(Properties adminProps) {
// 添加审计日志配置
adminProps.put("audit.logger",
"com.kafka.audit.LoggingAuditLogger");
adminProps.put("audit.add.for.read", "true");
adminProps.put("audit.add.for.write", "true");
adminProps.put("audit.add.for.delete", "true");
}

public static void checkUnauthorizedAccess() throws Exception {
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
"localhost:9093");

try (AdminClient adminClient = AdminClient.create(adminProps)) {

// 列出所有ACL
DescribeAclsResult result = adminClient.describeAcls(
AclBindingFilter.WILDCARD_RESOURCE_FILTER);

Map<ResourcePattern, List<AclBinding>> aclByResource =
new HashMap<>();

for (AclBinding acl : result.all().get()) {
aclByResource.computeIfAbsent(
acl.resource(), k -> new ArrayList<>())
.add(acl);
}

// 检查异常访问模式
System.out.println("=== 安全审计报告 ===");
for (Map.Entry<ResourcePattern, List<AclBinding>> entry :
aclByResource.entrySet()) {

ResourcePattern resource = entry.getKey();
List<AclBinding> acls = entry.getValue();

System.out.println("Resource: " + resource);
System.out.println("ACLs: " + acls.size());
}
}
}
}

总结

Kafka安全配置是保护数据安全的关键。通过合理配置SSL/TLS加密、SASL认证、ACL授权以及审计日志，可以构建一个安全可靠的Kafka集群。在实际应用中，需要根据业务需求和安全级别选择合适的安全机制，并定期进行安全审计和密钥轮换，确保系统的长期安全性。