1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（一）认证

Written by Posted on Less than 0 min read

kafka安全涉及3部份：传输加密，用户认证与授权，ZK开启ACL（Zookeeper存储了kafka的元数据以及用户信息，默认不开启acl所有用户可改，内网环境机器不对外开放可考虑使用默认不开启ZK ACL）。

词汇说明：

认证，即用户登陆。

授权，即管理用户可见的资源。

ACL，Access Control List 访问控制列表

SASL认证与Kerberos认证：SASL资料很多, java的见这里，Kerberos的资料，点这里。

Kafka权限控制指引

支持的认证方式

Kafka支持的认证类别有kerberos（和hadoop一样，大多数公司应该没用kerberos）、ldap（在传统企业中比较普遍）与rbac（这两个是要企业license，基于Confluence的平台组件MDS，自建集群如果没用kafka connect之类的组件，没用conflunece的cli，也用不了）、sasl/plain（用户信息用文件进行管理，修改需重启Kafka，生产大概率不会接受并使用）和sasl/scram（用户信息用api或命令行进行管理，存储在zk上，不需重启）。考虑用哪一种取决于公司自身的情况，这里主要是用sasl/scram。

接入现有认证系统

如果对zk不放心或想对接已有的认证系统，接入kafka权限管控，可参考这个kip-86自定义认证方式，文中有场景列表及sample code。

开启认证主要的关注点

  1. 使用者的需求，生产者和消费者都有哪些，场景如何，需求怎样，支持以后需要先验证，对使用方提供技术上的指引文档等。
  2. 开启方式，是重启集群还是迁移集群。开启认证不是平滑的，需要短暂中断业务流程。
  3. 使用的认证方式：基于现有的环境及实际需要。
  4. 运维成本：考虑运维、监控的方式及成本（监控需要admin账号）
  5. kafka版本是否需要升级：公司现有的版本是0.10.1，1.1，2.4等，需要升级吗？看使用哪种认证方式,rbac需要2.4+，sasl/scram只要0.9+，本文描述的基于2.4.1版本。

本文主要描述用户认证：

参考：

一、zookeeper sasl开启（可选，内网环境，用户无机器访问权限时）

  • vim zookeeper/conf/zoo.cfg
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
  • vim zookeeper/conf/zookeeper_jaas.conf Server是ZK SERVER之间， Client是zkclient与zk server之间
Client {
  org.apache.zookeeper.server.auth.DigestLoginModule required
    user_admin="zkPassword";
};
Server {
  org.apache.zookeeper.server.auth.DigestLoginModule required
    user_admin="zkPassword";
};
  • vim zookeeper/bin/zkEnv.sh zk启动引入jaas配置
export SERVER_JVMFLAGS="-Djava.security.auth.login.config=${ZOOBINDIR}/../conf/zookeeper_jaas.conf $SERVER_JVMFLAGS"
  • vim zookeeper/conf/adminclient_jaas.conf
Client {
  org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="zkPassword";
};
  • vim zookeeper/bin/zkCli.sh zkCli命令默认添加jaas
"$JAVA" "-Djava.security.auth.login.config=${ZOOBIN}/../conf/adminclient_jaas.conf" ${后面接上原来的参数}
  • 重启zk bin/zkServer.sh restart
  • 验证 bin/zkCli.sh 是否成功，调整密码验证

二、配置kafka sasl

  • vim kafka/config/server.properties
listeners=SASL_PLAINTEXT://:9093
zookeeper.set.acl=true
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=false
# 参数 allow.everyone.if.no.acl.found
# 设置为 true，ACL 机制改为黑名单机制，只有黑名单中的用户无法访问
super.users=User:admin
#不能用，zk开启sasl后，jaas需包含KafkaServer内容，故这里用不了，除非zk不开启sasl
#listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required
#   username="admin"
#   password="kXQ[5/BIKJwAhYU";
  • vim kafka/config/kafka_server_jaas.conf KafkaServer是broker之间验证， Client是与ZK验证
KafkaServer {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="admin"
   password="kafkaPassword";
};
Client {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="zkPassword";
};
  • vim kafka/bin/kafka-server-start.sh server启动添加jaas
export KAFKA_OPTS="-Djava.security.auth.login.config=$base_dir/../config/kafka_server_jaas.conf"

三、新增scram 用户

密码与KafkaServer保持一致，否则会报sasl验证失败

bin/kafka-configs --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=kafkaPassword],SCRAM-SHA-512=[password=kafkaPassword]' --entity-type users --entity-name admin

四、start server

bin/start-kafka-server.sh

[2022-02-16 18:45:28,713] INFO [ZooKeeperClient ACL authorizer] Initializing a new session to 192.168.91.226:2181. (kafka.zookeeper.ZooKeeperClient)
[2022-02-16 18:45:28,713] INFO Initiating client connection, connectString=192.168.91.226:2181 sessionTimeout=6000 watcher=kafka.zookeeper.ZooKeeperClient$ZooKeeperClientWatcher$@290b1b2e (org.apache.zookeeper.ZooKeeper)
[2022-02-16 18:45:28,713] INFO jute.maxbuffer value is 4194304 Bytes (org.apache.zookeeper.ClientCnxnSocket)
[2022-02-16 18:45:28,714] INFO zookeeper.request.timeout value is 0. feature enabled= (org.apache.zookeeper.ClientCnxn)
[2022-02-16 18:45:28,714] INFO [ZooKeeperClient ACL authorizer] Waiting until connected. (kafka.zookeeper.ZooKeeperClient)
[2022-02-16 18:45:28,715] INFO Client successfully logged in. (org.apache.zookeeper.Login)
[2022-02-16 18:45:28,715] INFO Client will use DIGEST-MD5 as SASL mechanism. (org.apache.zookeeper.client.ZooKeeperSaslClient)
[2022-02-16 18:45:28,715] INFO Opening socket connection to server ATHADOOPDQ06/192.168.91.226:2181. Will attempt to SASL-authenticate using Login Context section 'Client' (org.apache.zookeeper.ClientCnxn)
[2022-02-16 18:45:28,716] INFO Socket connection established, initiating session, client: /192.168.91.226:31564, server: ATHADOOPDQ06/192.168.91.226:2181 (org.apache.zookeeper.ClientCnxn)
[2022-02-16 18:45:28,719] INFO Session establishment complete on server ATHADOOPDQ06/192.168.91.226:2181, sessionid = 0x102165975e5000d, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2022-02-16 18:45:28,720] INFO [ZooKeeperClient ACL authorizer] Connected. (kafka.zookeeper.ZooKeeperClient)
[2022-02-16 18:45:28,806] INFO [/kafka-acl-changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2022-02-16 18:45:28,806] INFO [/kafka-acl-extended-changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2022-02-16 18:45:28,851] INFO [ExpirationReaper-226-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2022-02-16 18:45:28,868] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2022-02-16 18:45:28,870] INFO Processing notification(s) to /config/changes (kafka.common.ZkNodeChangeNotificationListener)
[2022-02-16 18:45:28,877] INFO Processing override for entityPath: users/admin with config: Map(SCRAM-SHA-256 -> [hidden], SCRAM-SHA-512 -> [hidden]) (kafka.server.DynamicConfigManager)
[2022-02-16 18:45:28,882] INFO Removing PRODUCE quota for user admin (kafka.server.ClientQuotaManager)
[2022-02-16 18:45:28,886] INFO Removing FETCH quota for user admin (kafka.server.ClientQuotaManager)
[2022-02-16 18:45:28,886] INFO Removing PRODUCE quota for user admin (kafka.server.ClientQuotaManager)
[2022-02-16 18:45:28,886] INFO Removing REQUEST quota for user admin (kafka.server.ClientRequestQuotaManager)
[2022-02-16 18:45:28,886] INFO Removing FETCH quota for user admin (kafka.server.ClientQuotaManager)
[2022-02-16 18:45:28,886] INFO Removing REQUEST quota for user admin (kafka.server.ClientRequestQuotaManager)
[2022-02-16 18:45:28,898] INFO [SocketServer brokerId=226] Started data-plane processors for 1 acceptors (kafka.network.SocketServer)
[2022-02-16 18:45:28,899] INFO Kafka version: 2.4.1 (org.apache.kafka.common.utils.AppInfoParser)
[2022-02-16 18:45:28,899] INFO Kafka commitId: c57222ae8cd7866b (org.apache.kafka.common.utils.AppInfoParser)
[2022-02-16 18:45:28,899] INFO Kafka startTimeMs: 1645008328899 (org.apache.kafka.common.utils.AppInfoParser)
[2022-02-16 18:45:28,901] INFO [KafkaServer id=226] started (kafka.server.KafkaServer)

具体ACL配置、ZK ACL、以及验证，见下一篇文章。

Related articles

  1. kafka

Another Kafka Monitor – Kowl

kafka监控工具众多，使用过 yahoo的Kafka Manager, 滴滴的kafka manager, kafka eagle, 也造过kafka offset monitor, 甚至有结合kibana与prometheus做的kafka消费监控面板。最近又接触了这个go写的界面清爽的kowl，详见github。 官方支持docker，参考Dockerfile整理一下本地开发环境搭建过程: 1. 后端构建 安装依赖 go mod download 构建 cd backend && go build -o ./bin/kowl ./cmd/api，生成名为kowl二进制可执行文件 2. 前端 安装依赖 npm install 构建 注释package.json中的build script中的ssl option，npm run build 3. 运行 目录结构 其中，kowl.yaml为doc中的config拷贝，调整相关配置 启动 ./kowl -config.filepath ./kowl.yaml {"level":"info","ts":"2022-02-16T16:08:24.297+0800","msg":"started Kowl","version":"","git_sha":"dev","built":""}{"level":"info","ts":"2022-02-16T16:08:24.299+0800","msg":"connecting to Kafka seed brokers, trying to fetch […]
Written by
  1. kafka

kafka schema

why schema? 引用Jay kreps的话，关于topic的schema。原文见《Why Avro for Kafka Data?》。 one thing you’ll need to do is pick a data format. The most important thing to do is be consistent across your usage。 json虽然表达能力强，但冗余，尤其对于kafka这类数据存储。 引入序列化协议及schema注册中心，可以优化数据存储、支持schema evolve，保证生产消费对同个字段的相同解析，简化etl处理。 但也有折衷，引入注册中心无于多了组件，如果registry宕机，kafka 不可用。所以很多公司不会使用json，而是使用自已约定的消息格式，除非像一些传统公司技术能力欠缺或者kafka消息体量很小还不考虑吞吐和性能问题，当然也有可能公司不差钱。 confluent官方提供了schema registry，用于schema的注册等级，生产消费都会从它那里登记或获取消息序列反序列化schema。 官方支持的schema类型有avro,json,protobuf，推荐avro，原因见上面的链接，摘要如下 We chose Avro as a schema representation language after evaluating all the common options—JSON, XML, Thrift, […]
Written by
  1. kafka

Kafka序列化反序列化解析

kafka有自己的rpc协议，即nio bytebuf中的数据格式，详见之前的kafka相关介绍的文章。这里我们来看一下大家常用，有时又疑惑的序列化反序列化，对应rpc协议中的records，kafka叫Serdes，实际上也是字面上的意思serialize and deserialize。在程序中序列化就是toBinary，反序列化是fromBinary，存储和传输都是用二进制。 通过以上描述，我们知道序列化和反序列化的作用。我们来看kafka的实现。Serdes工厂类，和提供基本数据类型序列化。 一、先看基础的抽象 interface：Serializer 包括 configure，例如配置ChartSet字符集，serialize，序列化。 interface：Deserializer 同理。 再来看包装接口：Serde 指定了serializer和deserializer。 二、 实现类，以Serializer为例 提供了几大应用的实现，分别是common，connect，stream，也即对应了kafka一般场景、kafka抽数场景如debezium，kafka stream原生流计算。我们看common，主要是实现了基本的数据类型的序列化，举两个例子，Long和String。 从上我们知道实现Serialize类，主要做了toBinary动作。 三、包装类，工厂方法 这里看WrapperSerde 为静态内部类，有一批子类： 看其中一个常用的StringSerde 引入String序列化类和反序列化类。
Written by
  1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（二）ACL

由上一篇博文，讲解了SASL认证的安装过程。本文主要围绕ACL设置进行说明。 参考： kafka authentorization：官方 一、开启ZK ACL（可选，内网环境，用户无机器访问权限时） 给kafka meta都加上zk的acl，默认是world，即任何人可访问。 zk acl详见：https://www.cnblogs.com/dalianpai/p/12748144.html 执行已有kafka meta的迁移 vim kafka/bin/zookeeper-security-migration.sh 执行迁移 到zk上查看生效 二、设置kafka acl 上篇文章已经在server.properties中添加了acl的配置。这里执行命令设置acl。 kafka的acl对象，即资源类型分为 cluster, topic, consumer group, transation id, token。实际应用中，主要以topic及consumer group为主进行运维管控。即控制”哪个用户的哪个消费组可以访问哪个topic“。 常用acl操作 列出 test 这个 topic 的 ACL 列表 对用户 test 授权，以访问test 这个 topic。 先添加scram用户 test 因为添加了zk的acl，所以需指定ZK的acl jaas账号信息。若不指定ZK的jaas账号信息，会报如下错误： 加acl认证信息的方式：先调整bin/kafka-run-class.sh，用于zk acl验证。这个文件被大多数的kafka bin目录下的script所引用，如添加topic的脚本等，所以改它最方便，添加一行。 再执行一次添加即可，输出如下提示表示添加用户test完成。 添加consumer 授权方式1，先加topic权限： 输出： 再添加消费组权限： 添加consumer 授权方法2，参考文档。一步到位： […]
Written by
  1. kafka

Kafka SASL认证授权（四）认证源码解析

一、认证流程 在了解kafka网络模型的基础上，了解它的认证流程： ApiVersionsRequest->SaslHandshakeRequest->a series of SASL client and server tokens corresponding to the mechanism are sent->认证成功，继续处理后续的请求，否则关闭连接。 状态转换见SaslServerAuthenticator https://kafka.apache.org/protocol#sasl_handshake 具体逻辑chanel验证流程以sasl/plain为例： 分析这个包的内容： server factory创建sasl plain server单例。server负责按协议获取username 与 password，再最终调用plain server callback handler进行验证（handler的指定是在SaslChannelBuilder中根据具体配的协议类型进行指定）。 下面看一下关键类SaslChannelBuilder，创建sasl channel的实现： 初始化时，配置handler： 再看具体的channel，结合selector poll的过程KafkaChannel： java doc有说明具体的网络模型： 我们来看prepare方法： prepare触发的地方，调用频率在selector，事件通知触发： 也就是每次网络读事件且channle尚未验证成功都会触发认证，和官方文档描述一致。一旦验证通过就不再认证。所以开启认证对吞吐影响可以忽略不计。 校验不过，会直接抛异常捕获后，断开连接了。 二、总结 以上我们了解到kafka sasl的认证流程，现在，可以自己添加认证方式，例如加上自己公司平台的权限认证，注意尽量使用轻量级的方式即cache。 scram方式，除了以上流程，还会多一步，见ScramSaslServer。第一步获取credential，无效则报错，第二步计算校验。详见scram rfc定义：这里 kafka server这边的final message 截图： 自定义认证可以参考这篇：博文 三、token验证 kafka实现了基于token验证的sasl。详见这里：文档 流程是：admin client管理client（创建和刷新），存放在zk上，验证时会验token是否存在，以及用户信息，密码是否一致。token的生命周期内有效，过期无效。 弊端：在公司内一般不会频繁用到这种方式， […]
Written by
  1. kafka

Kafka SASL认证授权（五）ACL源码解析

一、ACL检查流程解析 一起看一下kafka server的启动与监听流程： Kafka -> KafkaServer -> SocketServer、KafkaRequestHandler 其中KafkaServer做相关的初始化，包括SocketServer 与 handler pool。 SocketServer的start up流程 看一下acceptor acceptor实际执行流程：监听网络事件，提交给processor处理 看一下processor，截取KafkaServer代码：这边的numIoThreads就是配置的具体处理线程数量。KafkaApis封装具体的各api的处理逻辑，handler pool则是工作线程池。 看看kafkaApis 接着看pool中每个线程的while循环，从request队列拿一个请求，调用apis处理，结果发送给对应channel。 request的入队，见SocketServer，这里也可以看到acl使用的principal和sasl认证是同一个。 之前我们看到，在selector做channel初始化时，已经做了sasl认证。那acl在哪里处理？KafkaApis。以offset commit为例 看一下，这边的authorize，各api共用的方法。构建resource pattern，即资源描述，以及对应的操作。然后使用authorizer实现类进行处理。 看一下鉴权实现类AclAuthorizer，初始化，主要是zk相关的监听，因为token,user,acl都存在zk。 遍历每个action 每个action的检查，区分超管，否则根据acl进行匹配。使用acl 缓存，结合zk的linstener更新缓存。不管是否开启acl，都会去取acl缓存列表。 二、总结 kafka以事件、线程池的方式，对接口请求做处理。 每个api请求，都会做相关的acl校验，不管是否开启acl，都会去取acl缓存列表。差别在于是否使用acl cache map对action进行匹配，匹配本身是用遍历的方式，action不多的话，对吞吐的影响可以忽略不计。 网上有一篇性能测试：见这里
Written by
  1. kafka

Kafka 开启SASL/SCRAM认证 及 ACL授权（三）验证

前面两篇文章讲解了如何配置认证。本文接着说明如何做client验证ACL是否生效，我们之前开启了无acl信息不允许访问的配置。涉及的client有以下几个场景：shell脚本、python脚本、java应用、flink流。 kafka shell script验证 核心逻辑是，连zk的，使用zk的jaas，连kafka broker的，使用kafka 的jaas配置。 新建topic, kafka-run-class.sh已经配置了zk acl 添加admin shell认证配置文件 vim kafka/config/adminclient-configs.conf 查看所有用户 对应zk目录在config/users下 查看topic 列表，若未指定jaas配置文件，则会hang不响应 列出topic test的acl列表 尝试不指定用户 消费test topic 报错 消费与生产配置，添加用户jaas vim config/consumer.properties vim config/producer.properties 验证消费与生产 python验证 admin用户，验证通过。其他用户需添加用户，并添加acl分别验证 如果认证方式不对，报错信息是kafka.errors.NoBrokersAvailable: NoBrokersAvailable。因为client先获取了kafka version，获取失败就报了。如果用户信息不对，结果是进程hang挂起。 java 如果用户信息不对，会hang挂起 flink scala code 输出 : 若用户信息出错，flink不断重试，报错
Written by
  1. kafka
  2. 工具

本地搭建源码阅读开发构建环境示例: kafka

大家对开源项目有兴趣、想成为committer，或者工作需要，会从github上获取最新的开源项目源码。本文做一个示例，怎样搭建本地的源码阅读、开发、构建环境。 首先，在github上找到项目的链接，clone到本地自己的目录中，这步略过。 下面一起看一个示例：使用idea导入kafka 2.4.1版本的源码。 一、idea打开项目后，切换到指定的分支 获取最新分支列表（如果卡很久，shell加一下自己的网络代理，翻墙一下）： git fetch –all –tags 获取2.4前缀分支列表: -n 加描述 -l <pattern> 加正则 从git tag创建分支，切换为指定的版本：git checkout tags/<tag> -b <branch> 二、下载依赖 gradle build项目 报错1：本地默认gradle版本过高，没有MavenDeployment这个插件。 为了解决gradle版本问题，项目使用gradle wrapper版本配置。调整idea使用项目的gradle，preference->Build,->Gradle->Use Gradle from，选择gradle/wrapper/gradle-wrapper.properties，重新reload。或者直接执行项目根目录的gradlew，会自动下载指定版本编译。 三、本地开发 熟悉代码内容，方式推荐两种： 1、结合文档看 2、结合jira看，熟悉kafka的代码模块、开发、提交pr的流程 kafka2.8之前需要zk，所以先把本地zk起起来，入口类：core模块的kafka.Kafka 运行前读一下kafka-server-start.sh，知道它的log配置是通过命令参数的方式： 所以，运行时带上参数，如下图： 此外，依然报错： 排查gradle脚本，compile阶段没有依赖slf4j实现库，而是在发布时从test compile拷贝这个依赖，所以可以加一下依赖到core项目的依赖配置中： Tips: 本地一般会有各版本的kafka，这边是开发调试用的，改一下zk地址，添加一下chroot zookeeper.connect=localhost:2181/kafka_dev 四、构建 使用命令行（README.md中有说明），或者直接在idea gradle窗口双击执行相关task。 查看输出的release包 五、延伸 kafka code指引：https://kafka.apache.org/coding-guide.html kafka bug jira：jira kafka contribute：https://kafka.apache.org/contributing […]
Written by
Next
No Comments
Leave a comment Cancel
Comments to: Kafka 开启SASL/SCRAM认证 及 ACL授权（一）认证

Your email address will not be published. Required fields are marked *