1. Toy模板网首页
  2. > Toy博客

实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权

10月前 作者：vhicool 分类：Toy博客 阅读(37) 违法举报

这篇具有很好参考价值的文章主要介绍了实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、方案

1. 不接入外部存储，基于kafka原生ACL认证

环境： kafka-2.2.1、zookeeper-3.6.3

​ kafka给我们提供了SASL/SCRAM模式，将SASL、ACL规则信息存储到zookeeper中，并且通过KafkaClientAdmin Api，新增、编辑、删除规则，其特性如下

  • 应用发送、消费实现动态身份认证和授权

    基于kafka SASL/SCRAM模式，客户端会在建立连接进行SASL身份验证，在消费发送时，进行ACL鉴权

  • 安全认证代码无侵入性、兼容原生kafka api

    有两种配置方式，通过根据-Djava.security.auth.login.config指定jaas配置文件，并且配置producer.properties/consumer.properties 开启SASL配置，兼容原生kafka api

  • 账号级别区分：管理账号有最高权限、业务账号限定资源权限

    可通过配置kafka集群超级管理员，通过超级管理员账号登录是不进行ACL鉴权。因此超级管理员可以操作所有的资源权限

  • 安全认证支持动态开启、关闭

    kafka提供SASL和ACL管理的api，通过这些api可以新增、修改、查询、删除SASL和ACL策略。通过配置策略，可以实现安全策略开启、关闭

1.1 客户端与kafka集群认证、授权

实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权

1.2 安全认证、授权步骤如下

  1. 在集群初始化时，配置开启SASL、ACL，包括

    • broker与broker的SASL认证配置
    • broker与zookeeper的SASL认证配置

  2. 添加SASL、ACL规则

    • 管理员账号：添加SASL管理员账号，管理员账号新增完成后，broker与broker的认证才会成功。
    • 应用账号：新增主题、消费组时，通过KafkaAdminClient Api新增以alice为用户名，应用scret-1为密码的SASL认证信息。并且给这个用户授予消息写入权限

  3. 客户端配置SASL配置，并发送消息

    在properties中配置SASL账号密码。或者指定jaas配置文件

  4. 服务端认证、鉴权

    在客户端与服务端建立连接时，对客户端alice进行身份认证（SASL），认证通过后，在发送消息时检查资源（主题）是否授权（ACL）给改用户alice

zookeeper与broker认证、鉴权流程（SASL/SCRAM + ACL）
实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权

二、实战：kafka SASL+ACL配置步骤

第一步：集群配置

  • Zookeeer SASL配置

    修改zoo.cfg配置，开启SASL

    ...
sessionRequireClientSASLAuth=true
zookeeper.sasl.client=true
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider

    • sessionRequireClientSASLAuth:默认为false， 当设置为true 时，Zookeeper 服务器将只接受通过 SASL 向服务器进行身份验证来自客户端的连接和请求。使用 SASL 配置但身份验证失败（即使用无效凭据）将无法与服务器建立会话

    • zookeeper.sasl.client: 将值设置为false以禁用SASL身份验证。默认值为true

    • authProvider.1: 自定义ZooKeeper 添加新类型的SASL身份验证方案。指定的类需要实现AuthenticationProvider接口。zookeeper提供实现： org.apache.zookeeper.server.auth.SASLAuthenticationProvider

    新增JAAS配置文件：zoo_jaas.conf，用于定义认证身份信息。我们使用的LoginModule实现为： org.apache.zookeeper.server.auth.DigestLoginModule，它是基于明文用户、密码的形式进行身份认证

    Server {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       username="zookeeper"
       password="zookeepersecret”
       user_kafka="kafkasecret";
};

    指定jaas配置文件

    export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/Users/vhicool/zookeeper/conf/zoo_jaas.conf"

    启动zookeeper节点

    ./bin/zkServer.sh start-foreground

  • Kafka SASL配置

    创建证书

    • 创建证书

      在配置kafka broker之前，我们需要新增broker账号，这一步是必须的，需要在broker重启应用SASL+ACL配置前执行。这个username=admin，password=adminsecret，用了进行broker直接的SASL认证

      bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=adminsecret]' --entity-type users --entity-name admin

    • 服务端

      用户认证信息配置 jaas.conf，其中KafkaServer是服务端用来broker与broker直接的SASL认证，Client是kafka broker作为客户端与zookeeper服务端直接的认证。与SASL/PALIN模式不同的是KafkaServer不能指定kafka客户端用户认证账号，而是通过bin/kafka-configs.sh来配置。在Client中的LoginModule同样是明文用户、密码认证，和zookeeper的org.apache.zookeeper.server.auth.DigestLoginModule对应

      //broker与broker
KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="adminsecret";
};
//broker与zookeeper
Client{
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafkasecret";
};

      指定kafka环境变量，是SASL生效

      -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf

      server.properties配置，通过指定super.users指定超级用户，超级用户是不需要进行ACL鉴权，可作为管理员账号

      #SASL/SCRAM
listeners=SASL_PLAINTEXT://host.name:port
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-25
#ACL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin

第二部：新增认证、授权规则

  • kafka cli工具连接

    除了使用kafka自带cli，同时也可以使用kafka通过的api去维护：KafkaClientAdmin

    • 新增kafka client JAAS认证文件：kafka_client_jaas.conf

      //kafka client与zookeeper SASL
Client{
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafkasecret";
};
// kafka client与broker SASL
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="adminsecret";
};

    • 新增broker启动环境变量，指定JAAS文件

      export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_client_jaas.conf"

    • 通过命令工具新增主题

      验证：kafka client与zookeeper SASL

      bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic-1

    • 给用户alice SASL认证

      bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret]' --entity-type users --entity-name alice

    • 给主题ACL授权

      授予用户alice对主题topic-1生产者权限。涉及验证：kafka client与zookeeper SASL（此处我们并没有开启zookeeper ACL，因此不需要zookeeper ACL权限验证）

      bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice --producer --topic topic-1

    • 测试发送消息

      涉及验证： kafka client与broker SASL+ACL

      bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-1 --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=SCRAM-SHA-256

第三步：客户端配置

在用户alice配置SASL认证，并且授权主题topic-1发送权限后，客户端通过原生api发送消息

		Properties kafkaProperties = new Properties();
		kafkaProperties.put(org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.StringSerializer");
		kafkaProperties.put(org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
				"org.apache.kafka.common.serialization.ByteArraySerializer");
		kafkaProperties.put("bootstrap.servers", "localhost:9092");
		//SASL配置
		kafkaProperties.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required username=\"alice\" password=\"alice-secret\";");
		kafkaProperties.put("security.protocol","SASL_PLAINTEXT");
		kafkaProperties.put("sasl.mechanism","SCRAM-SHA-256");

		KafkaProducer<String, byte[]> producer = new KafkaProducer<>(kafkaProperties);
		ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>("topic-1", null, null, "test message", null);
		producer.send(record).get();

三、如果想自定义SASL、ACL存储形式

1. SASL存储形式

  • 本地文件存储

    PLAIN模式

    SASL/PLAIN是一种简单的用户名/密码认证机制，通常与TLS一起用于加密以实现安全认证。他的SASL认证是存储在本地JAAS文件，因此如果需要实现动态身份认证，需要拓展SASL认证接口

  • zookeeper存储

    SASL/SCRAM

    它解决了与执行用户名/密码认证的传统机制（如PLAIN和DIGEST-MD5）的安全问题。 RFC 5802中定义了该机制。Kafka支持SCRAM-SHA-256和SCRAM-SHA-512，它们可以与TLS一起使用以执行安全认证。用户名用作配置ACL等的认证主体。Kafka中的默认SCRAM实现在Zookeeper中存储SCRAM凭据，适用于Zookeepher位于专用网络上的Kafka安装

  • 外部认证服务器存储

    SASL/Kerberos

kafka默认SASL通过静态JAAS文件配

从Kafka 2.0版开始，可以通过使用配置选项sasl.server.callback.handler.class和sasl.client.callback.handler.class配置自己的回调处理程序来从外部源获取用户名和密码，从而避免在磁盘上存储明文密码。

  • sasl.server.callback.handler.class

    实现 AuthenticateCallbackHandler 接口的 SASL 服务器回调处理程序类的完全限定名称。服务器回调处理程序必须以侦听器前缀和小写的 SASL 机制名称为前缀。例如，listener.name.sasl_ssl.plain.sasl.server.callback.handler.class=com.example.CustomPlainCallbackHandler

  • sasl.client.callback.handler.class

    实现 AuthenticateCallbackHandler 接口的 SASL 客户端回调处理程序类的完全限定名称。

2. ACL存储形式

​ kafka默认ACL存储在zookeeper

实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权

自定义扩展外部存储

通过在客户端配置authorizer.class.name，可自定义实现存储位置。kafka默认是通过zookeeper存储ACL规则，实现类为：kafka.security.auth.SimpleAclAuthorizer。如果我们要扩展ACL存储，需要自定义认证类并实现kafka.security.auth.Authorizer接口，包括实现authorize、addAcls、removeAcls、getAcls等方法文章来源地址https://www.toymoban.com/news/detail-425782.html

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

到了这里，关于实战：kafka、zookeeper SASL+ACL实现动态权限认证、授权的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • Apache zookeeper kafka 开启SASL安全认证

    Apache zookeeper kafka 开启SASL安全认证

    背景：我之前安装的kafka没有开启安全鉴权，在没有任何凭证的情况下都可以访问kafka。搜了一圈资料，发现有关于sasl、acl相关的，准备试试。 Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发，支持多语言（如Java、Python、Go等）客户端，它可

    2024年03月14日
    浏览(45)
  • storm统计服务开启zookeeper、kafka 、Storm（sasl认证）

    storm统计服务开启zookeeper、kafka 、Storm（sasl认证）

    部署storm统计服务开启zookeeper、kafka 、Storm（sasl认证） 当前测试验证结果： 单独配置zookeeper 支持acl 设置用户和密码，在storm不修改代码情况下和kafka支持 当kafka 开启ACL时，storm 和ccod模块不清楚配置用户和密码。 使用python脚本连接kafka用户和密码是能成功发送消息。 当前部署

    2024年01月22日
    浏览(51)

  • Apache zookeeper kafka 开启SASL安全认证 —— 筑梦之路

      Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发，支持多语言（如Java、Python、Go等）客户端，它可以水平扩展和具有高吞吐量特性而被广泛使用，并与多类开源分布式处理系统进行集成使用。   Kafka作为一款开源的、轻量级的、分布式、可

    2024年02月11日
    浏览(31)
  • Kafka动态认证SASL/SCRAM配置+整合springboot配置

    Kafka动态认证SASL/SCRAM配置+整合springboot配置

    zookeeper启动命令： [root@master-yzjgxh2571705819-1651919082731-99-0727183 bin]# ./zkServer.sh start [root@master-yzjgxh2571705819-1651919082731-99-0727183 bin]# ./zkServer.sh stop kafka启动命令： /data/program/kafka2.12/bin/kafka-server-start.sh /data/program/kafka2.12/config/server.properties 1）创建broker建通信用户：admin(在使用sasl之

    2024年01月16日
    浏览(40)
  • centos 7 kafka2.6单机安装及动态认证SASL SCRAM配置

    centos 7 kafka2.6单机安装及动态认证SASL SCRAM配置

    目录 1.kfaka安装篇 1.1 安装jdk 1.2安装kafka 2.安全篇 2.1 kafka安全涉及3部份： 2.2 Kafka权限控制认证方式 2.3 SASL/SCRAM-SHA-256 配置实例 2.3.1 创建用户 2.3.2 创建 JAAS 文件及配置 3.测试 3.1 创建测试用户 3.2 配置JAAS 文件 3.2.1 生产者配置 3.2.2 消费者配置 3.3 消息收发测试 依赖环境：

    2024年02月07日
    浏览(36)
  • Kafka安全（以SASL+ACL为例）

    Kafka安全（以SASL+ACL为例）

    在Kafka的官方文档中介绍了如下几种安全方式，以满足用户不同场景对安全的需求。 ● SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0。主要是为 Kerberos 使用，如果当前已有 Kerberos 认证，只需要为集群中每个 Broker 和访问用户申请 Principle ，然后在 Kafka 配置文件中开启 Kerberos 的支持

    2024年02月03日
    浏览(33)

  • Kafka 中 SASL ACL SSL 到底分别代表什么意思

    auth: huangyichun date: 2023-5-11 看各类帖子都没能指出这些到底是什么意思，他们是冲突的，还是互相作用的，还是隔离的？本文讲解 kafka 中 SASL 、 ACL 、 SSL 他们分别的作用以及含义。 SASL 是用来认证 C/S 模式也就是服务器与客户端的一种认证机制，全称 Simple Authentication and Secu

    2024年02月15日
    浏览(44)
  • Kafka3.4 SASL/kerberos/ACL 证以及 SSL 加密连接

    Kafka3.4 SASL/kerberos/ACL 证以及 SSL 加密连接

    前面我们使用 kafka3.3.1 on zookeeper 的模式进行多网段监听的 kafka 集群，顺便搭建起 kafkaui 后发现一些问题，我们 kafka 集群没有连接认证，万一谁知道了我们的 kafka 连接地址，岂不是随随便便就能消费数据、清空数据、胡乱修改数据了吗？ 所以本章节进行认证连接的搭建，参

    2024年02月14日
    浏览(40)

  • 单机部署Kafka和开启SASL认证

    操作系统：linux apache-zookeeper-3.8.1-bin.tar.gz kafka_2.13-3.4.0.tgz 1.上传zookeeper 与kafka到 /opt 2、解压 tar -zxvf apache-zookeeper-3.8.1-bin.tar.gz tar -zxvf  kafka_2.13-3.4.0.tgz mv apache-zookeeper-3.8.1-bin zookeeper mv kafka_2.13-3.4.0 kafka 3、修改zookeeper配置文件并启动 修改zoo.cfg，增加以下命令 创建zk_server_j

    2024年01月22日
    浏览(41)

  • Zookeeper 节点权限控制ACL详解

    Zookeeper类似于unix文件系统，节点类比文件，客户端可以删除节点，创建节点，修改节点。Zookeeper可以使用ACL（access control list）访问控制列表来对节点的权限进行控制。 acl权限控制使用：scheme id permission来标志，主要涵盖三个方面： 权限模式（scheme）：授权的策略。 权限对象

    2024年02月14日
    浏览(40)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包