企业官网建设流程全解析

在 Rasa 开源仓库中搭建无 TLS 的 Kafka SASL SCRAM-SHA-256 认证测试环境

【免费下载链接】rasa💬 Open source machine learning framework to automate text- and voice-based conversations: NLU, dialogue management, connect to Slack, Facebook, and more - Create chatbots and voice assistants项目地址: https://gitcode.com/GitHub_Trending/ra/rasa

本文以 Rasa 仓库中 test_environments/message_and_event_brokers/kafka/sasl_scram/no_tls/scram_sha_256/README.md 为主体，完整讲解如何用 Docker Compose 搭建一套启用 SASL SCRAM-SHA-256 认证、但不启用 TLS 加密的 Kafka 消息代理，用于验证 Rasa 对话引擎在"客户端必须认证、且走明文连接"场景下的事件流输出。读完本文，你将掌握 Kafka + ZooKeeper 双容器的认证配置、JAAS 登录文件的编写、SCRAM 用户的创建命令，以及 Rasa 端KafkaEventBroker对应的endpoints.yml连接参数与源码级实现原理。

背景与适用场景

Rasa 的核心对话引擎在运行时会持续产生 tracker 事件（用户消息、意图预测、动作执行等）。当需要把这些事件流交给下游系统消费时，Rasa 通过EventBroker抽象层把事件发布到消息中间件，而 Kafka 是其中最常用的实现之一（见 rasa/core/brokers/kafka.py）。

在生产环境中，Kafka 集群通常要求客户端携带凭证认证。为了在不依赖外部基础设施的前提下验证 Rasa 的认证接入能力，仓库在 test_environments/message_and_event_brokers/kafka/ 下维护了一整套"可一键拉起"的 Kafka 测试环境，按认证方式与加密方式两两组合：

• 无认证（no_authentication）
• SASL_PLAIN，分无 TLS 与带 TLS
• SASL_SCRAM，分无 TLS 与带 TLS，且 SCRAM 又细分为 SHA-256 与 SHA-512 两个算法版本

本文聚焦的组合是：SASL_SCRAM 认证 + SHA-256 算法 + 无 TLS 明文传输。它模拟的是"客户端必须通过 SCRAM 握手认证，但信道本身不加密"的中间态场景——这种配置常被用于先打通认证链路、再叠加 TLS 的分步改造路径。

环境概览：目录结构与端口规划

先看该组合的完整文件清单：

test_environments/message_and_event_brokers/kafka/sasl_scram/no_tls/scram_sha_256/ ├── README.md # 启动说明 ├── docker-compose.yml # 容器编排 ├── broker_jaas.conf # Kafka Broker 的 JAAS 登录配置 ├── zookeeper_client_jaas.conf # 连接 ZooKeeper 时使用的 JAAS 客户端配置 └── zookeeper_server_jaas.conf # ZooKeeper 服务端的 JAAS 配置

根据目录 README.md 与 docker-compose.yml，端口规划遵循仓库的通用约定：

端口编号遵循仓库约定：Kafka 监听909x、ZooKeeper 监听218x，数字随认证配置不同而变化（详见 kafka 目录总 README），从而允许多套测试环境在同一台机器上并行运行而不冲突。

深入 docker-compose.yml：ZooKeeper 与 Broker 的认证配置

ZooKeeper 服务

ZooKeeper 服务段 使用confluentinc/cp-zookeeper:7.3.2镜像，关键环境变量如下：

zookeeper: image: confluentinc/cp-zookeeper:7.3.2 container_name: zookeeper-sasl-scram-sha-256-no-tls ports: - "2186:2186" environment: ZOOKEEPER_SERVER_ID: 1 ZOOKEEPER_CLIENT_PORT: 2186 ZOOKEEPER_TICK_TIME: 2000 ZOOKEEPER_LOG4J_ROOT_LOGLEVEL: "DEBUG" KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/secrets/zookeeper_server_jaas.conf -Dquorum.auth.enableSasl=true -Dquorum.auth.learnerRequireSasl=true -Dquorum.auth.serverRequireSasl=true -Dquorum.cnxn.threads.size=20 -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider -Dzookeeper.authProvider.2=org.apache.zookeeper.server.auth.DigestAuthenticationProvider -DjaasLoginRenew=3600000 -DrequireClientAuthScheme=sasl -Dquorum.auth.learner.loginContext=QuorumLearner -Dquorum.auth.server.loginContext=QuorumServer volumes: - ./zookeeper_server_jaas.conf:/etc/kafka/secrets/zookeeper_server_jaas.conf - ./zookeeper_client_jaas.conf:/etc/kafka/client/zookeeper_client_jaas.conf

要点解读：

• -Djava.security.auth.login.config=...指定 ZooKeeper 进程使用的 JAAS 文件，即挂载进来的zookeeper_server_jaas.conf；
• quorum.auth.enableSasl=true以及 learner/server 两侧的RequireSasl=true开启 ZooKeeper 集群成员之间的 SASL 认证（单节点模式下同样生效）；
• 同时注册了SASLAuthenticationProvider与DigestAuthenticationProvider两个认证提供者，其中 Digest 用于 admin 用户，SASL 用于 quorum 内部与 Kafka broker 的连接；
• zookeeper_client_jaas.conf被挂载到容器内/etc/kafka/client/目录，供启动脚本进入容器创建 SCRAM 用户时使用。

Kafka Broker 服务

Broker 服务段 使用confluentinc/cp-kafka:7.3.2镜像：

kafka-broker: image: confluentinc/cp-kafka:7.3.2 container_name: kafka-broker-sasl-scram-sha-256-no-tls ports: - "9096:9096" depends_on: - zookeeper environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2186 KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://localhost:9096 KAFKA_MIN_INSYNC_REPLICAS: 1 KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-256 KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SASL_PLAINTEXT KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-256 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" KAFKA_OFFSETS_RETENTION_MINUTES: 172800 KAFKA_LOG4J_LOGGERS: "kafka.authorizer.logger=DEBUG,kafka.controller=DEBUG" KAFKA_LOG4J_ROOT_LOGLEVEL: "DEBUG" KAFKA_SUPER_USERS: User:kafkabroker;User:kafkaclient KAFKA_ZOOKEEPER_SASL_ENABLED: "true" KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false" KAFKA_OPTS: -Dzookeeper.sasl.client=true -Dzookeeper.sasl.clientconfig=Client -Djava.security.auth.login.config=/etc/kafka/secrets/conf/kafka_server_jaas.conf volumes: - ./broker_jaas.conf:/etc/kafka/secrets/conf/kafka_server_jaas.conf

要点解读：

• KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://localhost:9096表明监听器协议为SASL_PLAINTEXT（即"带 SASL 认证、但无 TLS 加密"），客户端必须携带凭证；
• KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-256与KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: SCRAM-SHA-256把服务端启用机制和 broker 间通信机制都固定为 SCRAM-SHA-256；
• KAFKA_ZOOKEEPER_SASL_ENABLED: "true"表示 broker 连接 ZooKeeper 时也要走 SASL 认证，配合KAFKA_OPTS中的-Dzookeeper.sasl.client=true与-Dzookeeper.sasl.clientconfig=Client，使用 JAAS 中名为Client的上下文（Digest 登录）完成对 ZooKeeper 的认证；
• KAFKA_SUPER_USERS: User:kafkabroker;User:kafkaclient声明了两个超级用户；
• KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false"表示未找到 ACL 时不默认放行所有人（注意：本配置未显式设置 authorizer 类，ACL 是否真正生效取决于 Kafka 版本与 authorizer 的启用情况，这一点在排障时需要留意）；
• 挂载的broker_jaas.conf被映射为kafka_server_jaas.conf，通过KAFKA_OPTS的-Djava.security.auth.login.config交给 JVM 加载。

JAAS 配置文件逐项解读

SASL 认证的密钥全部集中在三个 JAAS 文件中，它们是整套环境能否拉起的关键。

zookeeper_server_jaas.conf（ZooKeeper 服务端）

文件 zookeeper_server_jaas.conf：

Server { org.apache.zookeeper.server.auth.DigestLoginModule required user_admin="password"; }; QuorumServer { org.apache.zookeeper.server.auth.DigestLoginModule required user_zookeeper="password"; }; QuorumLearner { org.apache.zookeeper.server.auth.DigestLoginModule required username="zookeeper" password="password"; };

• Server上下文定义了 ZooKeeper 服务端可接受的客户端凭据：用户admin、密码password（这就是 broker 和运维脚本连接 ZooKeeper 时使用的账号）；
• QuorumServer/QuorumLearner两个上下文用于 ZooKeeper 集群成员之间的相互认证（对应docker-compose.yml中-Dquorum.auth.server.loginContext=QuorumServer与-Dquorum.auth.learner.loginContext=QuorumLearner），本测试环境为单节点，但这些配置保证了集群模式下也能直接复用。

zookeeper_client_jaas.conf（ZooKeeper 客户端）

文件 zookeeper_client_jaas.conf：

Client { org.apache.zookeeper.server.auth.DigestLoginModule required username="admin" password="password"; };

这个Client上下文被两处使用：一是 Kafka broker 通过KAFKA_OPTS的-Dzookeeper.sasl.clientconfig=Client引用它来连接 ZooKeeper；二是下面启动流程中执行kafka-configs创建 SCRAM 用户时，通过KAFKA_OPTS="-Djava.security.auth.login.config=zookeeper_client_jaas.conf"指定它。两处都使用admin/password与服务端的Server上下文对应。

broker_jaas.conf（Kafka Broker 服务端与客户端）

文件 broker_jaas.conf：

KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="kafkabroker" password="password"; }; Client { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="password"; }; KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="kafkaclient" password="password"; };

• KafkaServer上下文是 broker 自身向集群认证时使用的身份：用户kafkabroker（与 README 中创建的用户、KAFKA_SUPER_USERS中的声明保持一致），登录模块为ScramLoginModule；
• Client上下文用PlainLoginModule提供admin/password，对应连接 ZooKeeper 的 Digest 认证；
• KafkaClient上下文定义了 broker 以客户端身份与其它 broker 通信（inter-broker）时的 SCRAM 凭据kafkaclient/password。

一步步启动：创建用户与启动 Broker

原文档 README.md 给出了完整的启动流程，分三步：

第一步：启动 ZooKeeper

docker-compose up -d zookeeper

第二步：进入 ZooKeeper 容器，创建 SCRAM 用户

SCRAM 用户凭证存储在 ZooKeeper 中，因此必须在 ZooKeeper 容器内执行kafka-configs命令：

docker exec -it zookeeper-sasl-scram-sha-256-no-tls bash cd /etc/kafka/client KAFKA_OPTS="-Djava.security.auth.login.config=zookeeper_client_jaas.conf" kafka-configs --zookeeper localhost:2186 --alter --add-config 'SCRAM-SHA-256=[iterations=4096,password=password]' --entity-type users --entity-name kafkabroker KAFKA_OPTS="-Djava.security.auth.login.config=zookeeper_client_jaas.conf" kafka-configs --zookeeper localhost:2186 --alter --add-config 'SCRAM-SHA-256=[iterations=4096,password=password]' --entity-type users --entity-name client

参数说明：

第三步：退出容器并启动 Kafka Broker

exit docker-compose up -d kafka-broker

docker-compose.yml中kafka-broker通过depends_on: - zookeeper保证 ZooKeeper 先就绪。启动完成后，Kafka 将在localhost:9096上提供带 SASL_SCRAM 认证的服务。

仓库文件中的一处细节值得注意：README 创建的第二个用户名为client，而 broker_jaas.conf 的KafkaClient上下文与KAFKA_SUPER_USERS中使用的是kafkaclient。实际接入客户端时，请以你自己的客户端配置为准，确保"ZooKeeper 中创建的用户名"与"客户端 SASL 用户名"完全一致，否则会认证失败。

客户端侧：让 Rasa 通过 SASL SCRAM 连接 Kafka

endpoints.yml 配置

测试环境就绪后，Rasa 通过endpoints.yml中的event_broker段接入 Kafka。仓库在 data/test_endpoints/event_brokers/ 提供了各种协议组合的示例，其中kafka_sasl_plaintext_endpoint.yml展示了SASL_PLAINTEXT协议下的完整字段：

event_broker: type: kafka security_protocol: SASL_PLAINTEXT topic: topic url: localhost partition_by_sender: True sasl_username: username sasl_password: password sasl_mechanism: PLAIN

对接本文的 SCRAM-SHA-256 环境，只需把sasl_mechanism改为SCRAM-SHA-256、url指向localhost:9096、sasl_username/sasl_password填成在 ZooKeeper 中创建的用户（例如client/password）：

event_broker: type: kafka url: localhost:9096 topic: rasa_core_events security_protocol: SASL_PLAINTEXT sasl_mechanism: SCRAM-SHA-256 sasl_username: client sasl_password: password partition_by_sender: true

KafkaEventBroker 源码解析

rasa/core/brokers/kafka.py 中的KafkaEventBroker是这套配置的落地实现。其构造函数明确支持 SCRAM 系列机制（见sasl_mechanism参数 docstring）：

Valid values are: PLAIN, GSSAPI, OAUTHBEARER, SCRAM-SHA-256, SCRAM-SHA-512. Default:PLAIN

security_protocol的合法值为PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL，默认SASL_PLAINTEXT。

当security_protocol == "SASL_PLAINTEXT"时，_get_kafka_config()会生成如下 confluent-kafka 生产者配置（见 kafka.py 中_get_kafka_config方法）：

authentication_params = { "sasl.username": self.sasl_username, "sasl.password": self.sasl_password, "sasl.mechanism": self.sasl_mechanism, "security.protocol": self.security_protocol, }

也就是说：sasl_mechanism: SCRAM-SHA-256会被原样传递给 confluent-kafka 的sasl.mechanism，与测试环境 Broker 端KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-256一一对应。若传入非法机制或协议，_get_kafka_config会抛出ValueError（非法security_protocol）或由_create_producer抛出KafkaProducerInitializationError；仓库在 data/test_endpoints/event_brokers/ 也准备了kafka_invalid_sasl_mechanism.yml、kafka_invalid_security_protocol.yml等异常用例供测试参考。

此外，KafkaEventBroker的事件主题默认值为rasa_core_events（构造函数topic: Text = "rasa_core_events"），可用topic字段覆盖；partition_by_sender为true时，会以sender_id作为消息分区键（_publish方法中partition_key = bytes(event.get("sender_id"), ...)），保证同一会话的事件按序进入同一分区。

验证连接与常见故障排查

启动完成后可用任意支持 SASL 的 Kafka 客户端验证连接。若使用 Rasa 运行对话，可通过 event-brokers 文档 了解事件流的整体行为。

排查认证类问题时可重点关注 kafka.py 中的kafka_error_callback：

if ( err.code() == KafkaError._ALL_BROKERS_DOWN or err.code() == KafkaError._AUTHENTICATION or err.code() == KafkaError._MAX_POLL_EXCEEDED ): raise KafkaException(err)

即_AUTHENTICATION（认证失败）和_ALL_BROKERS_DOWN（broker 不可达）会被直接视为异常抛出，日志中通常会看到Failed to connect kafka.或Connection to kafka lost, reconnecting...（publish方法内部会尝试重建 producer 并重连）。常见原因与对策：

可复用的其他 Kafka 测试环境组合

本文环境只是仓库提供的一种组合。若你的验证场景不同，可直接切换到以下兄弟配置（均位于 kafka 测试环境目录）：

• 无认证的 Kafka：适合先验证事件流链路本身；
• SASL_PLAIN 无 TLS：用户名密码式明文认证；
• SASL_SCRAM 无 TLS + SHA-512：与本文结构完全相同，仅算法换为更长的 SHA-512；
• SASL_SCRAM 带 TLS + SHA-256：在本文基础上叠加 SSL 证书（该目录附带ca-cert、server.keystore.jks等文件），客户端需同时配置ssl_cafile等参数。

各组合的端口、容器名、JAAS 文件与endpoints.yml配置遵循同一套模式，掌握了本文的 SCRAM-SHA-256 无 TLS 方案后，迁移到其他组合只需对照相应 README 调整机制名、端口与证书参数。

注意事项与限制

• 仅用于测试：本环境关闭了 TLS，所有凭据（包括password）均为明文可读的固定值，只适合本地开发与 CI 验证，不能直接用于生产；
• 明文传输风险：SASL_PLAINTEXT只做认证、不加密数据，若需保护数据在途安全，请使用 带 TLS 的 SASL_SCRAM 组合；
• 用户名一致性：创建用户、Broker JAAS、客户端配置三处的用户名/密码必须全局一致（参考上文关于client与kafkaclient的提示）；
• 镜像版本：本套环境固定使用 Confluent 平台镜像7.3.2，若更换镜像版本，需自行核对 SASL/SCRAM 相关环境变量的兼容性；
• 仓库只读：本文描述的启动与配置均为使用仓库既有文件的方式，无需修改仓库内容即可复现整套环境。

【免费下载链接】rasa💬 Open source machine learning framework to automate text- and voice-based conversations: NLU, dialogue management, connect to Slack, Facebook, and more - Create chatbots and voice assistants项目地址: https://gitcode.com/GitHub_Trending/ra/rasa