未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

安全配置 - 构建Kafka的安全堡垒

为什么安全不是可选项

假设一家互联网公司凌晨3点收到安全告警：离职员工的 Kafka 访问权限未及时回收，生产环境没有启用认证，网络可达者能够连接；敏感数据明文传输，且缺少访问日志，难以追踪数据去向。为说明风险，假设有500万用户信息在数小时内被下载；这里的公司、调查结论和数量均为示例设定，不作为已核实事故引用。

在这一假设场景中，安全配置成为头等大事。Kafka 的安全体系由四层构成：

网络隔离层（防火墙 / VPC / 端口控制）
      |
加密层（SSL/TLS 传输加密）
      |
认证层（SASL - 确认你是谁）
      |
授权层（ACL - 确认你能做什么）

---

SSL/TLS 加密配置

核心概念：Keystore 与 Truststore

概念	存放内容	作用
Keystore	自己的私钥 + 证书	向对方证明自己的身份
Truststore	受信任的 CA 证书	决定信任哪些对方

CA与Broker证书生成模板（SAN签发步骤需另行补充）

# 第一步：生成 CA 私钥和自签名证书
openssl req -new -x509 -keyout ca-key.pem -out ca-cert.pem -days 365 \
  -subj "/C=CN/ST=Beijing/O=MyOrg/CN=kafka-ca" \
  -passout pass:ca-secret

# 第二步：为每个 Broker 生成 Keystore，并放入私钥和证书
BROKER=kafka-broker-01
PASSWORD=broker-secret

# 生成 Broker 的 Keystore（含私钥）
keytool -keystore ${BROKER}.keystore.jks \
  -alias ${BROKER} \
  -validity 365 \
  -genkey \
  -keyalg RSA \
  -storepass ${PASSWORD} \
  -keypass ${PASSWORD} \
  -dname "CN=${BROKER}, OU=IT, O=MyOrg, L=Beijing, C=CN"

# 生成证书签名请求（CSR）
keytool -certreq \
  -keystore ${BROKER}.keystore.jks \
  -alias ${BROKER} \
  -storepass ${PASSWORD} \
  -file ${BROKER}.csr

# 用 CA 签名生成 Broker 证书
openssl x509 -req \
  -CA ca-cert.pem \
  -CAkey ca-key.pem \
  -in ${BROKER}.csr \
  -out ${BROKER}-signed.pem \
  -days 365 \
  -CAcreateserial \
  -passin pass:ca-secret

# 将 CA 证书导入 Keystore（建立信任链）
keytool -keystore ${BROKER}.keystore.jks \
  -alias CARoot \
  -import -file ca-cert.pem \
  -storepass ${PASSWORD} \
  -noprompt

# 将签名后的 Broker 证书导入 Keystore
keytool -keystore ${BROKER}.keystore.jks \
  -alias ${BROKER} \
  -import -file ${BROKER}-signed.pem \
  -storepass ${PASSWORD}

# 第三步：创建 Truststore（供 Client 和其他 Broker 验证此 Broker）
keytool -keystore ${BROKER}.truststore.jks \
  -alias CARoot \
  -import -file ca-cert.pem \
  -storepass ${PASSWORD} \
  -noprompt

echo "证书生成完成：${BROKER}.keystore.jks 和 ${BROKER}.truststore.jks"

Broker 端 SSL 配置（server.properties）

# 监听器：同时开放明文端口和 SSL 端口
listeners=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
advertised.listeners=PLAINTEXT://kafka-broker-01:9092,SSL://kafka-broker-01:9093
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL

# Keystore：Broker 自己的证书和私钥
ssl.keystore.location=/kafka/ssl/kafka-broker-01.keystore.jks
ssl.keystore.password=broker-secret
ssl.key.password=broker-secret

# Truststore：Broker 信任的 CA（用于验证客户端证书）
ssl.truststore.location=/kafka/ssl/kafka-broker-01.truststore.jks
ssl.truststore.password=broker-secret

# 客户端认证模式
# none       = 不验证客户端证书（单向 TLS）
# requested  = 请求客户端证书，但不强制
# required   = 必须提供客户端证书（双向 mTLS）
ssl.client.auth=required

# 协议和算法
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.endpoint.identification.algorithm=HTTPS

SSL 配置项速查表

配置项	默认值	说明
ssl.keystore.location	无	JKS 格式 Keystore 文件路径
ssl.keystore.password	无	Keystore 密码
ssl.key.password	无	Keystore 中私钥的密码
ssl.truststore.location	无	Truststore 文件路径
ssl.truststore.password	无	Truststore 密码
ssl.client.auth	none	客户端认证：none/requested/required
ssl.enabled.protocols	TLSv1.2,TLSv1.3	启用的 TLS 版本
ssl.endpoint.identification.algorithm	https	主机名验证算法（设为空字符串可关闭）
ssl.cipher.suites	无	指定加密套件（留空使用 JVM 默认）

Client 端 SSL 配置（Java）

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-01:9093");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// SSL 安全协议
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");

// Keystore（客户端自己的证书，用于 mTLS 场景）
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/kafka/ssl/client.keystore.jks");
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "client-secret");
props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, "client-secret");

// Truststore（客户端信任的 CA，用于验证 Broker 证书）
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/kafka/ssl/client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "client-secret");

props.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, "TLSv1.2,TLSv1.3");
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS");

---

SASL 认证配置

五种 SASL 机制对比

机制	全称	认证原理	适用场景	安全强度
PLAIN	Simple Authentication	明文用户名密码（依赖 SSL 加密）	开发测试、内网简单场景	低（必须配 SSL）
SCRAM-SHA-256	Salted Challenge Response	基于盐值的挑战-响应，密码不在网络传输	生产环境推荐	高
SCRAM-SHA-512	同上，更长哈希	同 SCRAM-256，哈希更安全	高安全要求场景	更高
GSSAPI (Kerberos)	Generic Security Services	Kerberos票据认证	企业级，已有 AD/KDC 基础设施	极高
OAUTHBEARER	OAuth 2.0 Bearer Token	JWT/OAuth Token 验证	云原生、微服务体系	高（依赖 Token 实现）

SCRAM-SHA-256 配置（生产推荐）

第一步：通过已运行的 Broker 管理 SCRAM 用户凭据（以下命令使用 --bootstrap-server；首次启动前的凭据准备需按部署模式另行处理）

# 创建管理员用户
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-256=[iterations=8192,password=admin-pass-2024]' \
  --entity-type users \
  --entity-name kafka-admin

# 创建生产者用户
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-256=[password=producer-pass-2024]' \
  --entity-type users \
  --entity-name app-producer

# 创建消费者用户
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-256=[password=consumer-pass-2024]' \
  --entity-type users \
  --entity-name app-consumer

# 查看已创建的用户
kafka-configs.sh --bootstrap-server localhost:9092 \
  --describe --entity-type users

第二步：Broker server.properties 配置

# 监听器配置
listeners=PLAINTEXT://0.0.0.0:9092,SASL_SSL://0.0.0.0:9094
advertised.listeners=PLAINTEXT://kafka-01:9092,SASL_SSL://kafka-01:9094
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SASL_SSL:SASL_SSL

# SASL 机制
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256

# Broker 间通信也使用 SASL_SSL
security.inter.broker.protocol=SASL_SSL

# SSL 证书（同上一节）
ssl.keystore.location=/kafka/ssl/kafka-01.keystore.jks
ssl.keystore.password=broker-secret
ssl.key.password=broker-secret
ssl.truststore.location=/kafka/ssl/kafka-01.truststore.jks
ssl.truststore.password=broker-secret

第三步：JAAS 配置文件（kafka_jaas.conf）

// Broker 自身的认证配置（Broker 间通信使用）
KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka-admin"
    password="admin-pass-2024";
};

// Broker 作为 Client 连接其他 Broker 时使用
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka-admin"
    password="admin-pass-2024";
};

启动时通过 JVM 参数指定 JAAS 文件：

export KAFKA_OPTS="-Djava.security.auth.login.config=/kafka/config/kafka_jaas.conf"
bin/kafka-server-start.sh config/server.properties

第四步：Client 端 SASL_SSL 配置

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-01:9094");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 安全协议：SASL + SSL
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");

// JAAS 配置（内嵌方式，无需外部文件）
String jaasTemplate =
    "org.apache.kafka.common.security.scram.ScramLoginModule required " +
    "username=\"app-producer\" " +
    "password=\"producer-pass-2024\";";
props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasTemplate);

// SSL Truststore（验证 Broker 证书）
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/kafka/ssl/client.truststore.jks");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "client-secret");
props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "HTTPS");

SASL/PLAIN 配置（内网简单场景）

以下 PLAIN 示例在 kafka_jaas.conf 的 KafkaServer 块中静态定义用户；采用此静态配置方式时，修改用户配置后需重启相应 Broker：

// kafka_jaas.conf - PLAIN 机制
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_producer="producer-secret"
    user_consumer="consumer-secret";
};

---

ACL 授权配置

启用 ACL 授权器

在 server.properties 中添加：

# 启用 ACL 授权器（Kafka 内置）
authorizer.class.name=kafka.security.authorizer.AclAuthorizer

# 若没有匹配 ACL，拒绝访问（生产必须为 false）
allow.everyone.if.no.acl.found=false

# 超级用户（不受 ACL 约束，多个用分号分隔）
super.users=User:kafka-admin;User:kafka-ops

kafka-acls.sh 完整命令速查

生产者权限模板（对指定 Topic 写权限）：

# 授予 app-producer 对 orders Topic 的写权限
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:app-producer \
  --operation Write \
  --operation Describe \
  --topic orders

# 若要对所有以 "app-" 开头的 Topic 授权（前缀匹配）
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:app-producer \
  --operation Write \
  --operation Create \
  --operation Describe \
  --topic "app-" \
  --resource-pattern-type prefixed

消费者权限模板（Topic 读权限 + 消费者组权限）：

# 授予 app-consumer 对 orders Topic 的读权限
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:app-consumer \
  --operation Read \
  --operation Describe \
  --topic orders

# 授予 app-consumer 对消费者组的权限（必须同时配置，否则无法提交 offset）
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:app-consumer \
  --operation Read \
  --group order-consumer-group

管理员权限模板（Topic 管理操作）：

# 授予 kafka-ops 对所有 Topic 的管理权限
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:kafka-ops \
  --operation All \
  --topic "*"

# 授予 kafka-ops 对集群级别操作的权限
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --add \
  --allow-principal User:kafka-ops \
  --operation Alter \
  --operation AlterConfigs \
  --operation DescribeConfigs \
  --cluster

查看和删除 ACL：

# 查看所有 ACL
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --list

# 查看特定 Topic 的 ACL
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --list --topic orders

# 删除 ACL（参数与 --add 相同，改为 --remove）
kafka-acls.sh --bootstrap-server kafka-01:9094 \
  --command-config /kafka/config/admin-client.properties \
  --remove \
  --allow-principal User:app-producer \
  --operation Write \
  --topic orders

管理员客户端配置文件（admin-client.properties）：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="kafka-admin" \
    password="admin-pass-2024";
ssl.truststore.location=/kafka/ssl/client.truststore.jks
ssl.truststore.password=client-secret
ssl.endpoint.identification.algorithm=HTTPS

多监听器与内外网隔离

以下片段仅演示内外网监听器的划分：内网使用 PLAINTEXT，外网使用 SASL_SSL。它不符合本文生产环境禁用明文的要求，也不包含完整的KRaft控制器安全配置，不能作为生产安全基线直接使用。

# 定义三个监听器
listeners=INTERNAL://0.0.0.0:9092,EXTERNAL://0.0.0.0:9094,CONTROLLER://0.0.0.0:9093

# 每个监听器对应的安全协议
listener.security.protocol.map=\
  INTERNAL:PLAINTEXT,\
  EXTERNAL:SASL_SSL,\
  CONTROLLER:SSL

# 对外广播地址（内外网不同 IP 或域名）
advertised.listeners=\
  INTERNAL://10.0.1.5:9092,\
  EXTERNAL://kafka.example.com:9094

# Broker 间通信使用内网监听器
inter.broker.listener.name=INTERNAL

---

安全审计日志

启用授权日志

以下片段仅适用于使用Log4j 1配置语法的环境；采用其他日志实现时不能直接照搬。在适用的 log4j.properties 中添加：

# 授权事件日志（此 INFO 配置不保证记录全部允许事件；需按授权器实现确认日志级别与覆盖范围）
log4j.logger.kafka.authorizer.logger=INFO, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false

log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=/kafka/logs/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

授权日志字段示例（不代表上述INFO配置会输出所有事件；常用内置授权器的允许事件通常需要DEBUG级别）：

[2026-03-21 10:23:45,123] DEBUG Principal = User:app-consumer is Allowed Operation = Read
  from host = 10.0.2.33 on resource = Topic:LITERAL:orders (kafka.authorizer.logger)

[2026-03-21 10:23:46,789] INFO Principal = User:unknown is Denied Operation = Write
  from host = 192.168.1.200 on resource = Topic:LITERAL:orders (kafka.authorizer.logger)

关键字段说明：

字段	含义
Principal	发起请求的用户
Allowed/Denied	授权结果
Operation	操作类型（Read/Write/Describe 等）
host	客户端 IP
resource	被访问的资源（Topic/Group/Cluster）

---

常见安全配置错误表

错误现象	根本原因	正确做法
SSL handshake failed: certificate_unknown	客户端 Truststore 没有导入 Broker CA 证书	将 ca-cert.pem 导入客户端 Truststore
Authentication failed: Invalid username or password	SCRAM 用户未在 Broker 创建，或密码错误	执行 kafka-configs.sh --add-config 'SCRAM-SHA-256=...'
Authorization failed: User is not authorized	未配置 ACL，且 allow.everyone.if.no.acl.found=false	为用户添加对应 Topic/Group 的 ACL
消费者无法提交 Offset	只授权了 Topic 的 Read，没有授权 ConsumerGroup	同时添加 --operation Read --group <groupId>
Broker 间无法通信	security.inter.broker.protocol 与 JAAS 中 KafkaServer 的机制不匹配	确保 sasl.mechanism.inter.broker.protocol 在 JAAS KafkaServer 块中已定义
开启主机名验证后连接失败	证书缺少与实际连接主机名或IP地址匹配的SAN身份	重新签发包含匹配SAN的证书并保持主机名验证开启；不应以关闭验证作为生产修复方案
PLAIN 密码在日志中泄露	应用、配置输出或调试日志记录了密码	停止记录并脱敏秘密、限制日志访问并轮换泄露凭据；TLS保护网络传输，但不能消除日志泄露

---

安全最佳实践检查清单

证书管理：

• 使用企业 CA 签发证书，而非自签名（生产）
• 设置证书有效期不超过 1 年，配置自动轮换提醒
• 私钥文件权限设置为 600，只有 kafka 用户可读
• 不同环境（dev/test/prod）使用不同的 CA 和证书

认证授权：

• 生产环境禁用 PLAINTEXT 监听器
• 外网连接应同时具备传输加密和客户端认证，可采用 SASL_SSL，或强制客户端证书验证的 mTLS；单向 TLS 本身不提供客户端认证
• 遵循最小权限原则：按实际操作授予生产者所需的 Topic 及其他资源权限、消费者所需的 Topic 和消费者组权限，不授予无关权限
• 离职员工/下线服务应及时撤销其认证凭据及相关 ACL；SCRAM 凭据可按实际启用机制通过 kafka-configs.sh --delete-config 删除，其他认证机制需在对应身份系统中撤销
• 定期（每季度）审查 ACL 列表，清理冗余权限

监控审计：

• 开启授权日志，保留至少 90 天
• 对 DENIED 日志配置告警，超过阈值立即通知安全团队
• 关键操作（ACL 变更、配置变更）纳入审计系统

---

掌握检查

完成以下检查，确认已掌握 Kafka 安全配置：

• [ ] 能解释本文CA、Broker证书及Keystore/Truststore生成模板，并识别实际TLS练习仍需补充的SAN签发步骤和客户端mTLS材料
• [ ] 能区分 SASL/PLAIN、SCRAM-SHA-256、GSSAPI 的适用场景
• [ ] 能配置 server.properties 中的 SASL_SSL 监听器和 JAAS 文件
• [ ] 能用 kafka-acls.sh 完成生产者/消费者/管理员的标准权限配置
• [ ] 能解释 allow.everyone.if.no.acl.found=false 的重要性
• [ ] 能读懂授权日志并识别安全异常

练习题

1. 公司要求所有 Kafka 连接必须双向 TLS（mTLS），请说明需要在哪些配置文件做哪些改动？
2. 新员工需要读取 user-events Topic 的权限，同时只能使用消费者组 analytics-group，写出完整的 kafka-acls.sh 命令。
3. 你发现授权日志中某个 IP 在 10 分钟内有 500 次 DENIED 记录，可能是什么情况？应如何处置？
4. 解释为什么 SASL/SCRAM 比 SASL/PLAIN 更安全（从密码存储和传输两个维度说明）。

---

学习连接

前置知识: 监控与故障排查 下一步学习: Kafka Connect - 构建数据管道 相关概念:

• 数据安全基础
• PKI 证书体系
• 企业安全管理

---