公司Kafka集群跑了两年多一直没加安全认证。直到有一天某个业务的Topic被不知道哪来的消费者把数据全部拉走我们才意识到问题有多严重——Kafka默认配置下只要网络能通到9092端口谁都能读、谁都能写明文数据裸奔在链路上。这篇文章我从零到一复盘了Kafka集群加装TLS加密传输、SASL/SCRAM认证、ACL授权控制的全过程包括证书生成、Broker配置、客户端适配和线上踩过的坑给准备给Kafka上锁的团队一个可直接参考的落地路径。1. 裸奔的Kafka集群为什么安全机制不是可选项1.1 Kafka默认配置下的安全隐患先说结论Kafka默认安装完成后几乎等于光着身子跑在网络上。Broker监听在0.0.0.0:9092没有认证、没有传输加密、没有访问控制。这个设计在早期内网环境问题不大但在现在的多云、混合云架构下完全不可接受。具体来说裸奔的Kafka主要有三个层面问题未认证Authentication缺失任何能访问到9092端口的人都能用任意client.id连接Broker然后对Topic做Produce、Consume、Describe操作。你根本不知道对面是谁。明文传输Encryption缺失Producer发送的消息体、Metadata信息全部以明文在网络上传输。只要有人在内网用tcpdump抓包消息内容一览无余。对于包含用户手机号、订单信息的Topic这等于直接泄露。无授权Authorization缺失就算你给Broker设置了认证默认的super.users之外所有用户对所有Topic都有读写权限因为allow.everyone.if.no.acl.found默认是true。也就是说认证通过等于万能钥匙。我们当时的线上事故就是第三种情况加第一种情况的叠加一个有网络访问权限的第三方服务用默认的消费者配置把我们某个核心业务Topic的数据全拉走了。事后排查发现对方甚至不知道自己连接的什么只是配置了一个通配符订阅数据就像水一样流过去了。1.2 安全三件套的定位门禁、保险箱、权限表很多刚接触Kafka安全的人会把认证和加密混为一谈实际上它们是三个独立维度互相配合才能形成完整防护安全维度解决什么问题生活类比Kafka对应的实现认证Authentication你是谁小区门禁先验明身份SASL/PLAIN、SASL/SCRAM、Kerberos、mTLS加密Encryption数据在路上会不会被偷看/篡改快递保密运输拆开也看不懂SSL/TLS加密传输授权Authorization你能做什么物业权限表只能进自己那栋楼Kafka ACLAccess Control Lists注意这三个维度可以单独开启也可以组合使用。比如你只开SASL认证但不开TLS加密那认证时的用户名密码走的还是明文抓包就能看到。所以生产环境的标准操作是SSL/TLS加密传输 SASL/SCRAM认证 ACL授权三件套一起上。1.3 什么时候必须做安全加固我的判断标准很简单满足以下任何一条就不要再拖Kafka集群能被公司外部网络访问或者处于共享网络环境中比如K8s集群内部多租户Topic中传输的数据涉及用户隐私、业务敏感信息、财务数据有多个团队/多个业务线共用一套Kafka集群需要区分权限公司有等保合规要求明文传输和未授权访问过不了审计如果只是本地开发环境或者完全物理隔离的内网短期内裸奔可以理解。但作为基础设施安全越早做越省事——一旦业务规模上来Topic几十上百个客户端五花八门再想加认证就要协调所有客户端升级成本成倍放大。2. 集群安全方案选型SASL认证机制怎么挑2.1 四种认证机制的对比与适用场景Kafka支持的认证机制有好几种选错了后面会很痛苦。我把主要的四种放在一起对比认证机制原理优点缺点适用场景SASL/PLAIN明文用户名密码在Channel层使用TLS加密保护配置简单客户端兼容性最好密码存储于ZK/Broker中无法动态更新必须配合TLS否则密码裸奔内网可信环境、客户端类型杂的团队SASL/SCRAM-SHA-256/512基于质询-响应机制密码不明文传输即使没有TLS也能防抓包强密码存储服务端不知道明文密码支持动态创建用户社区推荐配置比PLAIN略复杂性能略低于PLAIN生产环境首选尤其是多团队共用集群SASL/GSSAPIKerberos基于Kerberos票据认证企业级最强与AD域集成需要维护KDC运维成本高排错难度大大型企业、已有Kerberos基础设施SSL/mTLS客户端和Broker互相验证证书证书粒度控制无需账号密码与TLS天然一体证书管理复杂每个客户端要签发证书、处理轮转无授权扩展设备身份认证场景、IoT场景2.2 我的选型结论SCRAM TLS是默认最优解如果你是从零开始搭建我的建议是别犹豫直接上SASL/SCRAM-SHA-256 SSL/TLS组合理由有三点第一SCRAM不需要维护Kerberos那套独立的KDC用kafka-configs脚本就能在集群中动态创建、删除用户操作成本低。第二SCRAM的密码是以哈希加盐形式存储的Broker端不保留明文密码即使某个节点的配置被泄露也不会直接暴露所有账号的密码。第三它与TLS加密是正交关系——就算TLS证书出了问题SCRAM机制本身也能防止密码在链路上被抓包。当然如果你的公司已经有Kerberos基础设施且所有客户端都是Java系走GSSAPI也合理。但对绝大多数中小团队来说SCRAM足够安全运维复杂度又可控。2.3 端口规划9092还是9093安全机制开启后建议不要在原端口上原地加认证而是把加密和认证监听放到新的端口。常规做法是listenersPLAINTEXT://:9092仅用于内网日志采集、本地调试等不需要认证的场景advertised.listenersSSL://:9093用于需要安全传输和认证的生产流量sasl.enabled.mechanismsSCRAM-SHA-256端口分离的好处是老的客户端不需要立刻改造可以逐步迁移同时你可以在网络层做ACL只允许特定网段访问9093。我们迁移的时候就是这么做的先在9093上把新机制全部跑通再约各个业务方分批切换零停机完成。3. TLS加密传输从证书生成到Broker配置3.1 证书规划自己签CA还是用现有CATLS加密的第一步是解决证书问题。生产环境有一条铁律客户端永远不要直接信任Broker自签名的单个证书。正确做法是自己搭建一个内部CACertificate Authority用这个CA给每个Broker节点签发证书然后在客户端配置中信任这个CA。自建CA的好处是证书签发、吊销、轮转都是自己控制而且后续如果Broker数量增加只需要用同一套CA签新证书即可客户端不需要改任何配置。我们用的是openssl在企业内网建了一套离线CA签名机不联网签发流程靠人工走审批安全性和可追溯性都有保障。3.2 openssl生成CA与Broker证书的完整过程下面是我在3节点集群上实际执行过的操作。假设三个Broker主机名是kafka1.example.com、kafka2.example.com、kafka3.example.com分别绑定IP10.10.1.11、10.10.1.12、10.10.1.13。证书需要同时包含IP和域名因为有些客户端走IP连接有些走域名连接SAN不匹配会导致握手失败。第一步创建CA私钥和自签名根证书# 生成CA私钥密码注意保存好 openssl genrsa -aes256 -out ca-key.pem 4096 # 生成CA根证书有效期设置10年 openssl req -new -x509 -key ca-key.pem -days 3650 -out ca-cert.pem \ -subj /CCN/OMyCompany/OUSecurity/CNKafka-CA第二步为每个Broker生成私钥和证书签名请求CSR。这里必须配置SAN包含broker的域名和IP# 以kafka1为例 openssl genrsa -out kafka1-key.pem 2048 cat kafka1.cnf EOF [req] distinguished_name dn req_extensions v3_req prompt no [dn] C CN O MyCompany OU Kafka CN kafka1.example.com [v3_req] subjectAltName alt_names [alt_names] DNS.1 kafka1.example.com DNS.2 localhost IP.1 10.10.1.11 IP.2 127.0.0.1 EOF openssl req -new -key kafka1-key.pem -out kafka1.csr -config kafka1.cnf第三步用CA签名签发证书有效期设为730天2年方便定期轮转openssl x509 -req -in kafka1.csr \ -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial \ -out kafka1-cert.pem -days 730 \ -extensions v3_ca -extfile kafka1.cnf验证证书和SAN配置正确openssl x509 -in kafka1-cert.pem -noout -text | grep -A3 Subject Alternative Name输出里应当能看到DNS.1、IP.1对应的内容这个必须核验因为我在实际项目里踩过坑签出来的证书SAN漏了IP导致生产客户端全部握手失败。3.3 Broker端TLS配置把ca-cert.pem、kafka1-cert.pem、kafka1-key.pem拷贝到每个Broker节点的/etc/kafka/ssl/目录然后修改server.properties。# 监听SSL端口9093 listenersPLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093 advertised.listenersPLAINTEXT://kafka1.example.com:9092,SSL://kafka1.example.com:9093 inter.broker.listener.namePLAINTEXT # SSL配置 ssl.keystore.typeJKS ssl.keystore.location/etc/kafka/ssl/kafka1.keystore.jks ssl.keystore.passwordchange-me ssl.key.passwordchange-me ssl.truststore.location/etc/kafka/ssl/kafka1.truststore.jks ssl.truststore.passwordchange-me # 可选要求客户端连接也通过TLS验证双向TLS # ssl.client.authrequired注意这里JKS只是一种格式选择。如果你倾向于PEM格式用以下配置ssl.keystore.typePEM ssl.keystore.certificate.chain/etc/kafka/ssl/kafka1-cert.pem ssl.keystore.key/etc/kafka/ssl/kafka1-key.pem ssl.truststore.certificates/etc/kafka/ssl/ca-cert.pem我建议用JKS格式因为大部分Java客户端的信任库配置都以JKS为例兼容性最省心。转换命令keytool -importcert -alias ca -file ca-cert.pem -keystore kafka1.truststore.jks -storepass change-me -noprompt openssl pkcs12 -export -in kafka1-cert.pem -inkey kafka1-key.pem -certfile ca-cert.pem -out kafka1.p12 -password pass:change-me keytool -importkeystore -srckeystore kafka1.p12 -srcstoretype PKCS12 -srcstorepass change-me -destkeystore kafka1.keystore.jks -deststoretype JKS -deststorepass change-me3.4 TLS验证用openssl检查加密链路配置完Broker后重启Kafka注意滚动重启然后用openssl验证SSL端口是否正常协商openssl s_client -connect kafka1.example.com:9093 -CAfile ca-cert.pem连接成功并且证书链验证通过说明TLS监听没问题。接下来用Kafka自带的客户端工具测试# 不带认证只测TLS链路 bin/kafka-console-consumer.sh \ --bootstrap-server kafka1.example.com:9093 \ --topic test-topic \ --consumer.config ssl-client.propertiesssl-client.properties至少要包含security.protocolSSL ssl.truststore.location/etc/kafka/client.truststore.jks ssl.truststore.passwordchange-me这里有个容易忽略的点如果Broker配置了ssl.client.authrequired双向认证客户端还必须提供自己的证书如果没配客户端只验证Broker证书即可。生产环境建议先做单向TLS跑通之后再根据场景开启双向。4. SCRAM认证实战让谁都能连变成只有指定账号能连4.1 在Broker中初始化SCRAM用户凭据TLS解决了数据链路安全接下来解决谁能连的问题。我们采用的是SASL/SCRAM-SHA-256。首先需要在Broker上开启SCRAM机制并配置相应的JAAS文件。Kafka 2.x之后推荐使用kafka-configs.sh创建用户而不是直接改JAAS文件这样用户凭据存储在Zookeeper中Broker动态读取。启用SCRAM之前先修改server.propertiessasl.enabled.mechanismsSCRAM-SHA-256,SCRAM-SHA-512 ssl.endpoint.identification.algorithm第二行ssl.endpoint.inidentification.algorithm留空表示关闭主机名校验避免客户端用IP连接时证书域名不匹配报错。但如果你确认所有客户端都用域名连接可以保留默认的HTTPS校验逻辑。然后创建第一个管理员账号bin/kafka-configs.sh --zookeeper localhost:2181 \ --alter --add-config SCRAM-SHA-256[passwordadmin-secret] \ --entity-type users --entity-name admin bin/kafka-configs.sh --zookeeper localhost:2181 \ --alter --add-config SCRAM-SHA-512[passwordadmin-secret] \ --entity-type users --entity-name admin注意如果是KRaft模式Kafka 3.x移除了Zookeeper要改用--bootstrap-server配合--command-configbin/kafka-configs.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties \ --alter --add-config SCRAM-SHA-256[passwordadmin-secret] \ --entity-type users --entity-name admin然后配置JAAS文件启用Broker端的SCRAM验证。创建一个kafka_broker_jaas.confKafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required usernameadmin passwordadmin-secret; };不能手动把密码写在这里也没关系ScramLoginModule会自动从Zookeeper拉取。但为了方便生产环境通常把管理员账号写在JAAS里其余普通用户用kafka-configs.sh管理。4.2 同时启用TLS和SASL组合协议配置当你把TLS和SASL同时打开客户端的security.protocol要改成SASL_SSLBroker的listener也要调整sasl.enabled.mechanismsSCRAM-SHA-256 sasl.mechanism.inter.broker.protocolSCRAM-SHA-256 listener.name.sasl_ssl.scram-sha-256.sasl.jaas.configorg.apache.kafka.common.security.scram.ScramLoginModule required usernameadmin passwordadmin-secret;注意Kafka 2.x开始不推荐在静态JAAS文件里配密码给Broker间通信更好的做法是给broker单独建一个专用于broker间通信的账号。比如bin/kafka-configs.sh --zookeeper localhost:2181 \ --alter --add-config SCRAM-SHA-256[passwordbroker-secret] \ --entity-type users --entity-name broker然后listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config中usernamebroker passwordbroker-secret。这样万一某个Broker的JAAS配置泄露也只影响Broker间通信不影响外部客户端账号安全。4.3 客户端连接带SASL的Java配置客户端侧需要增加SASL配置用sasl.jaas.config直接在Java属性里设置避免依赖外部JAAS文件Properties props new Properties(); props.put(bootstrap.servers, kafka1.example.com:9093,kafka2.example.com:9093,kafka3.example.com:9093); props.put(security.protocol, SASL_SSL); props.put(sasl.mechanism, SCRAM-SHA-256); props.put(sasl.jaas.config, org.apache.kafka.common.security.scram.ScramLoginModule required username\app-user\ password\app-secret\;); props.put(ssl.truststore.location, /etc/kafka/security/client.truststore.jks); props.put(ssl.truststore.password, change-me);这里有一个关键点sasl.jaas.config这个配置项是Kafka 2.0之后才引入的它可以直接写进客户端Properties而不用在每台机器上配java.security.auth.login.config。对团队来说把用户名密码直接放在配置中心或者环境变量中管理方式更灵活。用命令行工具测试连接cat client-sasl.properties EOF security.protocolSASL_SSL sasl.mechanismSCRAM-SHA-256 sasl.jaas.configorg.apache.kafka.common.security.scram.ScramLoginModule required usernameapp-user passwordapp-secret; ssl.truststore.location/etc/kafka/security/client.truststore.jks ssl.truststore.passwordchange-me EOF bin/kafka-console-producer.sh \ --broker-list kafka1.example.com:9093 \ --producer.config client-sasl.properties \ --topic test-topic能正常输入消息并发送成功说明SASL认证链路通了。4.4 认证过程中的常见坑SCRAM配置过程中我踩过几个有代表性的坑列出来供参考坑一Authenticate failed 但日志不直观。这个报错的根因通常是用户名或密码错误但在日志中只会显示远端异常断开。解决办法是先确认kafka-configs.sh创建用户时用的密码是否和客户端一致再确认客户端确实走了SASL_SSL而不是纯SSL。坑二Broker间通信认证失败导致副本分区异常。加了认证之后Broker之间的连接也需要认证。很多人在客户端测试通过后忽略了这个结果发现ISR列表一直在缩。解决方法是专门给inter.broker账号配置并确保sasl.mechanism.inter.broker.protocol与listener中配置的mechanism一致。坑三连接超时而不是认证失败。当Broker配置了SASL而客户端仍然使用PLAINTEXT协议时连接通常表现为超时因为Broker在等待的握手数据客户端根本没发。排查方式是用抓包或者看Broker端日志中的异常记录确认port对应的协议。5. 授权控制ACL让认证通过的账号也有权限边界5.1 认证通过不等于能操作一切很多团队加了认证之后就以为完事大吉其实认证只解决了你是谁的问题。默认情况下Kafka集群的allow.everyone.if.no.acl.found是true意思是如果某个资源没有设置ACL所有已认证用户都有全部权限。这意味着只要你创建了一个账号就能读所有Topic写所有Topic。所以生产环境必须做两件事一是把allow.everyone.if.no.acl.found设为false让没有明确ACL授权的资源默认拒绝二是配置super.users也就是超级管理员账号绕过ACL控制。在server.properties中设置allow.everyone.if.no.acl.foundfalse super.usersUser:admin注意如果你开了SCRAM认证这里的User:admin一定要和你创建的SCRAM用户名一致否则管理员自己都会被拒之门外。5.2 用kafka-acls配置最小权限账号以我的生产环境为例假设有># 给data-pipeline授予ods_events的写入权限 bin/kafka-acls.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties \ --add --allow-principal User:data-pipeline \ --operation Write --topic ods_events # 给data-pipeline授予raw_logs的读取权限 bin/kafka-acls.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties \ --add --allow-principal User:data-pipeline \ --operation Read --topic raw_logs --group pipeline-group # 查看已有ACL bin/kafka-acls.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties --list这里再提一句如果开启了allow.everyone.if.no.acl.foundfalse别忘了给管理员账号也配置Cluster级权限否则连创建Topic都做不了bin/kafka-acls.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties \ --add --allow-principal User:admin \ --operation All --topic * --group * bin/kafka-acls.sh --bootstrap-server kafka1.example.com:9093 \ --command-config admin.properties \ --add --allow-principal User:admin \ --operation All --cluster kafka-cluster5.3 授权失败的报错和排查思路ACL最经典的就是客户端突然报TOPIC_AUTHORIZATION_FAILED。我们第一次遇到时花了不少时间排查后来总结出三步法第一确认客户端账号确实是报错时用的账号注意连接池里如果混用了多个账号很可能用的是其他账号的权限。第二用kafka-acls.sh --list --topic 对应Topic检查Topic的ACL列表确认Principal写法和大小写是否一致。第三检查allow.everyone.if.no.acl.found的值——如果还是true说明根本不会拒绝报错原因可能在其他地方。另外注意ACL匹配是精确匹配User:data-pipeline和User:data_pipeline是两个完全不同的Principal配置时很容易在这栽跟头。6. 生产环境客户端适配与升级中的三个大坑6.1 多客户端协议适配Kafka生态的客户端不止Java一种现在很多团队用Python、Go写服务适配方案要提前想清楚。Java客户端参考上一节配置即可。这里给Python的示例用的confluent-kafka库from confluent_kafka import Producer conf { bootstrap.servers: kafka1.example.com:9093,kafka2.example.com:9093,kafka3.example.com:9093, security.protocol: SASL_SSL, sasl.mechanism: SCRAM-SHA-256, sasl.username: data-pipeline, sasl.password: app-secret, ssl.ca.location: /etc/kafka/security/ca-cert.pem, } producer Producer(**conf)注意ssl.ca.location指向CA证书的PEM文件而不是JKS格式。Python的librdkafka底层不吃JKS这个坑几乎每个团队都会遇到。Go的segmentio/kafka-go不支持SASL_SSL建议使用confluent-kafka-go或franz-go。如果项目里已经用了segmentio/kafka-go可以考虑只连9092明文端口不做安全认证但这就违背了加锁初衷所以还是建议统一换库。6.2 坑一证书过期引发的集群雪崩TLS证书默认设2年实际运行中大家很容易忘掉这个到期时间。我们有一次在证书过期后一周才发现问题——压测时客户端的SSL握手间歇性失败才想起证书到期了。避免方案在CI/CD流程里加证书到期检查脚本提前30天告警同时建立证书轮转SOP。证书轮转时要确认Broker的keystore更新后旧证书还在信任库中并且每个Broker逐个滚动重启避免整个集群同时下线。6.3 坑二用户名密码动态更新时客户端连接池失效SCRAM用户是支持动态更新的但有一个隐藏问题如果客户端是长连接Broker端更新密码后旧连接仍然有效直到连接断开重连。这带来一个隐患——密码泄露后你要踢掉旧的客户端连接但Kafka并没有提供主动断开某个用户连接的能力。我的做法是更新密码的同时在网关或负载均衡层对客户端节点做滚动重启强制旧连接断开。因为大多数客户端应用在Kubernetes中跑滚动重启的代价不高。6.4 坑三可视化工具与监控体系的兼容性团队日常会依赖Kafka的工具来排查问题比如Kafka Tool、Kafka UI、Burrow等。加认证后这些工具都需要同步配置TLS和认证信息。如果你的监控脚本还在用JMX裸连9092端口记得一并升级。我们用的是开源的Kafka UI配置项里支持SASL_SSL和SCRAM-SHA-256但有一个细节如果集群用了多个listener需要在工具的配置里明确选择SASL_SSL地址否则它默认连PLAINTEXT端口认证永远失败。最后聊两句实操体会安全认证这套东西一次性配置完成之后日常基本感觉不到它的存在。真正让人头疼的是证书管理和账号生命周期管理。我现在的习惯是每个新Topic创建时同步写好对应的ACL脚本每次证书快到期时提前一个月在日历上提醒新应用接入集群时给一套标准化的客户端配置模板而不是让各团队自己摸索。Kafka集群的安全加固不是在业务量增长后再补的功课而是从第一天起就该规划好的基础设施。