Kafka 已经成为实时流处理的“血脉”。只是若不做好平安防护，敏感信息就像娱乐的电线一样随时可能被窃取或篡改。本文将手把手带你在 Ubuntu 系统上为 Kafka 装上“防弹衣”， 不如... 从证书生成到 SASL 认证、ACL 权限细粒度控制，一路敲出最可靠的数据通道。

一、整体平安思路与架构概览

先把大局想清楚：加密 + 认证 + 授权 + 网络隔离 四层防护缺一不可。下面这张简图帮你快速定位每一层的职责：

二、准备工作：安装依赖与创建专用系统用户

# 更新系统并安装 JDK
sudo apt update && sudo apt install -y openjdk-11-jdk
# 创建运行用户，降低权限风险
sudo groupadd kafka
sudo useradd -r -g kafka -s /sbin/nologin kafka
# 为 Kafka 指定目录
sudo mkdir -p /opt/kafka/{config,logs,data}
sudo chown -R kafka:kafka /opt/kafka

别小看这一步，很多漏洞都是主要原因是 Kafka 直接跑在 root 下导致的，精辟。。

三、 生成 TLS/SSL 证书

以下示例使用 OpenSSL 手动创建根 CA 与服务端/客户端证书，生产环境请换成内部 CA 或商业证书。

# 创建存放证书的目录
mkdir -p /etc/kafka/ssl/{ca,server,client}
cd /etc/kafka/ssl
# 1️⃣ 根 CA
openssl req -new -x509 -days 3650 \
   -keyout ca/ca-key.pem -out ca/ca-cert.pem \
   -subj "/CN=Kafka-Root-CA"
# 2️⃣ Broker 密钥 & CSR
openssl req -newkey rsa:2048 -nodes \
   -keyout server/server-key.pem \
   -out server/server.csr \
   -subj "/CN=kafka-broker"
# 用根 CA 为 Broker 签名
openssl x509 -req -in server/server.csr \
   -CA ca/ca-cert.pem -CAkey ca/ca-key.pem \
   -CAcreateserial -days 3650 \
   -out server/server-cert.pem
# 同理生成 client/client-key.pem 与 client/client-cert.pem
# 
# 合并为 JKS
keytool -importcert -trustcacerts \
   -alias kafka-ca \
   -file ca/ca-cert.pem \
   -keystore truststore.jks \
   -storepass changeit \
   -noprompt
keytool -importkeystore \
   -srckeystore server/server-keystore.p12 \
   -srcstoretype PKCS12 \
   -srcstorepass changeit \
   -destkeystore keystore.jks \
   -deststoretype JKS \
   -deststorepass changeit

*温馨提示*: 为了避免 “找不到信任库” 的尴尬，请务必把根 CA 导入所有节点的 truststore 中，总的来说...。

四、 配置 SASL 身份验证

SASL 能让每个生产者和消费者都拥有独立的凭据，失窃后也能快速吊销。

# 在 /opt/kafka/config/kafka_server_jaas.conf 中写入：
KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required;
};
# 客户端 JAAS 示例
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="writer"
    password="WriterPass!2024";
};

SASL 配置完成后 需要在 broker 的 s 推倒重来。 erver.properties 中打开对应协议：

# 启用双向加密 + SASL
listeners=SASL_SSL://0.0.0.0:9093
advertised.listeners=SASL_SSL://your.host.com:9093
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
# 指定 JAAS 文件路径，让 JVM 能加载它
KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
export KAFKA_OPTS
# SSL 基础参数
ssl.keystore.location=/opt/kafka/config/keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/opt/kafka/config/truststore.jks
ssl.truststore.password=changeit
# 强制使用 TLSv1.2+ 与平安套件
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.cipher.suites=TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384
require.ssl.client.auth=true          # 双向认证必须开启！
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false # 没有 ACL 时拒绝访问！
super.users=User:admin                # 超级管理员账号， 可自行添加多条，用逗号分隔。
principal.builder.class=org.apache.kafka.common.security.auth.DefaultPrincipalBuilder
enable.sasl=true                       # 必须打开！
sasl.enabled.mechanisms=SCRAM-SHA-512,PASSWORD,PLAIN     # 根据需求增删。
listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;
listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required;
listener.name.sasl_ssl.gssapi.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required;
listener.name.sasl_ssl.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
listener.name.sasl_ssl.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.server.auth.oauthbearer.secured.OAuthBearerValidatorCallbackHandler;
listener.name.sasl_ssl.oauthbearer.token.endpoint.uri=https://your-idp.example.com/oauth/token;
producer.ssl.truststore.location=/opt/kafka/config/truststore.jks 
producer.ssl.truststore.password=changeit 
consumer.ssl.truststore.location=/opt/kafka/config/truststore.jks 
consumer.ssl.truststore.password=changeit 
client.id=my-producer-client-id               # 为每个客户端设定唯一 ID，有助于审计。
group.id=my-consumer-group                     # 消费组同理。
offsets.topic.replication.factor=3             # 高可用，防止单点故障。
transaction.state.log.replication.factor=3     # 开启事务时必备。
transaction.state.log.min.isr=2                # 保持足够 ISR。
log.retention.hours=168                        # 默认保留一周，可根据业务调节。
log.segment.bytes=1073741824                   # 每段文件约 1GB，有利于磁盘回收。
num.partitions=6                               # 初始分区数，可根据吞吐量调优。
## ==== 将上述内容复制进 server.properties ====
## 注意：不要遗漏等号两侧的空格，否则启动会报错！ 
## 如果你仍然使用 Zookeeper， 请把 zookeeper.connect 加进来；如果迁移到 KRaft，则不需要此项。

五、 细粒度 ACL 权限管理——只让该来的进，只让该走的出！

看好你哦！ SASL 验证完身份后还要决定它能干什么。下面演示如何为生产者 & 消费者分别授予写读权限：

# 创建 SCRAM 用户
$KAFKA_HOME/bin/kafka-configs.sh --zookeeper localhost:2181 \ 
      --alter --add-config 'SCRAM-SHA-512=' \ 
      --entity-type users --entity-name writer
$KAFKA_HOME/bin/kafka-configs.sh --zookeeper localhost:2181 \ 
      --alter --add-config 'SCRAM-SHA-512=' \ 
      --entity-type users --entity-name reader
# 为 writer 添加写权限 & 创建 topic 权限
$KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ 
      --add --allow-principal User:writer \ 
      --operation Write --operation Create \ 
      --topic orders
# 为 reader 添加读权限 & 描述权限
$KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ 
      --add --allow-principal User:reader \ 
      --operation Read --operation Describe \ 
      --topic orders
# 为 consumer group 授权读取偏移量 
$KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ 
      --add --allow-principal User:reader \ 
      --operation Read --operation Describe \ 
      --group my-consumer-group
# 查看已配置 ACL 列表
$KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \ 
      --list –topic orders

出道即巅峰。 小贴士：如果你看到 “AuthorizationException” 报错， 请先检查是否忘记打开 authorizer.enable=true，并确认 super.users 包含 admin。

六、网络层面加固——只让可信 IP 踏进来！

• UFW 防火墙规则：

# 打开内部通信端口， 仅限内部网段访问  
sudo ufw allow from 10.0.0.0/16 to any port 9093 proto tcp comment 'Kafka SSL'  
# 禁止明文端口  
sudo ufw deny 9092/tcp  
# 启动 UFW  
sudo ufw enable  
sudo ufw status verbose  

• Iptables 高级过滤：

# 保存到 /etc/iptables/rules.v4  
*filter  
-A INPUT –p tcp –m tcp –dport 9093 –s 10.0.0.0/16 –j ACCEPT  
-A INPUT –p tcp –m tcp –dport 9093 –j DROP  
COMMIT  
iptables-restore 

• 云平台平安组：

如果你的机器跑在 AWS、阿里云或 GCP，请同样只开放内部 VPC 子网 IP 段； 说到底。 外部访问只能走 VPN 或跳板机。

\

七、 验证与排错——让每一步都有回声

SASL+SSL 握手测试

# 使用内置 console producer 测试写入   
KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf" \\
$KAFKA_HOME/bin/kafka-console-producer.sh \\   
--broker-list localhost:9093 \\   
--topic orders \\   
--producer-property security.protocol=SASL_SSL \\   
--producer-property sasl.mechanism=SCRAM-SHA-512  
# 同理，用 console consumer 拉取   
KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_client_jaas.conf" \\
$KAFKA_HOME/bin/kafka-console-consumer.sh \\   
--bootstrap-server localhost:9093 \\   
--topic orders \\   
--from-beginning \\   
--consumer-property security.protocol=SASL_SSL \\   
--consumer-property sasl.mechanism=SCRAM-SHA-512    
\end{verbatim}
未来可期。 If you see “Auntication failed”, double‑check  password in  JAAS file and ensure  SCRAM credentials have been added to Zookeeper/KRaft.
Acl 生效检查 
$KAFKA_HOME/bin/kafka-acls.sh --authorizer-properties zookeeper.connect localhost:2181 --list –topic orders  
输出示例:
Current ACLs for resource `TopicPattern:`orders`:
    Principal    Host    Operation    PermissionType    PatternType    
    User:writer   *       Write           Allow                 Literal    
    User:writer   *       Create         Allow                  Literal    
    User:reader   *       Read            Allow                 Literal    
    User:reader   *       Describe        Allow                 Literal    
    Group:*        *       Read            Allow               Literal    
若缺失某行，请重新施行对应 kafka-acls.sh 命令。 
八、日常运维与监控建议 

The broker’s logs are your first line of defence—set log.level = INFO for production and keep an eye on “SASL auntication failed” or “Failed to verify SSL certificate”. Use tools like Filebeat or Loki to ship logs centrally.
Migrate to KRaft mode when possible; it eliminates Zookeeper as a single point of failure and lets you store ACL/SCRAM data in  same quorum.
If you have multiple data centers, enable inter‑cluster replication over SASL_SSL with separate keystores per region—this avoids cross‑region credential leakage.
Create automated scripts that rotate keystores every six months and purge old SCRAM passwords via cron jobs.
The easiest way to audit who accessed which topic is to enable broker’s audit logging , n push it into Elasticsearch for quick queries.
If you notice spikes in latency after enabling TLS, consider enabling hardware offload  or using NGINX/TCP Proxy as a TLS terminator—但记得仍保留双向认证！​
永远不要把密码硬编码进脚本或 Docker 镜像中，而是使用 Vault 或 Kubernetes Secret 来动态注入。
九、 ——从“开门见山”到“闭门严锁”只差一步之遥 
Kakfa 本身已经足够强大，但没有平安装甲，它就是暴露在外的易碎玻璃。通过本文所述四层防护， 你可以确保：

所有网络流量都经过 TLS 加密，不会被抓包。
每个客户端必须，即使拿到了网络包也无从下手。
ACL 精准控制谁能读写哪些 Topic，杜绝横向越权。
只允许受信任 IP 接入，彻底封锁外部攻击面。

©2026 技术分享站 | 本文基于个人实践撰写，仅供参考。
若用于生产，请结合实际环境进行充分测试。