对于许多企业来说，Kafka是整个企业数据系统的中枢和源泉，因此保护事件流平台对于数据安全至关重要，并且通常是管理层要求的。本文会讲述五个安全类别以及Kafka和Confluent平台的基本功能，这些功能可以帮助用户保护事件流平台。

认证方式

认证机制会验证接入Kafka和Confluent平台的用户和应用的身份。需要关注的与认证相关的主要领域有三个：Kafka的Broker或Confluent的服务端，Apache ZooKeeper的服务端以及基于HTTP的服务。HTTP服务包括Kafka Connect工作节点、ksqlDB服务端、Confluent REST代理、模式注册表和Control Center。验证与这些服务的所有连接是安全的基础。

Kafka Broker和Confluent服务端

Kafka Broker和Confluent服务端使用简单认证和安全层（SASL）或mutual TLS（mTLS）对来自客户端和其他Broker的连接进行身份验证。

SASL是用于身份验证的框架，提供了多种身份验证机制。当前支持的SASL机制包括PLAIN、SCRAM、GSSAPI（Kerberos）和OAUTHBEARER（注意：OAUTHBEARER的开发者必须提供代码以获取和验证凭据）。Kafka Broker可以同时启用多种机制，然后客户端可以选择用于身份验证的机制。Kafka监听器大致是Broker接受连接的IP、端口和安全协议。每种机制的SASL配置略有不同，但通常包括启用所需机制的列表并配置代理监听器，以下是GSSAPI配置的摘要：

sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
listeners=SASL_SSL://kafka1:9093
advertised.listeners=SASL_SSL://localhost:9093

详细的信息，请参见SASL文档。

使用mTLS进行认证时，SSL/TLS证书在服务端和客户端之间提供基本的信任关系，但是mTLS不能与任何SASL机制结合使用，为特定的监听器启用SASL后，将禁用SSL/TLS客户端认证。

在支持的认证方法中，mTLS、SASL SCRAM和SASL GSSAPI是当前的KafkaBroker建议的认证方法。对于Confluent服务端，当与Confluent提供的LDAP回调和SSL/TLS配对时，还可以使用PLAIN机制。使用这些方法可为企业生产提供更好的安全性实践。强烈建议始终集成SSL/TLS以保护网络上的身份验证凭据，下面的加密部分在保护传输中的数据方面将进一步展开介绍。

ZooKeeper服务端

Kafka利用ZooKeeper来执行各种关键的集群管理任务。防止在ZooKeeper中进行未经授权的修改对于维护稳定和安全的Kafka集群至关重要。Kafka的2.5版本开始附带了支持mTLS认证的ZooKeeper，通过启用/禁用SASL认证可以启用ZooKeeper的身份验证。在2.5版之前，仅支持DIGEST-MD5和GSSAPI SASL机制。

要为ZooKeeper启用mTLS身份验证，必须同时配置ZooKeeper和Kafka。

以下是启用mTLS的ZooKeeper配置片段示例：

zookeeper.connect=zk1:2182,zk2:2182,zk3:2182
zookeeper.ssl.client.enable=true
zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty
zookeeper.ssl.keystore.location=<path-to-kafka-keystore>
zookeeper.ssl.keystore.password=<kafka-keystore-password>
zookeeper.ssl.truststore.location=<path-to-kafka-trustore>
zookeeper.ssl.truststore.password=<kafka-truststore-password>
zookeeper.set.acl=true

HTTP服务

Kafka Connect、Confluent模式注册表、REST代理、MQTT代理、ksqlDB和Control Center，在HTTP（S）监听器、REST API和用户界面上都支持身份验证。Kafka和Confluent平台支持Basic身份验证和mTLS，不过Confluent平台还提供了通过HTTP Basic身份验证对LDAP进行认证的选项。

授权

身份验证验证用户和应用的身份，而授权则定义连接后允许其执行的操作。

ACLs

Apache Kafka带有现成的授权机制来建立ACL配置，以保护特定的Kafka资源。Kafka ACL存储在ZooKeeper中，并通过kafka-acls命令进行管理。对于Kafka ACL，可以使用SimpleAclAuthorizer，以验证在身份验证过程中确定的主体。可以使用SCRAM用户名、Kerberos主体或客户端证书来标识客户端。Confluent平台还提供了旧版LDAP授权器及其替代者Confluent服务端授权器，可用于与现有LDAP基础架构集成。

Kafka ACL提供了细粒度的方法来管理对单个Kafka集群的访问，但没有扩展到Kafka Connect、Confluent模式注册表、Control Center或ksqlDB等其他服务。基于ZooKeeper的ACL也是面向集群的，因此如果要管理多个集群，需要独立管理所有集群中的ACL。最后，由于没有角色的概念，使用大量的主体（用户、组或应用）和控件可能难以管理ACL。

Confluent平台通过集中式ACL可以帮助解决其中一些ACL管理缺陷，通过基于角色的访问控制，从一个集中的权威集群管理多个集群的ACL。

基于角色的访问控制（RBAC）

Confluent平台提供了基于角色的访问控制（RBAC），可解决上面列出的问题。RBAC和ACL功能类似，但处理问题的方式不同，可以独立使用也可协同使用。RBAC由Confluent的元数据服务（MDS）提供支持，该元数据服务与LDAP集成在一起，并充当授权和身份验证数据的核心协调者。RBAC利用角色绑定来确定哪些用户和组可以访问特定资源，以及可以在这些资源（角色）中执行哪些操作。RBAC通过Confluent服务端在Kafka集群上获得授权，Confluent服务端是完全兼容的Kafka Broker，集成了RBAC等商业安全功能。

管理员为各种资源上的用户和组分配预定义的角色，每个用户可以在每个资源上分配多个角色。资源就是诸如Kafka集群，消费者组或仅是主题之类的东西。管理员创建角色绑定，将主体、角色和资源映射在一起，可以使用Control Center或者Confluent CLI中的iam rolebinding子命令管理角色绑定，例如：

confluent iam rolebinding create --principal User:ConnectAdmin --role ResourceOwner --resource Topic:connect-statuses --kafka-cluster-id abc123

RBAC还与Confluent平台中的许多其他应用紧密集成，以提供面向特定应用量身定制的集群级和资源级绑定。例如，带RBAC的Kafka Connect可以将连接器作为资源进行控制，并通过集群级角色绑定管理对整个分布式Kafka Connect集群的访问。RBAC模式注册表集成还允许将模式主题视为资源，从而将一组模式注册表视为集群并绑定角色。

加密

保护传输中的数据

对Broker与应用之间传输的数据进行加密可减轻跨网络意外拦截数据的风险。身份验证和加密是在Kafka中的每个监听器上定义的。Kafka监听器默认使用PLAINTEXT这样的不安全协议，这适用于开发环境，但是所有的生产环境都应启用加密。使用listener配置以及管理信任库和密钥库来配置Broker上的加密，下面的代码片段显示了两个监听器，其在Broker上一起使用SSL和SASL_SSL：

listeners=SSL://:9093,SASL_SSL://:9094

使用ssl配置键来配置密钥和可信证书，例如：

ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

SSL/TLS还可以通过在客户端和服务端之间分配可信证书来用于客户端身份验证，有关此方法的更多详细信息，请参见Confluent的相关文档。

为了将不受保护的集群迁移到受保护的集群，Kafka支持多个监听器，只需配置一个新的受保护的监听器并迁移客户端应用即可。迁移所有客户端后，就可以停用不受保护的监听器。

Broker受保护后，还必须配置客户端的安全性。按照与Broker类似的过程，为所有客户端（包括ksqlDB、Kafka Connect和Kafka Streams）配置加密。

机密保护

启用加密和安全功能需要配置机密值，包括密码、密钥和主机名。Kafka没有提供任何保护这些值的方法，开发者经常将它们存储在明文配置文件和源代码控制系统中，通常会看到这样的机密配置：

ssl.key.password=test1234

通用安全实践规定，机密不应该以明文形式存储在配置文件中，也不应体现在应用日志的输出中，以防止机密无意泄漏。Confluent平台提供了机密保护功能，它利用了信封加密，这是一种通过高度安全的方法保护加密机密的行业标准。

应用此功能将导致配置文件中包含用于ConfigProvider获取机密值而非实际明文值的指令：

ssl.key.password=${securepass:secrets.file:server.properties/ssl.key.password}

可监测性

监测安全事件和服务指标并对之做出反应有助于维护可靠和安全的集群。服务中断可能是由于配置错误、应用有故障等原因导致，Confluent平台提供了相关的核心功能，使运维人员可以监控平台内发生的事件。

审计日志

Kafka没有直接提供维护授权决定记录的机制，运维人员需要从Broker那里捕获和解析非结构化的日志，以便建立有关ACL决策的监测结果。

Confluent平台通过结构化的审计日志对Kafka进行了补充，该日志提供了一种捕获、保护的机制，并将授权活动保存到主题中。具体来说，审计日志记录权限检查的运行时决策，这些权限检查是在用户/应用尝试执行受ACL和RBAC保护的操作时发生的。虽然消费者可能会无序地消费审计日志消息，但是可审计的事件是按时间顺序记录的。这些结构化事件使运维人员可以在其集群上实时调查可操作的安全事件。审计日志是使用CloudEvents构建的，CloudEvents是与供应商无关的规范，用于定义事件数据的模式，并可以与各种行业工具集成。

监控

监控和跟踪集群指标是监控异常（包括安全漏洞）的关键，在Kafka中可以配置JMX端口并实现自定义指标和监控解决方案，或者与各种第三方监控解决方案集成。

借助Confluent平台，监控功能可以直接在各个组件之间紧密集成。Confluent Metrics Reporter会收集来自Kafka Broker的各种指标然后将数据输出到Kafka主题。Confluent Metrics Reporter可以将数据输出到与其监测的集群不同的集群，从而减少生产集群上的操作负载。Confluent Metrics Reporter与Confluent Control Center集成在一起，可以提供多集群，仪表盘风格的视图。

Confluent Control Center是与RBAC集成的，后者使用户可以查看自己的权限，在拥有其他权限的情况下管理其他用户的角色绑定，并完全保持对关键运营指标的可见性，这简化了整个企业的安全管理。

Confluent Monitoring Interceptors是客户端插件，可捕获客户端性能统计信息并将其转发给Kafka主题。和Confluent Metrics Reporter一样，监控拦截器可以将这些记录转发到其他集群，以减少生产集群上的操作负载。Confluent Control Center与Confluent Monitoring Interceptors集成在一起，可以提供客户端状态以及Broker数据的完整视图。

报警

指标和数据是了解集群的运行状况和安全性的关键，但是如果不采取任何措施，它们就会失效。Confluent Control Center提供了报警功能，这是提高集群状态感知能力的关键。报警是通过定义触发器和动作来工作的，触发器是与条件配对的指标，这些条件定义了何时应触发触发器。动作与触发器相关联，并定义触发触发器时发生的活动。操作可以与电子邮件，Slack或流行的PagerDuty事件响应平台集成。

数据治理

数据治理包括企业的策略和控制措施，以确保高可用和高质量的数据系统。数据系统中的消费端应用依赖于格式良好的数据记录和可靠的连接，Confluent平台扩展了Kafka，以在此领域提供增强的功能。

数据质量

Confluent Schema Validation通过为数据生产者强制执行模式合规性，从而提供记录级数据完整性。使用模式验证，如果无法从中央模式注册表中获得有效模式标识，就可以阻止生产者写入数据。这种策略通常称为写模式，除了保护数据完整性外，还使消费者应用免于在读取时验证模式的负担。为了构建灵活的系统，模式验证支持模式进化的概念，允许消费者和生产者跨版本进行通信。使用Confluent服务端，可以在Confluent Control Center或命令行中按主题启用模式验证：

kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test-validation-true --config confluent.value.schema.validation=true

在Confluent平台5.5版本中，现在支持Protobuf和JSON Schema格式，以及现有的Apache Avro™支持，模式验证完全支持这些新格式。

配额

客户端应用可能以很高的速率生产或消费数据，从而占满了集群资源，从而导致剩余客户端资源不足。Kafka配额通过限制有问题的应用的资源利用率来为所有客户端提供资源保护。当Broker检测到超过配额时，会试图通过引入人为延迟以使违规客户端低于阈值来减慢他的速度，这些配置可以保护集群避免遇到流量高峰的客户端带来的资源不足。在Broker级别设置的默认配额还可以保护Broker免受恶意客户端的拒绝服务攻击。可以向下面这样在Broker中配置默认​​配额，它将默认的生产者和消费者速率设置为10MBps：

quota.producer.default=10485760
quota.consumer.default=10485760

在集群运行时，可以为每个客户端ID或用户分配自定义配额。下面的示例将客户端clientA的生产者和消费者速率阈值分别设置为1KB和2KB：

bin/kafka-configs --bootstrap-server localhost:9092 --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048' --entity-name clientA --entity-type clients

合规性

许多行业要求政府强制执行其他法规。Confluent平台支持HIPAA和SSAE 18 SOC2合规性认证，Confluent Cloud支持GDPR，ISO 27001，PCI 2级，SOC 1、2和3标准，并且Confluent Cloud企业版还支持HIPAA合规性。