进击消息中间件系列（十九）：Kafka 安全配置最佳实践

民工哥

发布于 2023-08-22 14:24:43

1.6K0

发布于 2023-08-22 14:24:43

文章被收录于专栏：民工哥技术之路

Kafka 安全性配置

Kafka 在整个大数据生态系统中扮演着核心的角色，对于系统数据的安全性要求相对较高。因此进行 Kafka 安全配置是非常必要的。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

安全配置的必要性

通过合理的安全配置，可以有效地保障 Kafka 系统数据的机密性与完整性。这样可以有效地防止信息泄漏与篡改等安全风险。

提高 Kafka 系统的可靠性

通过合理的安全配置方案，可以提高 Kafka 系统的可靠性。这是因为安全配置能够有效地降低系统出现意外故障的风险，从而保障 Kafka 系统稳定性和可靠性。

添加认证配置代码示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");

// 添加认证配置
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

在上述代码中，我们针对 KafkaConsumer 进行了安全配置。具体来说，我们添加了认证配置，即 security.protocol 和 sasl.mechanism 两个参数。这样，在 KafkaConsumer 连接到 Kafka 集群时，就会使用 SASL_PLAINTEXT 的认证方式进行身份验证。

添加 SSL 配置代码示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");

// 添加 SSL 配置
props.put("security.protocol", "SSL");
props.put("ssl.truststore.location", "/path/to/truststore/file");
props.put("ssl.truststore.password", "password");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

在上述代码中针对 KafkaProducer 进行了 SSL 配置。具体来说添加了 security.protocol、ssl.truststore.location 和 ssl.truststore.password 三个参数。这样，在 KafkaProducer 发送数据到 Kafka 集群时，就会使用 SSL 的加密方式进行数据传输，从而保障系统数据安全性。

安全性配置的要素

认证

SSL 安全协议

SSL 安全协议可以用于保障 Kafka 服务器和客户端之间的通信安全性。SSL 证书可以使用自签名或第三方机构签署的证书。

以下是一个 Java 代码示例，展示如何使用 SSL 安全协议进行认证：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/path/to/truststore");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "truststorePassword");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/path/to/keystore");
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "keystorePassword");

AdminClient adminClient = AdminClient.create(props);

SASL 验证机制

SASL 验证机制用于通过用户名和密码进行身份验证。Kafka 支持多种 SASL 机制，包括 PLAIN，SCRAM-SHA-256，SCRAM-SHA-512 等。

以下是一个 Java 代码示例，展示如何使用 SASL 机制进行认证：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"alice\" password=\"alice-secret\";");

AdminClient adminClient = AdminClient.create(props);

授权

ACL 权限控制

ACL 权限控制用于控制 Kafka 中各种资源的访问权限，例如 topic，consumer group 等。ACL 类型包括 Allow 和 Deny 两种，可以通过配置文件设置访问权限。

以下是一个 Java 代码示例，展示如何使用 ACL 进行授权：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"alice\" password=\"alice-secret\";");

AdminClient adminClient = AdminClient.create(props);

ResourcePattern resourcePattern = new ResourcePattern(ResourceType.TOPIC, "myTopic", PatternType.LITERAL);
AclBinding aclBinding = new AclBinding(resourcePattern, new AccessControlEntry("User:alice", "*", AclOperation.READ, AclPermissionType.ALLOW));

adminClient.createAcls(Collections.singleton(aclBinding));

RBAC 权限管理

RBAC 权限管理用于维护用户与角色之间的映射关系，并控制角色访问资源的权限。RBAC 可以使得权限的管理变得更具可扩展性和灵活性。

加密

数据传输加密

数据传输加密可以防止通信期间的数据泄漏，保障 Kafka 服务器和客户端之间的通信机密性。Kafka 支持多种传输协议加密，包括 SSL 和 SASL_SSL 等。

以下是一个 Java 代码示例，展示如何使用 SSL 进行传输加密：

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/path/to/truststore");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "truststorePassword");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/path/to/keystore");
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "keystorePassword");

AdminClient adminClient = AdminClient.create(props);

数据存储加密

数据存储加密可以防止在磁盘上存储的数据被非法访问。Kafka 可以使用加密文件系统等技术来保障数据存储加密的需求。

需要注意的是，对于已经存在的数据，在加密方式改变后需要进行转换才能继续使用。

安全性配置实践

通用实践

在进行 Kafka 安全配置时，需要遵循以下通用实践。

安全相关配置集中管理

在进行安全配置时，需要将所有安全相关的配置集中管理，包括认证、授权和加密等方面的配置。这样可以方便统一管理，降低出错率。

支持动态安全配置更新

一旦 Kafka 集群已经在生产环境中运行，即使安全配置出现问题，也不希望因此中断服务而影响正常业务运转。因此，在进行安全配置时，需要确保支持动态安全配置更新，以便在服务运行时及时更新配置。

数据与应用分离

为了进一步提高数据的安全性，建议将数据和应用程序分开部署，以避免数据泄露。这也意味着需要对 Kafka 集群进行适当的网络隔离，以减少攻击面。

认证配置实践

启用 SSL 加密

启用 SSL 加密可以对 Kafka 与客户端之间的数据进行安全传输，而不会被黑客监听截取。可以使用官方的 SSL 工具生成证书以进行配置。

进行双向身份验证

启用双向身份验证可以确保只有经过授权的客户端才能与 Kafka 集群通信，进一步提高数据安全性。在配置时需要对服务端和客户端都生成证书，并将其互相交换。

使用 SASL/Kerberos 进行身份认证

SASL 是一种安全认证协议，Kafka 支持使用 SASL/Kerberos 进行用户身份认证。通过 Kerberos 的验证机制，可以实现用户仅在通过 Kerberos 认证后才能访问 Kafka 集群。

授权配置实践

授权是 Kafka 安全性配置的又一个重要方面。

授权粒度优化

在进行 Kafka 授权配置时，应该尽可能精细地配置权限，避免赋予不必要的权限。例如，可以将不同 Topic 的权限分配给不同的用户或用户组，降低攻击者攻击的风险。

定期审计权限设置

在进行安全配置后，应该对授权权限进行定期审计。这样可以及时发现潜在的安全隐患，并对授权权限进行适当调整。同时，定期审计也可以帮助提高 Kafka 膜的使用效率。

加密配置实践

加密是 Kafka 安全性配置的最后一个方面。

启用 TLS/SSL 加密

在进行 Kafka 加密配置时，建议启用 TLS/SSL 加密，以保护 Kafka 集群与客户端之间的数据传输。同样，需要使用官方的 SSL 工具生成证书并进行配置。

启用数据加密

在进行 Kafka 加密配置时，还需启用数据加密，以便在数据库或文件系统中存储的数据也能得到保护。

加密算法的选择

在进行 Kafka 加密配置时，需要根据实际情况选择合适的加密算法，以确保安全性和性能。在选择加密算法时，需要权衡安全性和性能等因素，并结合实际情况进行灵活配置。更多关于消息中间件 Kafka 系列的学习文章，请参阅：消息中间件 Kafka，本系列持续更新中。

实践案例

金融领域数据应用实践

在金融领域，Kafka 被广泛应用于数据传输和处理。但是，由于金融领域对数据安全的要求非常高，因此必须采取一些措施来确保 Kafka 的安全性。

SSL/TLS 加密通信

使用 SSL/TLS 加密通信，可以确保 Kafka 集群与客户端之间的数据传输安全。建议使用证书进行身份验证，确保只有受信任的客户端才能访问 Kafka 集群。

Properties props = new Properties();
props.put("bootstrap.servers", "server1:9092,server2:9092");
props.put("security.protocol", "SSL");
props.put("ssl.truststore.location", "/path/to/truststore");
props.put("ssl.truststore.password", "truststorePassword");
props.put("ssl.keystore.type", "JKS");
props.put("ssl.keystore.location", "/path/to/keystore");
props.put("ssl.keystore.password", "keystorePassword");

集群间认证

在 Kafka 集群之间进行通信时，建议使用 SASL/PLAIN 或 SASL/SCRAM 认证机制。这可以确保只有受信任的节点才能加入集群，并防止攻击者对集群进行 DoS 攻击。

Properties props = new Properties();
props.put("bootstrap.servers", "server1:9092,server2: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='admin-secret';");

基于 ACL 的授权机制

使用基于 ACL（Access Control List）的授权机制，可以限制用户对 Kafka 主题和分区的访问权限。例如，只允许特定用户或用户组发布消息或消费消息。

AdminClient adminClient = KafkaAdminClient.create(props);
List<AclBindingFilter> filters = new ArrayList<>();
filters.add(AclBindingFilter.forResource(new ResourcePattern(ResourceType.TOPIC, "myTopic", PatternType.LITERAL))
            .withPermission(PermissionType.WRITE)
            .withPrincipal("User:alice"));
DescribeAclsResult aclResult = adminClient.describeAcls(filters);
Set<AclBinding> aclBindings = aclResult.values().get();

大规模日志采集系统安全配置最佳实践

大规模日志采集系统通常将日志发送到 Kafka 集群进行存储和分析。然而，由于包含敏感信息的日志非常容易成为攻击者的目标，因此必须采取措施来确保 Kafka 的安全性。以下是一些最佳实践：

拦截器

使用拦截器对发送到 Kafka 的消息进行预处理可以帮助识别并过滤掉潜在的攻击数据。例如，可以添加一个拦截器来检查每个消息是否包含 SQL 注入等常见的攻击。

public class SecurityInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // Perform security checks here
        if (isAttack(record.value())) {
            return null;
        }
        return record;
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "server1:9092,server2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SecurityInterceptor.class.getName());

数据脱敏

当日志中包含敏感信息时，建议在将日志发送到 Kafka 之前进行数据脱敏。这可以确保敏感信息不会被泄漏。

String message = "User:alice made a purchase of $100 with credit card 4111-1111-1111-1111";
String sanitizedMessage = message.replaceAll("\\b(\\d{4}-){3}\\d{4}\\b", "****-****-****-****");

防御 DoS 攻击

为了防止 Kafka 集群受到 DoS 攻击，可以使用另一种拦截器来限制发送到 Kafka 的消息数量。如果达到了预定义的阈值，拦截器将拒绝后续的消息。

public class ThrottleInterceptor implements ProducerInterceptor<String, String> {
    private final int MAX_MESSAGE_COUNT = 100;
    private int messageCount = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        if (messageCount >= MAX_MESSAGE_COUNT) {
            throw new LimitExceededException("Maximum message count exceeded");
        }
        messageCount++;
        return record;
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "server1:9092,server2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ThrottleInterceptor.class.getName());

云原生环境下 Kafka 安全性配置实践

在云原生环境下使用 Kafka，可以采用以下最佳实践来确保安全性：

访问控制

使用云提供商的访问控制功能，可以限制 Kafka 集群的访问权限。例如，可以设置只有特定 IP 地址或者虚拟机才能访问 Kafka 集群。

日志审计

启用日志审计记录 Kafka 集群上的所有操作可以帮助发现和防止潜在的攻击。

安全组件

安全性配置的漏洞与治理

常见漏洞类型及风险评估

Kafka 的安全性配置问题可能导致以下常见漏洞类型：

未经授权访问：攻击者未经授权即可从 Kafka 集群中读取/写入数据。
数据泄露：已授权用户可访问到其无权访问的数据。
服务停机：攻击者通过 DoS/DDoS 等方式导致 Kafka 集群或其相关服务停机。
加密不足：如果消息在传输过程中未加密，则可能会被拦截和窃听。

风险评估应考虑能力、动机和资源。攻击者的技能水平、目标数据的价值以及攻击者可用的资源都是要考虑的因素。

安全漏洞的安全补丁治理

为了防止 Kafka 的安全配置漏洞被攻击者利用，Kafka 社区会发布安全补丁程序，修复发现的漏洞，开发人员应该关注这些更新，在第一时间安装最新版本的程序和组件。

此外，以下措施也可以帮助保护 Kafka：

只有授权用户才能访问 Kafka 集群。
限制 Kafka 的网络暴露范围，特别是暴露在公共互联网上的情况。
强制要求密码策略和访问控制。
启用 SSL/TLS 协议加密以保护网络传输数据。
启用专业的安全策略审核程序，以便快速检测和阻止可疑活动。

下面是 Java 代码演示如何使用 Kafka 安全配置：

// 设置 Kafka 管理员安全配置选项
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.kerberos.service.name", "kafka");

// 初始化 Kafka 管理员
AdminClient adminClient = AdminClient.create(props);

安全事件的应急处理流程

如果发生安全事件，以下流程可帮助快速响应和解决问题：

收集证据：尽可能地记录所有有关事件的信息。
停机和隔离：隔离目标集群或服务器，以确保攻击者无法再继续进一步的攻击。
通知有关人员：通知与事件相关的人员或组织部门，并要求他们提供支持。
分析根本原因：深入分析安全事件的根本原因，并尽快采取措施，以防类似事件再次发生。
恢复服务：根据紧急情况恢复受影响的服务。
监控和审计：重新评估安全策略并设置更为严格的配置，同时对当前系统和网络活动进行日志跟踪和实时审计。

除此之外通过定期漏扫、脆弱性评估等方式加强 Kafka 的安全性，也是必要的。

参考文章：https://baixinan.blog.csdn.net/article/ details/130935326

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2023-08-15，如有侵权请联系 cloudcommunity@tencent.com 删除

安全

本文分享自民工哥技术之路微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度