Kafka权限认证 | 如何实现Topic级权限动态认证

后端

2023-11-23 23:46:02

Kafka 的权限认证机制

简介

Kafka 作为分布式消息系统，提供了强大的权限认证体系，确保消息的安全性。Kafka 的权限认证体系包括身份认证（Authentication）和权限控制（Authorization）两个方面。

身份认证

身份认证负责验证客户端与服务器的连接身份，包括代理之间的连接、其他代理、工具与代理之间的连接。Kafka 使用 PLAINTEXT 协议进行身份认证，支持用户名和密码验证。

权限控制

权限控制负责消息级别的权限控制，具体包括数据读写操作（生产/消费/组）的权限。Kafka 使用 ACLAuthorizer 作为授权器，提供基于主题的权限控制功能。

Kafka 权限认证实践

配置 Kafka 权限认证

# 启用权限认证
security.inter.broker.protocol=PLAINTEXT
security.protocol=PLAINTEXT

# 设置认证用户列表
security.providers=org.apache.kafka.common.security.plain.PlaintextSaslServerProvider

# 设置授权规则
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:admin

创建授权用户

kafka-acls --create --authorizer-properties zookeeper.connect=localhost:2181 --resource-type topics --resource-name topic-1 --operation Read --principal User:user1

配置主题权限规则

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add --resource-type topics --resource-name topic-1 --operation Read --principal User:user1

Kafka 权限认证的动态配置

通过 API 动态配置

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AclBinding;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.security.auth.PrincipalBuilder;

public class DynamicAclUpdate {
    public static void main(String[] args) {
        AdminClient adminClient = AdminClient.create(kafkaConfig);

        KafkaPrincipal principal = new PrincipalBuilder("User").name("user1").build();

        AclBinding aclBinding = new AclBinding(
                new ResourcePattern(ResourcePattern.ResourceType.TOPIC, "topic-1"),
                principal,
                new AclOperation(AclOperation.READ));

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

        adminClient.close();
    }
}

通过命令行动态配置

kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add --resource-type topics --resource-name topic-1 --operation Read --principal User:user1