MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Kafka 的权限管理与安全配置

2021-09-144.4k 阅读

Kafka 权限管理概述

Kafka 作为一款高吞吐量的分布式消息队列系统，在企业级应用中广泛使用。随着数据的敏感性和安全性要求不断提高，Kafka 的权限管理变得至关重要。Kafka 的权限管理主要用于控制不同用户或客户端对 Kafka 集群中各种资源（如主题、消费者组等）的访问。

访问控制列表（ACL）

Kafka 通过访问控制列表（ACL）来实现权限管理。ACL 定义了哪些主体（用户、服务账号等）可以对哪些资源（主题、消费者组）执行哪些操作（读取、写入、管理等）。主体可以是 Kerberos 主体、SSL 证书中的通用名（CN）或简单的用户名。资源则涵盖了 Kafka 集群中的主题、消费者组以及集群操作等。操作类型包括 Read（读取消息）、Write（写入消息）、Create（创建主题）、Delete（删除主题）、Describe（查看主题或消费者组元数据）等。

例如，一个常见的场景是允许特定的应用程序用户组对某些业务主题具有读写权限，而只允许管理员用户对所有主题具有完全的管理权限。

Kafka 安全配置基础

在深入探讨 Kafka 的权限管理之前，了解其基本的安全配置是必要的。

身份验证

身份验证是 Kafka 安全的第一道防线，确保只有合法的客户端可以连接到 Kafka 集群。Kafka 支持多种身份验证机制，包括 PLAIN 文本、SASL/SCRAM 和 SSL。

  1. PLAIN 文本身份验证
    • PLAIN 文本身份验证是最简单的方式，客户端直接在连接请求中发送用户名和密码。虽然简单，但安全性较低，因为用户名和密码是以明文形式传输的。
    • 配置步骤：
      • 在 server.properties 文件中，启用 PLAIN 文本身份验证机制：
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN

 - 配置 SASL 服务器端配置，例如指定 JAAS 配置文件路径：

listeners=SASL_PLAINTEXT://your.host.name:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="admin - password" \
  user_admin="admin - password" \
  user_alice="alice - password";
  • 在客户端，需要配置相应的 JAAS 配置。例如，对于 Java 客户端：
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;

public class PlainTextConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your.host.name:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test - group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put("sasl.mechanism", "PLAIN");
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required \
          username=\"alice\" \
          password=\"alice - password\";");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test - topic"));
        while (true) {
            consumer.poll(100).forEach(record -> {
                System.out.println("Received message: " + record.value());
            });
        }
    }
}
  1. SASL/SCRAM 身份验证
    • SASL/SCRAM（Salted Challenge Response Authentication Mechanism）是一种更安全的身份验证机制，它通过挑战 - 响应的方式进行身份验证，避免了密码以明文形式传输。
    • 配置步骤：
      • 在 server.properties 文件中，启用 SASL/SCRAM 机制：
sasl.enabled.mechanisms=SCRAM - SHA - 256,SCRAM - SHA - 512
sasl.mechanism.inter.broker.protocol=SCRAM - SHA - 256

 - 配置 SASL 服务器端 JAAS 配置：

listeners=SASL_PLAINTEXT://your.host.name:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="admin" \
  password="admin - password" \
  user_admin="admin - password" \
  user_alice="alice - password";
  • 客户端配置类似 PLAIN 文本身份验证，但机制和 JAAS 配置不同：
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;

public class ScramConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your.host.name:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test - group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put("sasl.mechanism", "SCRAM - SHA - 256");
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required \
          username=\"alice\" \
          password=\"alice - password\";");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test - topic"));
        while (true) {
            consumer.poll(100).forEach(record -> {
                System.out.println("Received message: " + record.value());
            });
        }
    }
}
  1. SSL 身份验证
    • SSL 身份验证通过使用 SSL/TLS 协议来加密通信，并通过证书进行身份验证。这不仅提供了数据传输的加密，还确保了客户端和服务器的身份真实性。
    • 配置步骤：
      • 生成服务器端证书和密钥：
        • 使用 keytool 工具生成密钥库：
keytool -genkey -alias kafka -keyalg RSA -keystore kafka.keystore.jks -storepass password -keypass password -dname "CN=your.host.name, OU=your - org - unit, O=your - org, L=your - city, S=your - state, C=your - country"

   - 生成证书签名请求（CSR）：

keytool -certreq -alias kafka -keystore kafka.keystore.jks -file kafka.csr

   - 使用证书颁发机构（CA）签署 CSR 并获取证书，然后将 CA 证书和签署的证书导入密钥库：

keytool -import -alias CARoot -keystore kafka.keystore.jks -file ca.crt
keytool -import -alias kafka -keystore kafka.keystore.jks -file signed - kafka.crt

 - 在 `server.properties` 文件中配置 SSL 相关参数：

listeners=SSL://your.host.name:9093
security.inter.broker.protocol=SSL
ssl.keystore.location=/path/to/kafka.keystore.jks
ssl.keystore.password=password
ssl.key.password=password
ssl.truststore.location=/path/to/kafka.truststore.jks
ssl.truststore.password=password

 - 客户端配置：
   - 生成客户端密钥库和证书，类似服务器端步骤。
   - 配置客户端 `properties` 文件：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;

public class SSLConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your.host.name:9093");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test - group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put("security.protocol", "SSL");
        props.put("ssl.keystore.location", "/path/to/client.keystore.jks");
        props.put("ssl.keystore.password", "password");
        props.put("ssl.key.password", "password");
        props.put("ssl.truststore.location", "/path/to/client.truststore.jks");
        props.put("ssl.truststore.password", "password");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test - topic"));
        while (true) {
            consumer.poll(100).forEach(record -> {
                System.out.println("Received message: " + record.value());
            });
        }
    }
}

数据加密

除了身份验证，Kafka 还支持数据加密，确保在传输过程中数据的保密性。

  1. 传输层加密
    • 通过 SSL/TLS 协议，Kafka 可以对客户端与服务器之间以及服务器节点之间的数据传输进行加密。如上述 SSL 身份验证配置中，启用 SSL 后，数据在传输过程中就已经被加密。
  2. 存储加密
    • Kafka 也支持对存储在磁盘上的数据进行加密。这可以通过配置 log.segment.encryptor.class 属性来实现。例如，使用 org.apache.kafka.common.security.crypto.CiphertextSegmentEncryptor 类进行加密。
    • 配置步骤：
      • 在 server.properties 文件中添加：
log.segment.encryptor.class=org.apache.kafka.common.security.crypto.CiphertextSegmentEncryptor

 - 配置加密密钥管理，例如使用 `kafka - configs.sh` 命令设置加密密钥：

./bin/kafka - configs.sh --zookeeper your.zookeeper.host:2181 --alter --add - config \
  'SCRAM - SHA - 256=[password=your - password]' --entity - type users --entity - name alice

Kafka 权限管理实践

创建和管理 ACL

  1. 使用 Kafka 命令行工具
    • Kafka 提供了 kafka - acl.sh 命令行工具来管理 ACL。例如，要允许用户 alice 对主题 test - topic 具有读取权限，可以使用以下命令：
./bin/kafka - acl.sh --authorizer - props zookeeper.connect=your.zookeeper.host:2181 \
  --add --allow - principal User:alice --operation Read --topic test - topic
  • 查看主题 test - topic 的 ACL 配置：
./bin/kafka - acl.sh --authorizer - props zookeeper.connect=your.zookeeper.host:2181 \
  --list --topic test - topic
  • 删除用户 alice 对主题 test - topic 的读取权限：
./bin/kafka - acl.sh --authorizer - props zookeeper.connect=your.zookeeper.host:2181 \
  --remove --allow - principal User:alice --operation Read --topic test - topic
  1. 使用 Kafka 管理客户端 API
    • 在 Java 中，可以使用 AdminClient 来管理 ACL。以下是一个示例代码，用于创建一个允许用户 alice 对主题 test - topic 具有读取权限的 ACL：
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AclBinding;
import org.apache.kafka.clients.admin.AclBindingFilter;
import org.apache.kafka.clients.admin.AclBindingOperation;
import org.apache.kafka.clients.admin.CreateAclsResult;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AccessControlEntryFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

public class KafkaAclManager {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "your.host.name:9092");
        AdminClient adminClient = AdminClient.create(configs);

        ResourcePattern resourcePattern = new ResourcePattern(ResourceType.TOPIC, "test - topic", PatternType.LITERAL);
        AccessControlEntry accessControlEntry = new AccessControlEntry("User:alice", null, AclOperation.READ, AclPermissionType.ALLOW);
        AclBinding aclBinding = new AclBinding(resourcePattern, accessControlEntry);
        List<AclBindingOperation> operations = Collections.singletonList(new AclBindingOperation(AclBindingOperation.Operation.CREATE, aclBinding));
        CreateAclsResult result = adminClient.createAcls(operations);
        result.all().get();

        adminClient.close();
    }
}

基于角色的访问控制（RBAC）

虽然 Kafka 原生没有直接支持基于角色的访问控制，但可以通过一些自定义的方式来模拟实现。例如，可以创建不同的用户组，并为每个用户组分配一组 ACL。

  1. 定义角色和用户组
    • 假设定义两个角色：producer - role 和 consumer - role。producer - role 对应的用户组为 producer - group，consumer - role 对应的用户组为 consumer - group。
  2. 分配 ACL 到角色
    • 对于 producer - role，为 producer - group 中的用户分配对相关主题的写入权限：
./bin/kafka - acl.sh --authorizer - props zookeeper.connect=your.zookeeper.host:2181 \
  --add --allow - principal User:producer - group --operation Write --topic business - topic
  • 对于 consumer - role，为 consumer - group 中的用户分配对相关主题的读取权限：
./bin/kafka - acl.sh --authorizer - props zookeeper.connect=your.zookeeper.host:2181 \
  --add --allow - principal User:consumer - group --operation Read --topic business - topic

Kafka 权限管理的高级话题

跨集群权限管理

在多 Kafka 集群的场景下，可能需要统一管理权限。一种方法是使用外部的身份验证和授权服务，如 LDAP 或 Active Directory，并通过自定义的 Kafka 授权插件来集成。

  1. 集成 LDAP
    • 配置 Kafka 服务器使用 LDAP 进行身份验证和授权。首先，在 server.properties 文件中配置 LDAP 相关参数：
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="admin" \
  password="admin - password" \
  user_admin="admin - password";
authorizer.class.name=org.apache.kafka.authorizer.AclAuthorizer
authorizer.acl.default.entry=User:ANONYMOUS:Denied:All
authorizer.ldap.url=ldap://your.ldap.server:389
authorizer.ldap.userSearchBase=ou=users,dc=your,dc=com
authorizer.ldap.userSearchFilter=(uid={0})
authorizer.ldap.groupSearchBase=ou=groups,dc=your,dc=com
authorizer.ldap.groupSearchFilter=(member={0})
  • 客户端配置类似，需要配置通过 LDAP 进行身份验证的相关参数。

动态权限管理

在一些场景中，需要根据运行时的条件动态调整权限。例如，根据业务负载动态授予或撤销某些用户对主题的读写权限。

  1. 使用 Kafka Streams 实现动态权限管理
    • 可以使用 Kafka Streams 来监听特定的控制主题，当控制主题中出现权限变更消息时，通过 Kafka 管理客户端 API 动态调整 ACL。以下是一个简单的示例：
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

public class DynamicAclManager {
    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "dynamic - acl - manager");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your.host.name:9092");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, StringDeserializer.class);
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, StringDeserializer.class);

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> controlStream = builder.stream("control - topic", Consumed.with(StringDeserializer.class, StringDeserializer.class));
        controlStream.foreach((key, value) -> {
            // 解析 value 中的权限变更信息，例如 "add|User:alice|Read|test - topic"
            String[] parts = value.split("|");
            if ("add".equals(parts[0])) {
                // 使用 Kafka 管理客户端 API 添加 ACL
            } else if ("remove".equals(parts[0])) {
                // 使用 Kafka 管理客户端 API 删除 ACL
            }
        });

        KafkaStreams streams = new KafkaStreams(builder.build(), config);
        CountDownLatch latch = new CountDownLatch(1);

        Runtime.getRuntime().addShutdownHook(new Thread("streams - shutdown - hook") {
            @Override
            public void run() {
                streams.close();
                latch.countDown();
            }
        });

        try {
            streams.start();
            latch.await();
        } catch (InterruptedException e) {
            System.exit(1);
        }
    }
}

权限管理与监控

监控权限相关指标

为了确保 Kafka 权限管理的有效性，监控与权限相关的指标是必要的。

  1. 使用 Kafka 自带的 JMX 指标
    • Kafka 通过 JMX 暴露了一些与权限相关的指标，例如 kafka.server:type=Authorizer,name=AclCacheMissRate 指标表示 ACL 缓存未命中率。可以使用工具如 jconsole 或 Prometheus + Grafana 来监控这些指标。
    • 配置 Kafka 启用 JMX：
      • 在 kafka - env.sh 文件中添加：
export JMX_PORT="9999"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Djava.rmi.server.hostname=your.host.name"
  • 使用 jconsole 连接到 Kafka 服务器的 JMX 端口（9999），可以查看权限相关指标。
  1. 自定义监控指标
    • 可以通过 Kafka 的自定义指标功能，添加与权限管理相关的自定义指标。例如，统计特定用户或用户组在一段时间内的权限操作次数。
    • 在 Kafka 服务器代码中，可以使用 KafkaMetricsGroup 来注册自定义指标。以下是一个简单示例：
import org.apache.kafka.common.metrics.MetricName;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Counter;

public class CustomAclMetrics {
    private final Sensor aclOperationSensor;

    public CustomAclMetrics(Metrics metrics) {
        MetricName metricName = metrics.metricName("acl - operation - count", "authorizer", "Number of ACL operations");
        aclOperationSensor = metrics.sensor("acl - operation - sensor");
        aclOperationSensor.add(metricName, new Counter());
    }

    public void incrementAclOperationCount() {
        aclOperationSensor.record(1);
    }
}

审计日志

审计日志记录了 Kafka 权限相关的操作，对于安全审计和故障排查非常重要。

  1. 启用审计日志
    • 在 server.properties 文件中配置审计日志：
audit.logger=kafka.audit.logger
audit.log.dir=/path/to/audit/logs
audit.log.retention.hours=24
audit.log.retention.check.interval.hours=1
  • 配置审计日志的格式，例如在 log4j.properties 文件中：
kafka.audit.logger=INFO, kafkaAuditAppender
kafka.audit.appender.kafkaAuditAppender=org.apache.log4j.DailyRollingFileAppender
kafka.audit.appender.kafkaAuditAppender.File=/path/to/audit/logs/kafka - audit.log
kafka.audit.appender.kafkaAuditAppender.layout=org.apache.log4j.PatternLayout
kafka.audit.appender.kafkaAuditAppender.layout.ConversionPattern=%d{yyyy - MM - dd HH:mm:ss} %p %c{1}:%L - %m%n
  • 审计日志会记录诸如用户对主题的读写操作、ACL 的创建和删除等信息，有助于追溯权限管理相关的活动。

通过上述对 Kafka 权限管理与安全配置的详细介绍，包括身份验证、数据加密、权限管理实践以及高级话题和监控等方面，希望能帮助开发者全面掌握 Kafka 的安全配置和权限管理，确保 Kafka 集群在企业级应用中的安全稳定运行。在实际应用中，需要根据具体的业务需求和安全要求，灵活选择和配置 Kafka 的安全机制和权限管理策略。