本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # Amazon MSK 中的安全性 云安全 AWS 是重中之重。作为 AWS 客户，您可以受益于专为满足大多数安全敏感型组织的要求而构建的数据中心和网络架构。 安全是双方 AWS 的共同责任。[责任共担模式](https://aws.amazon.com/compliance/shared-responsibility-model/)将其描述为云*的*安全性和云*中*的安全性： + **云安全** — AWS 负责保护在 AWS 云中运行 AWS 服务的基础架构。 AWS 还为您提供可以安全使用的服务。作为[AWS 合规计划合规计划合规计划合](https://aws.amazon.com/compliance/programs/)的一部分，第三方审计师定期测试和验证我们安全的有效性。要了解适用于 Amazon Managed Streaming for Apache Kafka 的合规性计划，请参阅[按合规性计划提供的范围内 Amazon Web Services](https://aws.amazon.com/compliance/services-in-scope/)。 + **云端安全**-您的责任由您使用的 AWS 服务决定。您还需要对其它因素负责，包括您的数据的敏感性、您公司的要求以及适用的法律法规。 该文档可帮助您了解如何在使用 Amazon MSK 时应用责任共担模式。以下主题说明如何配置 Amazon MSK 以实现您的安全性与合规性目标。您还会了解如何使用其他 Amazon Web Services 帮助您监控和保护 Amazon MSK 资源。 **Topics** + [Amazon Managed Streaming for Apache Kafka 中的数据保护](data-protection.md) + [亚马逊 MSK 的身份验证和授权 APIs](security-iam.md) + [Apache Kafka 的身份验证和授权 APIs](kafka_apis_iam.md) + [更改 Amazon MSK 集群的安全组](change-security-group.md) + [控制对亚马逊 MSK ZooKeeper 集群中 Apache 节点的访问权限](zookeeper-security.md) + [Amazon Managed Streaming for Apache Kafka 的合规性验证](MSK-compliance.md) + [Amazon Managed Streaming for Apache Kafka 中的恢复能力](disaster-recovery-resiliency.md) + [Amazon Managed Streaming for Apache Kafka 中的基础设施安全性](infrastructure-security.md) # Amazon Managed Streaming for Apache Kafka 中的数据保护 分担责任模式 AWS [分担责任模型](https://aws.amazon.com/compliance/shared-responsibility-model/)适用于适用于 Apache Kafka 的 Apache Streaming for Apache Streaming 中的数据保护。如本模型所述 AWS ，负责保护运行所有内容的全球基础架构 AWS 云。您负责维护对托管在此基础结构上的内容的控制。您还负责您所使用的 AWS 服务 的安全配置和管理任务。有关数据隐私的更多信息，请参阅[数据隐私常见问题](https://aws.amazon.com/compliance/data-privacy-faq/)。有关欧洲数据保护的信息，请参阅 *AWS Security Blog* 上的 [AWS Shared Responsibility Model and GDPR](https://aws.amazon.com/blogs/security/the-aws-shared-responsibility-model-and-gdpr/) 博客文章。 出于数据保护目的，我们建议您保护 AWS 账户 凭证并使用 AWS IAM Identity Center 或 AWS Identity and Access Management (IAM) 设置个人用户。这样，每个用户只获得履行其工作职责所需的权限。还建议您通过以下方式保护数据： + 对每个账户使用多重身份验证（MFA）。 + 用于 SSL/TLS 与 AWS 资源通信。我们要求使用 TLS 1.2，建议使用 TLS 1.3。 + 使用设置 API 和用户活动日志 AWS CloudTrail。有关使用 CloudTrail 跟踪捕获 AWS 活动的信息，请参阅《*AWS CloudTrail 用户指南》*中的[使用跟 CloudTrail 踪](https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-trails.html)。 + 使用 AWS 加密解决方案以及其中的所有默认安全控件 AWS 服务。 + 使用高级托管安全服务（例如 Amazon Macie），它有助于发现和保护存储在 Amazon S3 中的敏感数据。 + 如果您在 AWS 通过命令行界面或 API 进行访问时需要经过 FIPS 140-3 验证的加密模块，请使用 FIPS 端点。有关可用的 FIPS 端点的更多信息，请参阅《美国联邦信息处理标准（FIPS）第 140-3 版》[https://aws.amazon.com/compliance/fips/](https://aws.amazon.com/compliance/fips/)。 强烈建议您切勿将机密信息或敏感信息（如您客户的电子邮件地址）放入标签或自由格式文本字段（如**名称**字段）。这包括您 AWS 服务 使用控制台、API 或与 Amazon MSK 或其他机构 AWS CLI合作时。 AWS SDKs在用于名称的标签或自由格式文本字段中输入的任何数据都可能会用于计费或诊断日志。如果您向外部服务器提供 URL，强烈建议您不要在网址中包含凭证信息来验证对该服务器的请求。 **Topics** + [Amazon MSK 加密](msk-encryption.md) + [Amazon MSK 加密入门](msk-working-with-encryption.md) + [将 Amazon MSK APIs 与接口 VPC 终端节点配合使用](privatelink-vpc-endpoints.md) # Amazon MSK 加密 Amazon MSK 提供了数据加密选项，可使用这些选项来满足严格的数据管理要求。Amazon MSK 用于加密的证书必须每 13 个月续订一次。Amazon MSK 会自动为所有集群续订这些证书。Amazon MSK 开始进行证书更新操作时，快速代理集群仍处于 `ACTIVE` 状态。对于标准代理集群，Amazon MSK 会集群在启动证书更新操作时将集群状态设置为 `MAINTENANCE`。待更新完成后，MSK 会将集群状态重新设置为 `ACTIVE`。当集群处于证书更新操作时，您可以继续生成和使用数据，但无法对数据执行任何更新操作。 ## Amazon MSK 静态加密 Amazon MSK 与 [AWS Key Management Service](https://docs.aws.amazon.com/kms/latest/developerguide/)（KMS）集成以提供透明的服务器端加密。Amazon MSK 始终加密您的静态数据。当创建 MSK 集群时，您可以指定您希望 Amazon MSK 用于加密静态数据的 AWS KMS key 。如果您不指定 KMS 密钥，Amazon MSK 会为您创建一个 [AWS 托管式密钥](https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#aws-managed-cmk) 并代表您使用它。有关 KMS 密钥的更多信息，请参阅《AWS Key Management Service 开发人员指南》**中的 [AWS KMS keys](https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#kms_keys)。 ## Amazon MSK 传输中加密 Amazon MSK 会使用 TLS 1.2。默认情况下，它会加密在 MSK 集群的代理之间传输的数据。可以在创建集群时覆盖此默认值。 对于客户端和代理之间的通信，您必须指定下列三项设置之一： + 仅允许 TLS 加密数据。这是默认设置。 + 同时允许明文数据和 TLS 加密数据。 + 仅允许明文数据。 亚马逊 MSK 经纪人使用公共 AWS Certificate Manager 证书。因此，任何信任 Amazon Trust Services 的信任库也会信任 Amazon MSK 代理的证书。 虽然我们强烈建议启用传输中加密，但它可能会增加额外的 CPU 开销和几毫秒的延迟。但是，大多数使用案例对这些差异并不敏感，影响的程度取决于集群、客户端和使用情况配置文件的配置。 # Amazon MSK 加密入门 创建 MSK 集群时，您可以使用 JSON 格式指定加密设置。示例如下： ``` { "EncryptionAtRest": { "DataVolumeKMSKeyId": "arn:aws:kms:us-east-1:123456789012:key/abcdabcd-1234-abcd-1234-abcd123e8e8e" }, "EncryptionInTransit": { "InCluster": true, "ClientBroker": "TLS" } } ``` 对于 `DataVolumeKMSKeyId`，您可以为账户 (`alias/aws/kafka`) 中的 MSK 指定[客户托管密钥](https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#customer-cmk)或 AWS 托管式密钥 。如果您未指定`EncryptionAtRest`，Amazon MSK 仍会加密您的静态数据。 AWS 托管式密钥要确定您的集群使用的密钥，请发送 `GET` 请求或调用 `DescribeCluster` API 操作。 对于 `EncryptionInTransit`，`InCluster` 的默认值为 true，但是如果您不想在代理之间传递数据时让 Amazon MSK 加密数据，则可以将此项设置为 false。 要为客户端和代理之间传输的数据指定加密模式，请将 `ClientBroker` 设置为以下三个值之一：`TLS`、`TLS_PLAINTEXT` 或 `PLAINTEXT`。 **Topics** + [创建 Amazon MSK 集群时指定加密设置](msk-working-with-encryption-cluster-create.md) + [测试 Amazon MSK TLS 加密](msk-working-with-encryption-test-tls.md) # 创建 Amazon MSK 集群时指定加密设置 此过程介绍了如何在创建 Amazon MSK 集群时指定加密设置。 **创建集群时指定加密设置** 1. 将上一示例的内容保存在文件中，并为该文件指定所需的任何名称。例如，将其命名为 `encryption-settings.json`。 1. 运行 `create-cluster` 命令并使用 `encryption-info` 选项指向您保存配置 JSON 的文件。示例如下：替换为*\$1YOUR MSK VERSION\$1*与 Apache Kafka 客户端版本相匹配的版本。有关如何查找 MSK 集群版本的信息，请参阅[确定 MSK 集群版本](create-topic.md#find-msk-cluster-version)。请注意，使用与 MSK 集群版本不同的 Apache Kafka 客户端版本，可能会导致 Apache Kafka 数据损坏、丢失和停机。 ``` aws kafka create-cluster --cluster-name "ExampleClusterName" --broker-node-group-info file://brokernodegroupinfo.json --encryption-info file://encryptioninfo.json --kafka-version "{YOUR MSK VERSION}" --number-of-broker-nodes 3 ``` 以下是运行此命令后的成功响应示例。 ``` { "ClusterArn": "arn:aws:kafka:us-east-1:123456789012:cluster/SecondTLSTest/abcdabcd-1234-abcd-1234-abcd123e8e8e", "ClusterName": "ExampleClusterName", "State": "CREATING" } ``` # 测试 Amazon MSK TLS 加密 此过程介绍了如何在 Amazon MSK 上测试 TLS 加密。 **测试 TLS 加密** 1. 按照[步骤 3：创建客户端计算机](create-client-machine.md)中的指导创建客户端计算机。 1. 在客户端计算机上安装 Apache Kafka。 1. 在本示例中，我们使用 JVM 信任库与 MSK 集群通信。为此，请首先在客户端计算机上创建一个名为 `/tmp` 的文件夹。然后，转到 Apache Kafka 安装的 `bin` 文件夹，并运行以下命令。（您的 JVM 路径可能不相同。） ``` cp /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64/jre/lib/security/cacerts /tmp/kafka.client.truststore.jks ``` 1. 仍在客户端计算机上的 Apache Kafka 安装的 `bin` 文件夹中，创建一个名为 `client.properties` 的文本文件，该文件包含以下内容。 ``` security.protocol=SSL ssl.truststore.location=/tmp/kafka.client.truststore.jks ``` 1. 在 AWS CLI 安装了的计算机上运行以下命令，*clusterARN*替换为集群的 ARN。 ``` aws kafka get-bootstrap-brokers --cluster-arn clusterARN ``` 成功结果如下所示。保存此结果，因为您需要在下一步中使用它。 ``` { "BootstrapBrokerStringTls": "a-1.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123,a-3.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123,a-2.example.g7oein.c2.kafka.us-east-1.amazonaws.com:0123" } ``` 1. 运行以下命令，*BootstrapBrokerStringTls*替换为您在上一步中获得的代理端点之一。 ``` /bin/kafka-console-producer.sh --broker-list BootstrapBrokerStringTls --producer.config client.properties --topic TLSTestTopic ``` 1. 打开新的命令窗口并连接到同一台客户端计算机。然后，运行以下命令以创建控制台使用器。 ``` /bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringTls --consumer.config client.properties --topic TLSTestTopic ``` 1. 在生成器窗口中，输入文本消息后点击回车键，并在使用器窗口中查找相同消息。Amazon MSK 对传输中的此消息进行了加密。 有关配置 Apache Kafka 客户端以使用加密数据的更多信息，请参阅[配置 Kafka 客户端](https://kafka.apache.org/documentation/#security_configclients)。 # 将 Amazon MSK APIs 与接口 VPC 终端节点配合使用 您可以使用由 AWS PrivateLink提供支持的接口 VPC 终端节点来防止您的 Amazon VPC 和 Amazon MSK 之间的流量 APIs 离开亚马逊网络。接口 VPC 终端节点不需要互联网网关、NAT 设备、VPN 连接或 AWS Direct Connect 连接。 [AWS PrivateLink](https://docs.aws.amazon.com/vpc/latest/privatelink/what-is-privatelink.html)是一种使用弹性网络接口实现 AWS 服务之间私有通信的 AWS 技术，在 Amazon VPC IPs 中使用私有网络。有关更多信息，请参阅[亚马逊 Virtual Private Cloud](https://docs.aws.amazon.com/vpc/latest/userguide/what-is-amazon-vpc.html) 和[接口 VPC 终端节点 (AWS PrivateLink)](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint)。 您的应用程序可以使用与 Amazon MSK Provisioned 和 MSK Connect 连接 APIs 。 AWS PrivateLink首先，为 Amazon MSK API 创建一个接口 VPC 端点，以便来自和前往 Amazon VPC 资源的流量开始流经接口 VPC 端点。启用 FIPS 的接口 VPC 端点适用于美国区域。有关更多信息，请参阅[创建接口端点](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint)。 使用此功能，Apache Kafka 客户端可动态获取连接字符串以连接预置 MSK 或 MSK Connect 资源，而无需遍历互联网来检索连接字符串。 创建接口 VPC 端点时，请选择以下服务名称端点之一： **对于预置 MSK：** + 新连接不再支持以下服务名称终端节点： + com.amazonaws.region.kafka + com.amazonaws.region.kafka-fips（启用 FIPS） + 支持两者的双栈端点服务 IPv4 以及 IPv6 流量： + wws.api.region.kafka-api + aws.api.region。 kafka-api-fips （支持 FIPS） 要设置双堆栈终端节点，必须遵循[双栈和 FIP](https://docs.aws.amazon.com/sdkref/latest/guide/feature-endpoints.html) S 端点指南。 其中区域指您的区域名称。选择此服务名称可与 MSK 预配置 APIs兼容。有关更多信息，请参阅 *https://docs.aws.amazon.com/msk/1.0/* apireference/ 中的[操作](https://docs.aws.amazon.com/msk/1.0/apireference/operations.html)。 **对于 MSK Connect：** + com.amazonaws.region.kafkaconnect 其中区域指您的区域名称。选择此服务名称即可与 MSK Connec APIs t 兼容。有关更多信息，请参阅《Amazon MSK Connect API 参考》**中的[操作](https://docs.aws.amazon.com/MSKC/latest/mskc/API_Operations.html)。 有关更多信息，包括创建接口 VPC 终端节点的 step-by-step说明，请参阅*AWS PrivateLink 指南*中的[创建接口终端节点](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint)。 ## 控制对亚马逊 MSK 预配置或 MSK Connect 的 VPC 终端节点的访问权限 APIs 借助 VPC 端点策略，您可以控制访问，方式是：将策略附加到 VPC 端点或使用附加到 IAM 用户、组或角色的策略中的额外字段，从而限制只能通过特定 VPC 端点进行访问。使用相应的示例策略，定义对预置 MSK 或 MSK Connect 服务的访问权限。 如果您在创建端点时未附加策略，Amazon VPC 会为您附加一个默认策略，该策略允许对服务的完全访问。终端节点策略不会覆盖或替换 IAM 基于身份的策略或服务特定的策略。这是一个单独的策略，用于控制从端点中对指定服务进行的访问。 有关更多信息，请参阅《AWS PrivateLink 指南》**中的[使用 VPC 端点控制对服务的访问](https://docs.aws.amazon.com/vpc/latest/privatelink/vpc-endpoints-access.html)。 ------ #### [ MSK Provisioned — VPC policy example ] **只读访问权限** 此示例策略可以附加到某个 VPC 端点。（有关更多信息，请参阅控制对 Amazon VPC 资源的访问）。它限制仅能通过其附加的 VPC 端点列出或描述操作。 ``` { "Statement": [ { "Sid": "MSKReadOnly", "Principal": "*", "Action": [ "kafka:List*", "kafka:Describe*" ], "Effect": "Allow", "Resource": "*" } ] } ``` **预置 MSK — VPC 端点策略示例** 限制对特定 MSK 集群的访问 此示例策略可以附加到某个 VPC 端点。它限制通过其附加的 VPC 端点访问特定的 Kafka 集群。 ``` { "Statement": [ { "Sid": "AccessToSpecificCluster", "Principal": "*", "Action": "kafka:*", "Effect": "Allow", "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/MyCluster" } ] } ``` ------ #### [ MSK Connect — VPC endpoint policy example ] **列出连接器并创建新的连接器** 下面是用于 MSK Connect 的端点策略示例。此策略允许指定角色列出连接器，并创建新的连接器。 ``` { "Version": "2012-10-17", "Statement": [ { "Sid": "MSKConnectPermissions", "Effect": "Allow", "Action": [ "kafkaconnect:ListConnectors", "kafkaconnect:CreateConnector" ], "Resource": "*", "Principal": { "AWS": [ "arn:aws:iam::111122223333:role/MyMSKConnectExecutionRole" ] } } ] } ``` **MSK Connect — VPC 端点策略示例** 仅允许指定 VPC 中特定 IP 地址的请求 以下示例显示的策略仅允许来自指定 VPC 中指定 IP 地址的请求成功。来自其它 IP 地址的请求将失败。 ``` { "Statement": [ { "Action": "kafkaconnect:*", "Effect": "Allow", "Principal": "*", "Resource": "*", "Condition": { "IpAddress": { "aws:VpcSourceIp": "192.0.2.123" }, "StringEquals": { "aws:SourceVpc": "vpc-555555555555" } } } ] } ``` ------ # 亚马逊 MSK 的身份验证和授权 APIs AWS Identity and Access Management (IAM) AWS 服务 可帮助管理员安全地控制对 AWS 资源的访问权限。IAM 管理员控制谁可以*通过身份验证*（登录）和*获得授权*（具有权限）来使用 Amazon MSK 资源。您可以使用 IAM AWS 服务 ，无需支付额外费用。 **Topics** + [Amazon MSK 如何与 IAM 配合使用](security_iam_service-with-iam.md) + [Amazon MSK 基于身份的策略示例](security_iam_id-based-policy-examples.md) + [Amazon MSK 的服务相关角色](using-service-linked-roles.md) + [AWS 亚马逊 MSK 的托管策略](security-iam-awsmanpol.md) + [排查 Amazon MSK 身份和访问问题](security_iam_troubleshoot.md) # Amazon MSK 如何与 IAM 配合使用 在使用 IAM 管理对 Amazon MSK 的访问权限之前，您应该了解哪些 IAM 功能可用于 Amazon MSK。要全面了解 Amazon MSK 和其他 AWS 服务如何与 IAM 配合使用，请参阅 IAM *用户指南*中的[与 IAM 配合使用的AWS 服务](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_aws-services-that-work-with-iam.html)。 **Topics** + [Amazon MSK 基于身份的策略](security_iam_service-with-iam-id-based-policies.md) + [Amazon MSK 基于资源的策略](security_iam_service-with-iam-resource-based-policies.md) + [基于 Amazon MSK 标签的授权](security_iam_service-with-iam-tags.md) + [Amazon MSK IAM 角色](security_iam_service-with-iam-roles.md) # Amazon MSK 基于身份的策略 通过使用 IAM 基于身份的策略，您可以指定允许或拒绝的操作和资源以及允许或拒绝操作的条件。Amazon MSK 支持特定的操作、资源和条件键。要了解在 JSON 策略中使用的所有元素，请参阅《IAM 用户指南》** 中的 [IAM JSON 策略元素参考](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements.html)。 ## Amazon MSK 基于身份的策略的操作 管理员可以使用 AWS JSON 策略来指定谁有权访问什么。也就是说，哪个**主体**可以对什么**资源**执行**操作**，以及在什么**条件**下执行。 JSON 策略的 `Action` 元素描述可用于在策略中允许或拒绝访问的操作。在策略中包含操作以授予执行关联操作的权限。 Amazon MSK 中的策略操作在操作前使用以下前缀：`kafka:`。例如，要授予某人使用 Amazon MSK `DescribeCluster` API 操作描述 MSK 集群的权限，您应将 `kafka:DescribeCluster` 操作纳入其策略。策略语句必须包含 `Action` 或 `NotAction` 元素。Amazon MSK 定义了自己的一组操作，以描述您可以使用该服务执行的任务。 请注意，MSK 主题的策略操作在操作前 APIs 使用`kafka-cluster`前缀，请参阅。[IAM 授权策略、操作和资源的语义](kafka-actions.md) 要在单个语句中指定多项操作，请使用逗号将它们隔开，如下所示： ``` "Action": ["kafka:action1", "kafka:action2"] ``` 您也可以使用通配符 （\$1) 指定多个操作。例如，要指定以单词 `Describe` 开头的所有操作，包括以下操作： ``` "Action": "kafka:Describe*" ``` 要查看 Amazon MSK 操作的列表，请参阅《IAM 用户指南》**中的 [Amazon Managed Streaming for Apache Kafka 的操作、资源和条件键](https://docs.aws.amazon.com/service-authorization/latest/reference/list_amazonmanagedstreamingforapachekafka.html)。 ## Amazon MSK 基于身份的策略的资源 管理员可以使用 AWS JSON 策略来指定谁有权访问什么。也就是说，哪个**主体**可以对什么**资源**执行**操作**，以及在什么**条件**下执行。 `Resource` JSON 策略元素指定要向其应用操作的一个或多个对象。作为最佳实践，请使用其 [Amazon 资源名称（ARN）](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html)指定资源。对于不支持资源级权限的操作，请使用通配符 (\$1) 指示语句应用于所有资源。 ``` "Resource": "*" ``` Amazon MSK 实例资源具有以下 ARN： ``` arn:${Partition}:kafka:${Region}:${Account}:cluster/${ClusterName}/${UUID} ``` 有关格式的更多信息 ARNs，请参阅 [Amazon 资源名称 (ARNs) 和 AWS 服务命名空间](https://docs.aws.amazon.com/general/latest/gr/aws-arns-and-namespaces.html)。 例如，要在语句中指定 `CustomerMessages` 实例，请使用以下 ARN： ``` "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/CustomerMessages/abcd1234-abcd-dcba-4321-a1b2abcd9f9f-2" ``` 要指定属于特定账户的所有实例，请使用通配符 (\$1)： ``` "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/*" ``` 无法对特定资源执行某些 Amazon MSK 操作，例如用于创建资源的操作。在这些情况下，您必须使用通配符（\$1)。 ``` "Resource": "*" ``` 要在单个语句中指定多个资源，请 ARNs 用逗号分隔。 ``` "Resource": ["resource1", "resource2"] ``` *要查看亚马逊 MSK 资源类型及其列表 ARNs，请参阅 IAM 用户指南中的[亚马逊托管流媒体为 Apache Kafka 定义的资源](https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonmanagedstreamingforkafka.html#amazonmanagedstreamingforkafka-resources-for-iam-policies)。*要了解您可以在哪些操作中指定每个资源的 ARN，请参阅 [Amazon Managed Streaming for Apache Kafka 定义的操作](https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonmanagedstreamingforkafka.html#amazonmanagedstreamingforkafka-actions-as-permissions)。 ## Amazon MSK 基于身份的策略的条件密钥 管理员可以使用 AWS JSON 策略来指定谁有权访问什么。也就是说，哪个**主体**可以对什么**资源**执行**操作**，以及在什么**条件**下执行。 `Condition` 元素根据定义的条件指定语句何时执行。您可以创建使用[条件运算符](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition_operators.html)（例如，等于或小于）的条件表达式，以使策略中的条件与请求中的值相匹配。要查看所有 AWS 全局条件键，请参阅 *IAM 用户指南*中的[AWS 全局条件上下文密钥](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html)。 Amazon MSK 定义了自己的一组条件键，还支持使用一些全局条件键。要查看所有 AWS 全局条件键，请参阅 *IAM 用户指南*中的[AWS 全局条件上下文密钥](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html)。 要查看 Amazon MSK 条件键的列表，请参阅《IAM 用户指南》**中的 [Amazon Managed Streaming for Apache Kafka 的条件键](https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonmanagedstreamingforkafka.html#amazonmanagedstreamingforkafka-policy-keys)。要了解您可以对哪些操作和资源使用条件键，请参阅 [Amazon Managed Streaming for Apache Kafka 定义的操作](https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonmanagedstreamingforkafka.html#amazonmanagedstreamingforkafka-actions-as-permissions)。 ## Amazon MSK 基于身份的策略的示例 要查看 Amazon MSK 基于身份的策略的示例，请参阅 [Amazon MSK 基于身份的策略示例](security_iam_id-based-policy-examples.md)。 # Amazon MSK 基于资源的策略 Amazon MSK 支持用于 Amazon MSK 集群的集群策略（也称为基于资源的策略）。您可以使用集群策略来定义哪些 IAM 主体拥有跨账户权限来设置与 Amazon MSK 集群的私有连接。当与 IAM 客户端身份验证一起使用时，您也可以使用集群策略为连接的客户端精细定义 Kafka 数据面板的权限。 集群策略支持的最大大小为 20 KB。 要查看如何配置集群策略的示例，请参阅[步骤 2：将集群策略附加到 MSK 集群](mvpc-cluster-owner-action-policy.md)。 # 基于 Amazon MSK 标签的授权 您可以将标签附加到 Amazon MSK 集群。要基于标签控制访问，您需要使用 `kafka:ResourceTag/key-name``aws:RequestTag/key-name` 或 `aws:TagKeys` 条件键在策略的[条件元素](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html)中提供标签信息。有关标记 Amazon MSK 资源的信息，请参阅[为 Amazon MSK 集群添加标签](msk-tagging.md)。 只能借助于标签来控制集群的访问。要向主题和使用者组添加标签，需在策略中添加一条不带标签的单独语句。 要查看基于身份的策略（用于基于集群上的标签来限制对该集群的访问）的示例，请参阅[根据标签访问 Amazon MSK 集群](security_iam_id-based-policy-examples-view-widget-tags.md)。 您可以在基于身份的策略中使用条件，以便基于标签控制对 Amazon MSK 资源的访问权限。此示例演示了如何创建允许用户描述集群、获取其引导代理、列出其代理节点、更新和删除集群的策略。但是，仅当集群标签 `Owner` 的值为该用户的 `username` 时，此策略才会授予权限。以下策略中的第二个语句允许访问集群中的主题。该策略中的第一个语句未授权任何主题访问权限。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "AccessClusterIfOwner", "Effect": "Allow", "Action": [ "kafka:Describe*", "kafka:Get*", "kafka:List*", "kafka:Update*", "kafka:Delete*" ], "Resource": "arn:aws:kafka:us-east-1:123456789012:cluster/*", "Condition": { "StringEquals": { "aws:ResourceTag/Owner": "${aws:username}" } } }, { "Effect": "Allow", "Action": [ "kafka-cluster:*Topic*", "kafka-cluster:WriteData", "kafka-cluster:ReadData" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/*" ] } ] } ``` ------ # Amazon MSK IAM 角色 [IAM 角色](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles.html)是 Amazon Web Services 账户中具有特定权限的实体。 ## 将临时凭证用于 Amazon MSK 可以使用临时凭证进行联合身份验证登录，分派 IAM 角色或分派跨账户角色。您可以通过调用[AssumeRole](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html)或之类的 AWS STS API 操作来获取临时安全证书[GetFederationToken](https://docs.aws.amazon.com/STS/latest/APIReference/API_GetFederationToken.html)。 Amazon MSK 支持使用临时凭证。 ## 服务关联角色 [服务相关角色](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-service-linked-role)允许 Amazon Web Services 访问其他服务中的资源，以代表您完成操作。服务关联角色显示在 IAM 账户中，并归该服务所有。管理员可以查看但不能编辑服务相关角色的权限。 Amazon MSK 支持服务相关角色。有关创建或管理 Amazon MSK 服务相关角色的详细信息，请参阅[Amazon MSK 的服务相关角色](using-service-linked-roles.md)。 # Amazon MSK 基于身份的策略示例 默认情况下，IAM 用户和角色无权执行 Amazon MSK API 操作。管理员必须创建 IAM policy，以便为用户和角色授予权限以对所需的指定资源执行特定的 API 操作。然后，管理员必须将这些策略附加到需要这些权限的 IAM 用户或组。 要了解如何使用这些示例 JSON 策略文档创建 IAM 基于身份的策略，请参阅*《IAM 用户指南》*中的 [在 JSON 选项卡上创建策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_create.html#access_policies_create-json-editor)。 **Topics** + [策略最佳实践](security_iam_service-with-iam-policy-best-practices.md) + [允许用户查看他们自己的权限](security_iam_id-based-policy-examples-view-own-permissions.md) + [访问一个 Amazon MSK 集群](security_iam_id-based-policy-examples-access-one-cluster.md) + [根据标签访问 Amazon MSK 集群](security_iam_id-based-policy-examples-view-widget-tags.md) # 策略最佳实践 基于身份的策略确定某个人是否可以创建、访问或删除您账户中的 Amazon MSK 资源。这些操作可能会使 AWS 账户产生成本。创建或编辑基于身份的策略时，请遵循以下指南和建议： + **开始使用 AWS 托管策略并转向最低权限权限** — 要开始向用户和工作负载授予权限，请使用为许多常见用例授予权限的*AWS 托管策略*。它们在你的版本中可用 AWS 账户。我们建议您通过定义针对您的用例的 AWS 客户托管策略来进一步减少权限。有关更多信息，请参阅《IAM 用户指南》**中的 [AWS 托管策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies)或[工作职能的AWS 托管策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_job-functions.html)。 + **应用最低权限**：在使用 IAM 策略设置权限时，请仅授予执行任务所需的权限。为此，您可以定义在特定条件下可以对特定资源执行的操作，也称为*最低权限许可*。有关使用 IAM 应用权限的更多信息，请参阅《IAM 用户指南》**中的 [IAM 中的策略和权限](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html)。 + **使用 IAM 策略中的条件进一步限制访问权限**：您可以向策略添加条件来限制对操作和资源的访问。例如，您可以编写策略条件来指定必须使用 SSL 发送所有请求。如果服务操作是通过特定的方式使用的，则也可以使用条件来授予对服务操作的访问权限 AWS 服务，例如 CloudFormation。有关更多信息，请参阅《IAM 用户指南》**中的 [IAM JSON 策略元素：条件](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html)。 + **使用 IAM Access Analyzer 验证您的 IAM 策略，以确保权限的安全性和功能性**：IAM Access Analyzer 会验证新策略和现有策略，以确保策略符合 IAM 策略语言（JSON）和 IAM 最佳实践。IAM Access Analyzer 提供 100 多项策略检查和可操作的建议，以帮助您制定安全且功能性强的策略。有关更多信息，请参阅《IAM 用户指南》**中的[使用 IAM Access Analyzer 验证策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access-analyzer-policy-validation.html)。 + **需要多重身份验证 (MFA**)-如果 AWS 账户您的场景需要 IAM 用户或根用户，请启用 MFA 以提高安全性。若要在调用 API 操作时需要 MFA，请将 MFA 条件添加到您的策略中。有关更多信息，请参阅《IAM 用户指南》**中的[使用 MFA 保护 API 访问](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_mfa_configure-api-require.html)。 有关 IAM 中的最佳实操的更多信息，请参阅《IAM 用户指南》**中的 [IAM 中的安全最佳实践](https://docs.aws.amazon.com/IAM/latest/UserGuide/best-practices.html)。 # 允许用户查看他们自己的权限 该示例说明了您如何创建策略，以允许 IAM 用户查看附加到其用户身份的内联和托管式策略。此策略包括在控制台上或使用 AWS CLI 或 AWS API 以编程方式完成此操作的权限。 ``` { "Version": "2012-10-17", "Statement": [ { "Sid": "ViewOwnUserInfo", "Effect": "Allow", "Action": [ "iam:GetUserPolicy", "iam:ListGroupsForUser", "iam:ListAttachedUserPolicies", "iam:ListUserPolicies", "iam:GetUser" ], "Resource": ["arn:aws:iam::*:user/${aws:username}"] }, { "Sid": "NavigateInConsole", "Effect": "Allow", "Action": [ "iam:GetGroupPolicy", "iam:GetPolicyVersion", "iam:GetPolicy", "iam:ListAttachedGroupPolicies", "iam:ListGroupPolicies", "iam:ListPolicyVersions", "iam:ListPolicies", "iam:ListUsers" ], "Resource": "*" } ] } ``` # 访问一个 Amazon MSK 集群 在此示例中，您想要为 Amazon Web Services 账户中的 IAM 用户授予访问某个集群 `purchaseQueriesCluster` 的权限。此策略允许用户描述集群、获取其引导代理、列出其代理节点并更新它。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement":[ { "Sid":"UpdateCluster", "Effect":"Allow", "Action":[ "kafka:Describe*", "kafka:Get*", "kafka:List*", "kafka:Update*" ], "Resource":"arn:aws:kafka:us-east-1:012345678012:cluster/purchaseQueriesCluster/abcdefab-1234-abcd-5678-cdef0123ab01-2" } ] } ``` ------ # 根据标签访问 Amazon MSK 集群 您可以在基于身份的策略中使用条件，以便基于标签控制对 Amazon MSK 资源的访问权限。此示例演示了如何创建允许用户描述集群、获取其引导代理、列出其代理节点、更新和删除集群的策略。但是，仅当集群标签 `Owner` 的值为该用户的用户名时，才能授予此权限。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "AccessClusterIfOwner", "Effect": "Allow", "Action": [ "kafka:Describe*", "kafka:Get*", "kafka:List*", "kafka:Update*", "kafka:Delete*" ], "Resource": "arn:aws:kafka:us-east-1:012345678012:cluster/*", "Condition": { "StringEquals": { "aws:ResourceTag/Owner": "${aws:username}" } } } ] } ``` ------ 您可以将该策略附加到您账户中的 IAM 用户。如果名为 `richard-roe` 的用户尝试更新 MSK 集群，必须将集群标记为 `Owner=richard-roe` 或 `owner=richard-roe`。否则，他将被拒绝访问。条件标签键 `Owner` 匹配 `Owner` 和 `owner`，因为条件键名称不区分大小写。有关更多信息，请参阅 *IAM 用户指南* 中的 [IAM JSON 策略元素：条件](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition.html)。 # Amazon MSK 的服务相关角色 Amazon MSK 使用 AWS Identity and Access Management (IAM) [服务相关](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_terms-and-concepts.html#iam-term-service-linked-role)角色。服务相关角色是一种独特类型的 IAM 角色，它与 Amazon MSK 直接关联。服务相关角色由 Amazon MSK 预定义，包括该服务代表您调用其他 AWS 服务所需的所有权限。 服务相关角色可让您更轻松地设置 Amazon MSK，因为您不必手动添加所需权限。Amazon MSK 可定义其服务相关角色的权限。除非另行定义，否则只有 Amazon MSK 才能代入其角色。定义的权限包括信任策略和权限策略，而且权限策略不能附加到任何其他 IAM 实体。 有关支持服务相关角色的其他服务的信息，请参阅[使用 IAM 的 Amazon Web Services](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_aws-services-that-work-with-iam.html)，并查找**服务相关角色**列中显示为**是**的服务。请选择**是**与查看该服务的服务关联角色文档的链接。 **Topics** + [服务相关角色权限](slr-permissions.md) + [创建服务相关角色](create-slr.md) + [编辑服务相关角色](edit-slr.md) + [服务相关角色的受支持区域](slr-regions.md) # Amazon MSK 的服务相关角色权限 Amazon MSK 使用名为 **AWSServiceRoleForKafka** 的服务相关角色。Amazon MSK 使用此角色来访问您的资源并执行以下操作： + `*NetworkInterface` – 在客户账户中创建和管理网络接口，使客户 VPC 中的客户端可以访问集群代理。 + `*VpcEndpoints`— 管理客户账户中的 VPC 终端节点，这些终端节点允许客户 VPC 中的客户使用集群代理 AWS PrivateLink。Amazon MSK 对 `DescribeVpcEndpoints`、`ModifyVpcEndpoint` 和 `DeleteVpcEndpoints` 使用权限。 + `secretsmanager`— 使用管理客户凭证 AWS Secrets Manager。 + `GetCertificateAuthorityCertificate` – 检索私有证书颁发机构的证书。 + `*Ipv6Addresses`— 为客户账户中的网络接口分配和取消分配 IPv6 地址，以启用 MSK IPv6 群集的连接。 + `ModifyNetworkInterfaceAttribute`— 修改客户帐户中的网络接口属性以配置 MSK 集群连接的 IPv6 设置。 此服务相关角色附加到以下托管策略：`KafkaServiceRolePolicy`。有关此策略的更新，请参阅[KafkaServiceRolePolicy](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/KafkaServiceRolePolicy.html)。 AWSServiceRoleForKafka 服务相关角色信任以下服务代入该角色： + `kafka.amazonaws.com` 角色权限策略允许 Amazon MSK 对资源完成以下操作。 您必须配置权限，允许 IAM 实体（如用户、组或角色）创建、编辑或删除服务关联角色。有关更多信息，请参阅《IAM 用户指南》**中的[服务关联角色权限](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#service-linked-role-permissions)。 # 为 Amazon MSK 创建服务相关角色 您无需手动创建服务相关角色。当您在 AWS 管理控制台、或 AWS API 中创建 Amazon MSK 集群时 AWS CLI，Amazon MSK 会为您创建服务相关角色。 如果您删除该服务关联角色，然后需要再次创建，您可以使用相同流程在账户中重新创建此角色。当您创建 Amazon MSK 集群时，Amazon MSK 将再次为您创建服务相关角色。 # 编辑 Amazon MSK 的服务相关角色 Amazon MSK 不允许您编辑 AWSServiceRoleForKafka 服务相关角色。创建服务关联角色后，您将无法更改角色的名称，因为可能有多种实体引用该角色。但是可以使用 IAM 编辑角色描述。有关更多信息，请参阅《IAM 用户指南》**中的[编辑服务关联角色](https://docs.aws.amazon.com/IAM/latest/UserGuide/using-service-linked-roles.html#edit-service-linked-role)。 # Amazon MSK 服务相关角色支持的区域 Amazon MSK 支持在该服务可用的所有区域中使用服务相关角色。有关更多信息，请参阅[AWS 区域和端点](https://docs.aws.amazon.com/general/latest/gr/rande.html)。 # AWS 亚马逊 MSK 的托管策略 AWS 托管策略是由创建和管理的独立策略 AWS。 AWS 托管策略旨在为许多常见用例提供权限，以便您可以开始为用户、组和角色分配权限。 请记住， AWS 托管策略可能不会为您的特定用例授予最低权限权限，因为它们可供所有 AWS 客户使用。我们建议通过定义特定于使用案例的[客户管理型策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#customer-managed-policies)来进一步减少权限。 您无法更改 AWS 托管策略中定义的权限。如果 AWS 更新 AWS 托管策略中定义的权限，则更新会影响该策略所关联的所有委托人身份（用户、组和角色）。 AWS 最有可能在启动新的 API 或现有服务可以使用新 AWS 服务 的 API 操作时更新 AWS 托管策略。 有关更多信息，请参阅《IAM 用户指南》**中的 [AWS 托管式策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_managed-vs-inline.html#aws-managed-policies)。 # AWS 托管策略：Amazon A MSKFull ccess 此策略授予管理权限，允许主体完全访问所有 Amazon MSK 操作。此策略中的权限分组如下： + Amazon MSK 权限允许所有 Amazon MSK 操作。 + **`Amazon EC2` 权限** – 此策略中需要才能验证 API 请求中的已传递资源。这是为了确保 Amazon MSK 能够成功在集群中使用资源。此策略中的其他 Amazon EC2 权限允许 Amazon MSK 创建必要的 AWS 资源，使您能够连接到您的集群。 + **`AWS KMS` 权限** – API 调用期间用于验证请求中的已传递资源。Amazon MSK 必须使用它们才能在 Amazon MSK 集群中使用传递的密钥。 + **`CloudWatch Logs, Amazon S3, and Amazon Data Firehose` 权限** – Amazon MSK 需要才能确保可到达日志传输目标及这些目标对代理日志使用有效。 + **`IAM` 权限** – Amazon MSK 需要才能在您的账户中创建服务相关角色，并允许您将服务执行角色传递给 Amazon MSK。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [{ "Effect": "Allow", "Action": [ "kafka:*", "ec2:DescribeSubnets", "ec2:DescribeVpcs", "ec2:DescribeSecurityGroups", "ec2:DescribeRouteTables", "ec2:DescribeVpcEndpoints", "ec2:DescribeVpcAttribute", "kms:DescribeKey", "kms:CreateGrant", "logs:CreateLogDelivery", "logs:GetLogDelivery", "logs:UpdateLogDelivery", "logs:DeleteLogDelivery", "logs:ListLogDeliveries", "logs:PutResourcePolicy", "logs:DescribeResourcePolicies", "logs:DescribeLogGroups", "S3:GetBucketPolicy", "firehose:TagDeliveryStream" ], "Resource": "*" }, { "Effect": "Allow", "Action": [ "ec2:CreateVpcEndpoint" ], "Resource": [ "arn:*:ec2:*:*:vpc/*", "arn:*:ec2:*:*:subnet/*", "arn:*:ec2:*:*:security-group/*" ] }, { "Effect": "Allow", "Action": [ "ec2:CreateVpcEndpoint" ], "Resource": [ "arn:*:ec2:*:*:vpc-endpoint/*" ], "Condition": { "StringEquals": { "aws:RequestTag/AWSMSKManaged": "true" }, "StringLike": { "aws:RequestTag/ClusterArn": "*" } } }, { "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:*:ec2:*:*:vpc-endpoint/*", "Condition": { "StringEquals": { "ec2:CreateAction": "CreateVpcEndpoint" } } }, { "Effect": "Allow", "Action": [ "ec2:DeleteVpcEndpoints" ], "Resource": "arn:*:ec2:*:*:vpc-endpoint/*", "Condition": { "StringEquals": { "ec2:ResourceTag/AWSMSKManaged": "true" }, "StringLike": { "ec2:ResourceTag/ClusterArn": "*" } } }, { "Effect": "Allow", "Action": "iam:PassRole", "Resource": "*", "Condition": { "StringEquals": { "iam:PassedToService": "kafka.amazonaws.com" } } }, { "Effect": "Allow", "Action": "iam:CreateServiceLinkedRole", "Resource": "arn:aws:iam::*:role/aws-service-role/kafka.amazonaws.com/AWSServiceRoleForKafka*", "Condition": { "StringLike": { "iam:AWSServiceName": "kafka.amazonaws.com" } } }, { "Effect": "Allow", "Action": [ "iam:AttachRolePolicy", "iam:PutRolePolicy" ], "Resource": "arn:aws:iam::*:role/aws-service-role/kafka.amazonaws.com/AWSServiceRoleForKafka*" }, { "Effect": "Allow", "Action": "iam:CreateServiceLinkedRole", "Resource": "arn:aws:iam::*:role/aws-service-role/delivery.logs.amazonaws.com/AWSServiceRoleForLogDelivery*", "Condition": { "StringLike": { "iam:AWSServiceName": "delivery.logs.amazonaws.com" } } } ] } ``` ------ # AWS 托管策略：Amazon MSKRead OnlyAccess 此策略授予只读权限，允许用户查看 Amazon MSK 中的信息。附加此策略的主体不能进行任何更新或删除现有资源，也不能创建新的 Amazon MSK 资源。例如，拥有这些权限的主体可以查看与其账户关联的集群和配置列表，但不能更改任何集群的配置或设置。此策略中的权限分组如下： + **`Amazon MSK` 权限** – 允许您列出 Amazon MSK 资源、对它们进行描述并获取有关它们的信息。 + **`Amazon EC2`权限** — 用于描述与集群关联的 Amazon VPC、 ENIs 子网、安全组。 + **`AWS KMS` 权限** – 用于描述与集群关联的密钥。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Action": [ "kafka:Describe*", "kafka:List*", "kafka:Get*", "ec2:DescribeNetworkInterfaces", "ec2:DescribeSecurityGroups", "ec2:DescribeSubnets", "ec2:DescribeVpcs", "kms:DescribeKey" ], "Effect": "Allow", "Resource": "*" } ] } ``` ------ # AWS 托管策略： KafkaServiceRolePolicy 您无法附加 KafkaServiceRolePolicy 到您的 IAM 实体。将此策略附加到服务相关角色，该角色允许 Amazon MSK 执行诸如管理 MSK 集群上的 VPC 端点（连接器）、管理网络接口和使用 AWS Secrets Manager管理集群凭证等操作。有关更多信息，请参阅 [Amazon MSK 的服务相关角色](using-service-linked-roles.md)。 下表描述了自 Amazon MSK 开始跟踪变更以来对 KafkaServiceRolePolicy 托管策略的更新。 | 更改 | 描述 | 日期 | | --- | --- | --- | | IPv6 已将@@ [连接支持添加到 KafkaServiceRolePolicy](#security-iam-awsmanpol-KafkaServiceRolePolicy) — 更新现有政策 | Amazon MSK KafkaServiceRolePolicy 向添加了启用 MSK IPv6 集群连接的权限。这些权限允许 Amazon MSK 为网络接口分配和取消分配 IPv6 地址，以及修改客户账户中的网络接口属性。 | 2025 年 11 月 17 日 | | [KafkaServiceRolePolicy](#security-iam-awsmanpol-KafkaServiceRolePolicy)：对现有策略的更新 | Amazon MSK 添加了支持多 VPC 私有连接的权限。 | 2023 年 3 月 8 日 | | Amazon MSK 开启了跟踪更改 | Amazon MSK 已开始跟踪 KafkaServiceRolePolicy 托管政策的变更。 | 2023 年 3 月 8 日 | # AWS 托管策略： AWSMSKReplicatorExecutionRole `AWSMSKReplicatorExecutionRole` 策略向 Amazon MSK 复制器授予在 MSK 集群之间复制数据的权限。此策略中的权限如下分组： + **`cluster`** – 向 Amazon MSK 复制器授予使用 IAM 身份验证连接到集群的权限。还授予描述和更改集群的权限。 + **`topic`** – 向 Amazon MSK 复制器授予描述、创建和更改主题以及更改主题动态配置的权限。 + **`consumer group`** – 向 Amazon MSK 复制器授予描述和更改消费者组、从 MSK 集群读取和写入日期以及删除复制器创建的内部主题的权限。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ClusterPermissions", "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:DescribeCluster", "kafka-cluster:AlterCluster", "kafka-cluster:DescribeTopic", "kafka-cluster:CreateTopic", "kafka-cluster:AlterTopic", "kafka-cluster:WriteData", "kafka-cluster:ReadData", "kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup", "kafka-cluster:DescribeTopicDynamicConfiguration", "kafka-cluster:AlterTopicDynamicConfiguration", "kafka-cluster:WriteDataIdempotently" ], "Resource": [ "arn:aws:kafka:*:*:cluster/*" ] }, { "Sid": "TopicPermissions", "Effect": "Allow", "Action": [ "kafka-cluster:DescribeTopic", "kafka-cluster:CreateTopic", "kafka-cluster:AlterTopic", "kafka-cluster:WriteData", "kafka-cluster:ReadData", "kafka-cluster:DescribeTopicDynamicConfiguration", "kafka-cluster:AlterTopicDynamicConfiguration", "kafka-cluster:AlterCluster" ], "Resource": [ "arn:aws:kafka:*:*:topic/*/*" ] }, { "Sid": "GroupPermissions", "Effect": "Allow", "Action": [ "kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup" ], "Resource": [ "arn:aws:kafka:*:*:group/*/*" ] } ] } ``` ------ # 亚马逊 MSK 更新了托 AWS 管政策 查看自该服务开始跟踪这些更改以来，Amazon MSK AWS 托管政策更新的详细信息。 | 更改 | 描述 | 日期 | | --- | --- | --- | | [WriteDataIdempotently 权限已添加到 AWSMSKReplicator ExecutionRole](security-iam-awsmanpol-AWSMSKReplicatorExecutionRole.md)-更新现有策略 | Amazon MSK 为 AWSMSKReplicatorExecutionRole 策略添加了支持 MSK 集群之间数据复制的 WriteDataIdempotently 权限。 | 2024 年 3 月 12 日 | | [AWSMSKReplicatorExecutionRole](security-iam-awsmanpol-AWSMSKReplicatorExecutionRole.md)：新策略 | 亚马逊 MSK 增加了支持 Amazon MSK Replicator 的 AWSMSKReplicatorExecutionRole 政策。 | 2023 年 12 月 4 日 | | [Amazon A MSKFull cces](security-iam-awsmanpol-AmazonMSKFullAccess.md) s — 更新现有政策 | Amazon MSK 添加了支持 Amazon MSK 复制器的权限。 | 2023 年 9 月 28 日 | | [KafkaServiceRolePolicy](security-iam-awsmanpol-KafkaServiceRolePolicy.md)：对现有策略的更新 | Amazon MSK 添加了支持多 VPC 私有连接的权限。 | 2023 年 3 月 8 日 | | [Amazon A MSKFull cces](security-iam-awsmanpol-AmazonMSKFullAccess.md) s — 更新现有政策 | Amazon MSK 添加了新的 Amazon EC2 权限，以便连接到集群。 | 2021 年 11 月 30 日 | | [Amazon A MSKFull cces](security-iam-awsmanpol-AmazonMSKFullAccess.md) s — 更新现有政策 | Amazon MSK 添加了新权限，允许其描述 Amazon EC2 路由表。 | 2021 年 11 月 19 日 | | Amazon MSK 开启了跟踪更改 | Amazon MSK 开始跟踪其 AWS 托管政策的变更。 | 2021 年 11 月 19 日 | # 排查 Amazon MSK 身份和访问问题 使用以下信息可帮助您诊断和修复在使用 Amazon MSK 和 IAM 时可能遇到的常见问题。 **Topics** + [我无权在 Amazon MSK 中执行操作](#security_iam_troubleshoot-no-permissions) ## 我无权在 Amazon MSK 中执行操作 如果 AWS 管理控制台 告诉您您无权执行某项操作，则必须联系管理员寻求帮助。管理员是向您提供登录凭证的人。 当 `mateojackson` IAM 用户尝试使用控制台删除集群，但没有 `kafka:DeleteCluster` 权限时，会发生以下示例错误。 ``` User: arn:aws:iam::123456789012:user/mateojackson is not authorized to perform: kafka:DeleteCluster on resource: purchaseQueriesCluster ``` 在这种情况下，Mateo 请求他的管理员更新其策略，以允许他使用 `kafka:DeleteCluster` 操作访问 `purchaseQueriesCluster` 资源。 # Apache Kafka 的身份验证和授权 APIs 您可以使用 IAM 对客户端进行身份验证并允许或拒绝 Apache Kafka 操作。或者，您可以使用 TLS 或对客户端 SASL/SCRAM 进行身份验证，使用 Apache Kafka ACLs 来允许或拒绝操作。 有关如何控制谁可以在您的集群上执行 [Amazon MSK 操作](https://docs.aws.amazon.com/msk/1.0/apireference/operations.html)的信息，请参阅[亚马逊 MSK 的身份验证和授权 APIs](security-iam.md)。 **Topics** + [IAM 访问控制](iam-access-control.md) + [Amazon MSK 的双向 TLS 客户端身份验证](msk-authentication.md) + [使用 S AWS ecrets Manager 进行登录凭据身份验证](msk-password.md) + [阿帕奇 Kafka ACLs](msk-acls.md) # IAM 访问控制 Amazon MSK 的 IAM 访问控制让您能够处理 MSK 集群的身份验证和授权。这样就不需要使用一种身份验证机制和另一种授权机制。例如，当客户端尝试写入您的集群时，Amazon MSK 使用 IAM 来检查该客户端是否是经过身份验证的身份，以及是否有权向您的集群生成数据。 IAM 访问控制适用于 Java 和非 Java 客户端，包括用 Python、 JavaScript Go 和.NET 编写的 Kafka 客户端。非 Java 客户端的 IAM 访问控制适用于 Kafka 版本 2.7.1 或更高版本的 MSK 集群。 为了能够进行 IAM 访问控制，Amazon MSK 对 Apache Kafka 源代码进行了少许修改。这些修改不会给您的 Apache Kafka 体验造成明显的影响。Amazon MSK 会记录访问事件，以方便您进行审计。 您可以 APIs 为使用 IAM 访问控制的 MSK 集群调用 Apache Kafka ACL。但是，Apache Kafka 对 IAM 身份的授权 ACLs 没有影响。您必须将 IAM 策略用于 IAM 身份的访问控制。 **重要注意事项** 对 MSK 集群使用 IAM 访问控制时，请记住以下重要注意事项： IAM 访问控制不适用于 Apache ZooKeeper 节点。有关如何控制对这些节点的访问权限的信息，请参阅 [控制对亚马逊 MSK ZooKeeper 集群中 Apache 节点的访问权限](zookeeper-security.md)。 如果您的集群使用 IAM 访问控制，则 `allow.everyone.if.no.acl.found` Apache Kafka 设置无效。 您可以 APIs 为使用 IAM 访问控制的 MSK 集群调用 Apache Kafka ACL。但是，Apache Kafka 对 IAM 身份的授权 ACLs 没有影响。您必须将 IAM 策略用于 IAM 身份的访问控制。 # Amazon MSK 的 IAM 访问控制的工作原理 要对 Amazon MSK 使用 IAM 访问控制，请执行以下步骤，以下主题将详细介绍这些步骤： + [创建使用 IAM 访问控制的 Amazon MSK 集群](create-iam-access-control-cluster-in-console.md) + [配置客户端以进行 IAM 访问控制](configure-clients-for-iam-access-control.md) + [为 IAM 角色创建授权策略](create-iam-access-control-policies.md) + [获取用于 IAM 访问控制的引导代理](get-bootstrap-brokers-for-iam.md) # 创建使用 IAM 访问控制的 Amazon MSK 集群 本节介绍如何使用 AWS 管理控制台、API 或创建使用 IAM 访问控制的 Amazon MSK 集群。 AWS CLI 有关如何为现有集群开启 IAM 访问控制的信息，请参阅[更新 Amazon MSK 集群的安全设置](msk-update-security.md)。 **使用创建 AWS 管理控制台 使用 IAM 访问控制的集群** 1. 在 [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/) 打开 Amazon MSK 控制台。 1. 选择**创建集群**。 1. 选择**使用自定义设置创建集群**。 1. 在**身份验证**部分中，选择 **IAM 访问控制**。 1. 完成创建集群的其余工作流程。 **使用 API 或创建 AWS CLI 使用 IAM 访问控制的集群** + 要创建启用了 IAM 访问控制的集群，请使用 [CreateCluster](https://docs.aws.amazon.com/msk/1.0/apireference/clusters.html#CreateCluster)API 或 [create-cluster CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kafka/create-cluster.html) 命令，并传递以下 JSON 作为`ClientAuthentication`参数：。`"ClientAuthentication": { "Sasl": { "Iam": { "Enabled": true } }` # 配置客户端以进行 IAM 访问控制 要让客户端能够与使用 IAM 访问控制的 MSK 集群通信，您可以使用以下任何一种机制： + 使用 SASL\$1OAUTHBEARER机制进行非 Java 客户端配置 + 使用 SASL\$1OAUTHBEARER机制或 AWS\$1MSK\$1IAM 机制配置 Java 客户端 ## 使用该 SASL\$1OAUTHBEARER机制配置 IAM 1. 使用以下 Python Kafka 客户端示例，编辑 client.properties 配置文件。其他语言的配置更改与之类似。 ``` from kafka import KafkaProducer from kafka.errors import KafkaError from kafka.sasl.oauth import AbstractTokenProvider import socket import time from aws_msk_iam_sasl_signer import MSKAuthTokenProvider class MSKTokenProvider(): def token(self): token, _ = MSKAuthTokenProvider.generate_auth_token('') return token tp = MSKTokenProvider() producer = KafkaProducer( bootstrap_servers='', security_protocol='SASL_SSL', sasl_mechanism='OAUTHBEARER', sasl_oauth_token_provider=tp, client_id=socket.gethostname(), ) topic = "" while True: try: inp=input(">") producer.send(topic, inp.encode()) producer.flush() print("Produced!") except Exception: print("Failed to send message:", e) producer.close() ``` 1. 下载所选配置语言的帮助程序库，然后按照该语言库主页的“*开始使用*”部分中的说明进行操作。 + JavaScript: [https://github.com/aws/aws-msk-iam-sasl-signer-js \$1getting-star](https://github.com/aws/aws-msk-iam-sasl-signer-js#getting-started) ted + Python：[https://github.com/aws/aws-msk-iam-sasl-signer-python \$1get-](https://github.com/aws/aws-msk-iam-sasl-signer-python#get-started) 已启动 + Go: [https://github.com/aws/aws-msk-iam-sasl-signer-](https://github.com/aws/aws-msk-iam-sasl-signer-go#getting-started) go \$1getting-started + .NET: [https://github.com/aws/aws-msk-iam-sasl-signer-net \$1getting-](https://github.com/aws/aws-msk-iam-sasl-signer-net#getting-started) 已启动 + JAVA：可通过 ja [https://github.com/aws/aws-msk-iam-auth/releases](https://github.com/aws/aws-msk-iam-auth/releases)r 文件获得对 Java 的 SASL\$1OAUTHBEARER支持 ## 使用 MSK 自定义 AWS\$1MSK\$1IAM 机制配置 IAM 1. 将以下内容添加到 `client.properties` 文件中。**替换为客户端上信任存储文件的完全限定路径。 **注意** 如果您不想使用特定证书，可以从 `client.properties` 文件中删除 `ssl.truststore.location=`。如果您不指定 `ssl.truststore.location` 的值，Java 进程将使用默认证书。 ``` ssl.truststore.location= security.protocol=SASL_SSL sasl.mechanism=AWS_MSK_IAM sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required; sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler ``` 要使用您为 AWS 凭证创建的命名配置文件，请将其包含`awsProfileName="your profile name";`在您的客户端配置文件中。有关命名配置文件的信息，请参阅文档中的[命名配置](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-profiles.html) AWS CLI 文件。 1. 下载最新的稳定版 [aws-msk-iam-auth](https://github.com/aws/aws-msk-iam-auth/releases)JAR 文件，并将其放在类路径中。如果您使用 Maven，请添加以下依赖项，并根据需要调整版本号： ``` software.amazon.msk aws-msk-iam-auth 1.0.0 ``` Amazon MSK 客户端插件在 Apache 2.0 许可证下是开源的。 # 为 IAM 角色创建授权策略 将授权策略附加到与客户端对应的 IAM 角色。在授权策略中，您可以指定角色允许或拒绝哪些操作。如果您的客户端位于 Amazon EC2 实例上，请将授权策略与该 Amazon EC2 实例的 IAM 角色关联。或者，您也可以将客户端配置为使用命名配置文件，然后将授权策略与该命名配置文件的角色关联。[配置客户端以进行 IAM 访问控制](configure-clients-for-iam-access-control.md) 介绍如何将客户端配置为使用命名配置文件。 有关如何创建 IAM policy 的信息，请参阅[创建 IAM policy](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_create.html)。 以下是名为的集群的授权策略示例 MyTestCluster。要了解 `Action` 和 `Resource` 元素的语义，请参阅 [IAM 授权策略、操作和资源的语义](kafka-actions.md)。 **重要** 您对 IAM 策略所做的更改会 AWS CLI 立即反映在 IAM APIs 中。但是，策略更改可能需要很长时间才能生效。在大多数情况下，策略更改会在不到一分钟的时间内生效。网络状况有时可能会增加延迟时间。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:AlterCluster", "kafka-cluster:DescribeCluster" ], "Resource": [ "arn:aws:kafka:us-east-1:111122223333:cluster/MyTestCluster/abcd1234-0123-abcd-5678-1234abcd-1" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:*Topic*", "kafka-cluster:WriteData", "kafka-cluster:ReadData" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/MyTestCluster/*" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:group/MyTestCluster/*" ] } ] } ``` ------ 要了解如何创建包含与常见 Apache Kafka 用例（例如创建和使用数据）对应的操作元素的策略，请参阅[客户端授权策略的常见使用场景](iam-access-control-use-cases.md)。 [对于 Kafka 版本 2.8.0 及更高版本，该**WriteDataIdempotently**权限已被弃用 (KIP-679)。](https://cwiki.apache.org/confluence/display/KAFKA/KIP-679%3A+Producer+will+enable+the+strongest+delivery+guarantee+by+default)默认情况下，设置了 `enable.idempotence = true`。因此，对于 Kafka 版本 2.8.0 及更高版本，IAM 不提供与 Kafka 相同的功能。 ACLs仅提供 `WriteData` 对某个主题的访问权限，无法 `WriteDataIdempotently` 到该主题。这不会影响将 `WriteData` 提供给**所有**主题的情况。在这种情况下，允许 `WriteDataIdempotently`。这是由于 IAM 逻辑的实现和 Kafka ACLs 的实现方式存在差异。此外，以幂等方式写入一个主题也需要访问 `transactional-ids`。 要解决这个问题，建议使用类似于以下策略的策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:AlterCluster", "kafka-cluster:DescribeCluster", "kafka-cluster:WriteDataIdempotently" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:cluster/MyTestCluster/abcd1234-0123-abcd-5678-1234abcd-1" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:*Topic*", "kafka-cluster:WriteData", "kafka-cluster:ReadData" ], "Resource": [ "arn:aws:kafka:us-east-1:123456789012:topic/MyTestCluster/abcd1234-0123-abcd-5678-1234abcd-1/TestTopic", "arn:aws:kafka:us-east-1:123456789012:transactional-id/MyTestCluster/abcd1234-0123-abcd-5678-1234abcd-1/*" ] } ] } ``` ------ 在这种情况下，`WriteData` 允许写入 `TestTopic`，而 `WriteDataIdempotently` 允许对集群进行幂等性写入。该策略还增加了对未来所需的 `transactional-id` 资源的访问权限。 由于 `WriteDataIdempotently` 是集群级权限，因此无法主题级别使用。如果 `WriteDataIdempotently` 仅限于主题级别，则此策略将不起作用。 # 获取用于 IAM 访问控制的引导代理 请参阅[获取 Amazon MSK 集群的引导代理](msk-get-bootstrap-brokers.md)。 # IAM 授权策略、操作和资源的语义 **注意** 对于运行 Apache Kafka 3.8 或更高版本的集群，IAM 访问控制支持用于终止 WriteTxnMarkers 交易的 API。对于运行 Kafka 版本低于 3.8 的集群，IAM 访问控制不支持内部集群操作，包括 WriteTxnMarkers。对于这些早期版本，要终止交易，请使用带有相应的 SCRAM 或 mTLS 身份验证， ACLs 而不是 IAM 身份验证。 本部分解释了可以在 IAM 授权策略中使用的操作和资源元素的语义。有关策略示例，请参阅 [为 IAM 角色创建授权策略](create-iam-access-control-policies.md)。 ## 授权策略操作 下表列出了在使用 Amazon MSK 的 IAM 访问控制时可以在授权策略中包含的操作。当您在授权策略中包含表*操作*列中的操作时，还必须包含*所需操作*列中的相应操作。 | Action | 说明 | 所需的操作 | 所需的资源 | 适用于无服务器集群 | | --- | --- | --- | --- | --- | | kafka-cluster:Connect | 授予连接和验证集群的权限。 | 无 | cluster | 是 | | kafka-cluster:DescribeCluster | 授予描述集群各个方面的权限，相当于 Apache Kafka 的 DESCRIBE CLUSTER ACL。 | `kafka-cluster:Connect` | cluster | 是 | | kafka-cluster:AlterCluster | 授予更改集群各个方面的权限，相当于 Apache Kafka 的 ALTER CLUSTER ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeCluster` | cluster | 否 | | kafka-cluster:DescribeClusterDynamicConfiguration | 授予描述集群动态配置的权限，相当于 Apache Kafka 的 DESCRIBE\$1CONFIGS CLUSTER ACL。 | `kafka-cluster:Connect` | cluster | 否 | | kafka-cluster:AlterClusterDynamicConfiguration | 授予更改集群动态配置的权限，相当于 Apache Kafka 的 ALTER\$1CONFIGS CLUSTER ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeClusterDynamicConfiguration` | cluster | 否 | | kafka-cluster:WriteDataIdempotently | 授予在集群上以幂等方式写入数据的权限，相当于 Apache Kafka 的 IDEMPOTENT\$1WRITE CLUSTER ACL。 | `kafka-cluster:Connect` `kafka-cluster:WriteData` | cluster | 是 | | kafka-cluster:CreateTopic | 授予在集群上创建主题的权限，相当于 Apache Kafka 的创建 CLUSTER/TOPIC ACL。 | `kafka-cluster:Connect` | topic | 是 | | kafka-cluster:DescribeTopic | 授予描述集群上主题的权限，相当于 Apache Kafka 的 DESCRIBE TOPIC ACL。 | `kafka-cluster:Connect` | topic | 是 | | kafka-cluster:AlterTopic | 授予更改集群上主题的权限，相当于 Apache Kafka 的 ALTER TOPIC ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` | topic | 是 | | kafka-cluster:DeleteTopic | 授予删除集群上主题的权限，相当于 Apache Kafka 的 DELETE TOPIC ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` | topic | 是 | | kafka-cluster:DescribeTopicDynamicConfiguration | 授予描述集群上主题动态配置的权限，相当于 Apache Kafka 的 DESCRIBE\$1CONFIGS TOPIC ACL。 | `kafka-cluster:Connect` | topic | 是 | | kafka-cluster:AlterTopicDynamicConfiguration | 授予更改集群上主题动态配置的权限，相当于 Apache Kafka 的 ALTER\$1CONFIGS TOPIC ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopicDynamicConfiguration` | topic | 是 | | kafka-cluster:ReadData | 授予从集群上主题中读取数据的权限，相当于 Apache Kafka 的 READ TOPIC ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:AlterGroup` | topic | 是 | | kafka-cluster:WriteData | 授予向集群上的主题写入数据的权限，相当于 Apache Kafka 的 WRITE TOPIC ACL | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` | topic | 是 | | kafka-cluster:DescribeGroup | 授予描述集群上群组的权限，相当于 Apache Kafka 的 DESCRIBE GROUP ACL。 | `kafka-cluster:Connect` | 组 | 是 | | kafka-cluster:AlterGroup | 授予加入集群上群组的权限，相当于 Apache Kafka 的 READ GROUP ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeGroup` | 组 | 是 | | kafka-cluster:DeleteGroup | 授予删除集群上群组的权限，相当于 Apache Kafka 的 DELETE GROUP ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeGroup` | 组 | 是 | | kafka-cluster:DescribeTransactionalId | 授予描述集群 IDs 上交易的权限，相当于 Apache Kafka 的 DESCRIBE TRANSACTIONAL\$1ID ACL。 | `kafka-cluster:Connect` | transactional-id | 是 | | kafka-cluster:AlterTransactionalId | 授予在集群 IDs 上更改事务的权限，相当于 Apache Kafka 的 WRITE TRANSACTIONAL\$1ID ACL。 | `kafka-cluster:Connect` `kafka-cluster:DescribeTransactionalId` `kafka-cluster:WriteData` | transactional-id | 是 | 在冒号之后的操作中，您可以任意次数地使用星号 (\$1) 通配符。示例如下。 + `kafka-cluster:*Topic` 代表 `kafka-cluster:CreateTopic`、`kafka-cluster:DescribeTopic`、`kafka-cluster:AlterTopic` 和 `kafka-cluster:DeleteTopic`。它不包括 `kafka-cluster:DescribeTopicDynamicConfiguration` 或 `kafka-cluster:AlterTopicDynamicConfiguration`。 + `kafka-cluster:*` 代表所有权限。 ## 授权策略资源 下表显示了在使用 Amazon MSK 的 IAM 访问控制时可在授权策略中使用的四种资源。您可以使用 [DescribeCluster](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn.html#DescribeCluster)API 或 d [es](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kafka/describe-cluster.html) AWS CLI cribe-cluster 命令从 AWS 管理控制台 或中获取集群 Amazon 资源名称 (ARN)。然后，您可以使用集群 ARN 来构造主题、群组和交易 ID。 ARNs要在授权策略中指定资源，请使用该资源的 ARN。 | 资源 | ARN 格式 | | --- | --- | | Cluster | arn: aws: kafka::: cluster//regionaccount-idcluster-namecluster-uuid | | Topic | arn: aws: kafka::: topic//regionaccount-idcluster-namecluster-uuidtopic-name | | Group | arn: aws: kafka::: group//regionaccount-idcluster-namecluster-uuidgroup-name | | 事务 ID | arn: aws: kafka::: transactional-id///regionaccount-idcluster-namecluster-uuidtransactional-id | 您可以在 ARN 中 `:cluster/`、`:topic/`、`:group/` 和 `:transactional-id/` 之后的任意位置，任意次数地使用星号 (\$1) 通配符。以下是如何使用星号 (\$1) 通配符引用多个资源的部分示例： + `arn:aws:kafka:us-east-1:0123456789012:topic/MyTestCluster/*`：任何名为的集群中的所有主题 MyTestCluster，无论集群的 UUID 如何。 + `arn:aws:kafka:us-east-1:0123456789012:topic/MyTestCluster/abcd1234-0123-abcd-5678-1234abcd-1/*_test`：集群中名称以 “\$1test” 结尾的所有主题，其名称为，UUID 为 abcd1234-0123-abcd-56 MyTestCluster 78-1234abcd-1。 + `arn:aws:kafka:us-east-1:0123456789012:transactional-id/MyTestCluster/*/5555abcd-1111-abcd-1234-abcd1234-1`：所有交易 ID 为 5555abcd-1111-abcd-1234-abcd-1234-1 的交易，适用于您的账户中命名的集群的所有化身。 MyTestCluster 这意味着，如果您创建了一个名为 MyTestCluster的集群，然后将其删除，然后创建另一个同名集群，则可以使用此资源 ARN 在两个集群上表示相同的交易 ID。但是，无法访问已删除的集群。 # 客户端授权策略的常见使用场景 下表中的第一列显示了一些常见用例。要授权客户端执行给定用例，请在客户端的授权策略中包含该用例所需的操作，并将 `Effect` 设置为 `Allow`。 有关 Amazon MSK 的 IAM 访问控制之所有操作的信息，请参阅 [IAM 授权策略、操作和资源的语义](kafka-actions.md)。 **注意** 默认情况下，操作将被拒绝。您必须明确允许要授权客户端执行的每个操作。 **** | 使用案例 | 所需的操作 | | --- | --- | | Admin | `kafka-cluster:*` | | 创建话题 | `kafka-cluster:Connect` `kafka-cluster:CreateTopic` | | 生成数据 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:WriteData` | | 使用数据 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:DescribeGroup` `kafka-cluster:AlterGroup` `kafka-cluster:ReadData` | | 以幂等方式生成数据 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:WriteData` `kafka-cluster:WriteDataIdempotently` | | 以事务方式生成数据 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:WriteData` `kafka-cluster:DescribeTransactionalId` `kafka-cluster:AlterTransactionalId` | | 描述集群的配置 | `kafka-cluster:Connect` `kafka-cluster:DescribeClusterDynamicConfiguration` | | 更新集群的配置 | `kafka-cluster:Connect` `kafka-cluster:DescribeClusterDynamicConfiguration` `kafka-cluster:AlterClusterDynamicConfiguration` | | 描述主题的配置 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopicDynamicConfiguration` | | 更新主题的配置 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopicDynamicConfiguration` `kafka-cluster:AlterTopicDynamicConfiguration` | | 更改主题 | `kafka-cluster:Connect` `kafka-cluster:DescribeTopic` `kafka-cluster:AlterTopic` | # Amazon MSK 的双向 TLS 客户端身份验证 可以为从应用程序到 Amazon MSK 代理的连接启用 TLS 客户端身份验证。要使用客户端身份验证，您需要有 AWS 私有 CA。 AWS 私有 CA 可以与您的集群位于 AWS 账户 同一个账户中，也可以位于不同的账户中。有关 AWS 私有 CA s 的信息，请参阅[创建和管理 AWS 私有 CA](https://docs.aws.amazon.com/acm-pca/latest/userguide/create-CA.html)。 Amazon MSK 不支持证书吊销列表 () CRLs。要控制对集群主题的访问权限或屏蔽已泄露的证书，请使用 Apache Kafka ACLs 和 AWS 安全组。有关使用 Apache Kafka 的信息 ACLs，请参阅。[阿帕奇 Kafka ACLs](msk-acls.md) **Topics** + [创建支持客户端身份验证的 Amazon MSK 集群](msk-authentication-cluster.md) + [将客户端设置为使用身份验证](msk-authentication-client.md) + [使用身份验证生成和使用消息](msk-authentication-messages.md) # 创建支持客户端身份验证的 Amazon MSK 集群 此过程向您展示如何使用启用客户端身份验证 AWS 私有 CA。 **注意** 在使用双向 TLS 控制访问时，我们强烈建议 AWS 私有 CA 对每个 MSK 集群使用独立模式。这样做可以确保由签名的 TLS 证书 PCAs 仅在单个 MSK 集群中进行身份验证。 1. 使用以下内容创建名为 `clientauthinfo.json` 的文件。*Private-CA-ARN*替换为您的 PCA 的 ARN。 ``` { "Tls": { "CertificateAuthorityArnList": ["Private-CA-ARN"] } } ``` 1. 创建一个名为 `brokernodegroupinfo.json` 的文件，如[使用创建预配置的 Amazon MSK 集群 AWS CLI](create-cluster-cli.md)中所述。 1. 客户端身份验证还要求您启用客户端和代理之间的传输中加密。使用以下内容创建名为 `encryptioninfo.json` 的文件。*KMS-Key-ARN*替换为您的 KMS 密钥的 ARN。可以将 `ClientBroker` 设置为 `TLS` 或 `TLS_PLAINTEXT`。 ``` { "EncryptionAtRest": { "DataVolumeKMSKeyId": "KMS-Key-ARN" }, "EncryptionInTransit": { "InCluster": true, "ClientBroker": "TLS" } } ``` 有关加密的更多信息，请参阅[Amazon MSK 加密](msk-encryption.md)。 1. 在 AWS CLI 安装了身份验证和传输中加密的计算机上，运行以下命令来创建集群。保存响应中提供的集群 ARN。 ``` aws kafka create-cluster --cluster-name "AuthenticationTest" --broker-node-group-info file://brokernodegroupinfo.json --encryption-info file://encryptioninfo.json --client-authentication file://clientauthinfo.json --kafka-version "{YOUR KAFKA VERSION}" --number-of-broker-nodes 3 ``` # 将客户端设置为使用身份验证 此过程描述了如何设置 Amazon EC2 实例以用作客户端来使用身份验证。 此过程介绍了如何通过创建客户端计算机、创建主题和配置所需的安全设置，使用身份验证来生成和使用消息。 1. 创建用作客户端计算机的 Amazon EC2 实例。为简单起见，请在用于集群的同一 VPC 中创建此实例。有关如何创建此类客户端计算机的示例，请参阅[步骤 3：创建客户端计算机](create-client-machine.md)。 1. 创建主题。有关示例，请参阅[步骤 4：在 Amazon MSK 集群中创建主题](create-topic.md)下的说明。 1. 在已 AWS CLI 安装的计算机上，运行以下命令以获取集群的引导代理。*Cluster-ARN*替换为集群的 ARN。 ``` aws kafka get-bootstrap-brokers --cluster-arn Cluster-ARN ``` 保存与响应中的 `BootstrapBrokerStringTls` 关联的字符串。 1. 在客户端计算机上，运行以下命令以使用 JVM 信任存储来创建客户端信任存储。如果您的 JVM 路径不同，请相应地调整命令。 ``` cp /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64/jre/lib/security/cacerts kafka.client.truststore.jks ``` 1. 在客户端计算机上，运行以下命令为客户端创建私有密钥。用您选择*Your-Key-Pass*的字符串替换*Distinguished-Name**Example-Alias**Your-Store-Pass*、、和。 ``` keytool -genkey -keystore kafka.client.keystore.jks -validity 300 -storepass Your-Store-Pass -keypass Your-Key-Pass -dname "CN=Distinguished-Name" -alias Example-Alias -storetype pkcs12 -keyalg rsa ``` 1. 在客户端计算机上，运行以下命令以使用您在上一步中创建的私有密钥创建证书请求。 ``` keytool -keystore kafka.client.keystore.jks -certreq -file client-cert-sign-request -alias Example-Alias -storepass Your-Store-Pass -keypass Your-Key-Pass ``` 1. 打开 `client-cert-sign-request` 文件，并确保该文件的开头为 `-----BEGIN CERTIFICATE REQUEST-----` 且结尾为 `-----END CERTIFICATE REQUEST-----`。如果该文件的开头为 `-----BEGIN NEW CERTIFICATE REQUEST-----`，请从文件的开头和结尾处删除单词 `NEW`（及其后面的单个空格）。 1. 在已 AWS CLI 安装证书的计算机上，运行以下命令对证书请求进行签名。*Private-CA-ARN*替换为您的 PCA 的 ARN。如果需要，您可以更改有效性值。在这里，我们以 300 为例。 ``` aws acm-pca issue-certificate --certificate-authority-arn Private-CA-ARN --csr fileb://client-cert-sign-request --signing-algorithm "SHA256WITHRSA" --validity Value=300,Type="DAYS" ``` 保存响应中提供的证书 ARN。 **注意** 要检索您的客户端证书，请使用 `acm-pca get-certificate` 命令并指定您的证书 ARN。有关更多信息，请参阅《AWS CLI Command Reference》**中的 [get-certificate](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/acm-pca/get-certificate.html)。 1. 运行以下命令获取为您 AWS 私有 CA 签名的证书。*Certificate-ARN*替换为您从对上一个命令的响应中获得的 ARN。 ``` aws acm-pca get-certificate --certificate-authority-arn Private-CA-ARN --certificate-arn Certificate-ARN ``` 1. 从运行上一条命令所获得的 JSON 结果中，复制与 `Certificate` 和 `CertificateChain` 关联的字符串。将这两个字符串粘贴到名为的新文件中 signed-certificate-from-acm。先粘贴与 `Certificate` 关联的字符串，然后粘贴与 `CertificateChain` 关联的字符串。将 `\n` 字符替换为新行。以下是将证书和证书链粘贴到其中之后的文件结构。 ``` -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- ``` 1. 在客户端计算机上运行以下命令将此证书添加到您的密钥库中，以便能在与 MSK 代理交流时出示此证书。 ``` keytool -keystore kafka.client.keystore.jks -import -file signed-certificate-from-acm -alias Example-Alias -storepass Your-Store-Pass -keypass Your-Key-Pass ``` 1. 使用以下内容创建名为 `client.properties` 的文件。将信任存储和密钥库位置调整为您将 `kafka.client.truststore.jks` 保存到的路径。用你的 Kafka 客户端版本代替*\$1YOUR KAFKA VERSION\$1*占位符。 ``` security.protocol=SSL ssl.truststore.location=/tmp/kafka_2.12-{YOUR KAFKA VERSION}/kafka.client.truststore.jks ssl.keystore.location=/tmp/kafka_2.12-{YOUR KAFKA VERSION}/kafka.client.keystore.jks ssl.keystore.password=Your-Store-Pass ssl.key.password=Your-Key-Pass ``` # 使用身份验证生成和使用消息 此过程介绍了如何使用身份验证生成和使用消息。 1. 运行以下命令以创建主题。名为 `client.properties` 的文件是您在上一过程中创建的文件。 ``` /bin/kafka-topics.sh --create --bootstrap-server BootstrapBroker-String --replication-factor 3 --partitions 1 --topic ExampleTopic --command-config client.properties ``` 1. 运行以下命令以启动控制台生成器。名为 `client.properties` 的文件是您在上一过程中创建的文件。 ``` /bin/kafka-console-producer.sh --bootstrap-server BootstrapBroker-String --topic ExampleTopic --producer.config client.properties ``` 1. 在客户端计算机上的新命令窗口中，运行以下命令以启动控制台使用器。 ``` /bin/kafka-console-consumer.sh --bootstrap-server BootstrapBroker-String --topic ExampleTopic --consumer.config client.properties ``` 1. 在生成器窗口中键入消息，并观察消息显示在使用器窗口中。 # 使用 S AWS ecrets Manager 进行登录凭据身份验证 您可以使用使用 S AWS ecrets Manager 存储和保护的登录凭证来控制对您的 Amazon MSK 集群的访问权限。将用户凭证存储在 Secrets Manager 中可以减少集群身份验证的开销，例如审计、更新和轮换凭证。Secrets Manager 还让您能够跨集群共享用户凭证。 将密钥与 MSK 集群关联后，MSK 会定期同步该凭证数据。 **Topics** + [登录凭证身份验证的工作原理](msk-password-howitworks.md) + [为 Amazon MSK 集群设置 SASL/SCRAM 身份验证](msk-password-tutorial.md) + [使用用户](msk-password-users.md) + [使用 SCRAM 密钥时的限制](msk-password-limitations.md) # 登录凭证身份验证的工作原理 Amazon MSK 的登录凭证身份验证使用的身份验证 SASL/SCRAM （简单身份验证和安全层/加盐质询响应机制）身份验证。要为集群设置登录凭证身份验证，您可以在 [AWS Secrets Manager](https://docs.aws.amazon.com//secretsmanager/?id=docs_gateway) 中创建密钥资源，并将登录凭证与该密钥关联。 SASL/SCRAM 在 [RFC 5802](https://tools.ietf.org/html/rfc5802) 中定义。SCRAM 使用安全哈希算法，不会在客户端和服务器之间传输明文登录凭证。 **注意** 当您为集群设置 SASL/SCRAM 身份验证时，Amazon MSK 会为客户端和代理之间的所有流量启用 TLS 加密。 # 为 Amazon MSK 集群设置 SASL/SCRAM 身份验证 要在 Secr AWS ets Manager 中设置密钥，请按照 Secrets Man [ager 用户指南中的[创建和检索密](https://docs.aws.amazon.com/secretsmanager/latest/userguide/tutorials_basic.html)AWS 钥](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html)教程进行操作。 在为 Amazon MSK 集群创建密钥时，请注意以下要求： + 对于密钥类型，请选择**其他密钥类型（例如 API 密钥）**。 + 您的密钥名称必须以前缀 **AmazonMSK\$1** 开头。 + 您必须使用现有的自定义 AWS KMS 密钥或为您的密 AWS KMS 钥创建新的自定义密钥。默认情况下，Secrets Manager 对密 AWS KMS 钥使用默认密钥。 **重要** 使用默认密钥创建的密 AWS KMS 钥不能用于 Amazon MSK 集群。 + 您的登录凭证数据必须采用以下格式，才能使用**明文**选项输入键值对。 ``` { "username": "alice", "password": "alice-secret" } ``` + 记录密钥的 ARN（Amazon 资源名称）值。 + **重要** 您不能将 Secrets Manager 密钥与超出 [调整集群的大小：每个标准代理的分区数量](bestpractices.md#partitions-per-broker) 中所述限制的集群关联。 + 如果您使用创建密钥，请为参数指定密钥 ID 或 ARN。 AWS CLI `kms-key-id`不要指定别名。 + 要将密钥与您的集群关联，请使用 Amazon MSK 控制台或[ BatchAssociateScramSecret](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn-scram-secrets.html#BatchAssociateScramSecret)操作。 **重要** 当您将密钥与集群关联时，Amazon MSK 会向该密钥附加资源策略，以允许您的集群访问和读取您定义的密钥值。您不应修改此资源策略。这样做可能会阻止您的集群访问密钥。如果对密钥资源策略和/或用于密钥加密的 KMS 密钥进行了任何更改，请确保将密钥重新关联至 MSK 集群。这可以确保集群能继续访问密钥。 `BatchAssociateScramSecret` 操作的以下 JSON 输入示例将密钥与集群关联： ``` { "clusterArn" : "arn:aws:kafka:us-west-2:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4", "secretArnList": [ "arn:aws:secretsmanager:us-west-2:0123456789019:secret:AmazonMSK_MyClusterSecret" ] } ``` # 使用登录凭证连接到集群 在创建密钥并将其与集群关联后，您便可以将客户端连接到集群。以下过程演示如何将客户端连接到使用 SASL/SCRAM 身份验证的集群。此外还展示了如何通过示例主题生成和使用。 **Topics** + [使用 SASL/SCRAM 身份验证将客户端连接到集群](#w2aab9c13c29c17c13c11b9b7) + [排除连接问题](#msk-password-tutorial-connect-troubleshooting) ## 使用 SASL/SCRAM 身份验证将客户端连接到集群 1. 在已 AWS CLI 安装的计算机上运行以下命令。*clusterARN*替换为集群的 ARN。 ``` aws kafka get-bootstrap-brokers --cluster-arn clusterARN ``` 从该命令的 JSON 结果中，保存与名为 `BootstrapBrokerStringSaslScram` 的字符串关联的值。该值将在后面的步骤中使用。 1. 在您的客户端计算机上，创建一个 JAAS 配置文件，其中包含存储在密钥中的用户凭证。例如，对于用户 **alice**，使用以下内容创建一个名为 `users_jaas.conf` 的文件。 ``` KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="alice" password="alice-secret"; }; ``` 1. 使用以下命令将 JAAS 配置文件导出为 `KAFKA_OPTS` 环境参数。 ``` export KAFKA_OPTS=-Djava.security.auth.login.config=/users_jaas.conf ``` 1. 在 `/tmp` 目录中创建一个名为 `kafka.client.truststore.jks` 的文件。 1. （可选）使用以下命令将 JDK 密钥存储文件从 JVM `cacerts` 文件夹复制到您在上一步中创建的 `kafka.client.truststore.jks` 文件。*JDKFolder*替换为实例上的 JDK 文件夹的名称。例如，您的 JDK 文件夹可能命名为 `java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64`。 ``` cp /usr/lib/jvm/JDKFolder/lib/security/cacerts /tmp/kafka.client.truststore.jks ``` 1. 在 Apache Kafka 安装的 `bin` 目录中，创建一个名为 `client_sasl.properties` 的客户端属性文件，其中包含以下内容。此文件可定义 SASL 机制和协议。 ``` security.protocol=SASL_SSL sasl.mechanism=SCRAM-SHA-512 ``` 1. 要创建示例主题，请运行以下命令。*BootstrapBrokerStringSaslScram*替换为您在本主题的步骤 1 中获得的引导代理字符串。 ``` /bin/kafka-topics.sh --create --bootstrap-server BootstrapBrokerStringSaslScram --command-config /client_sasl.properties --replication-factor 3 --partitions 1 --topic ExampleTopicName ``` 1. 要生成您创建的示例主题，请在客户端计算机上运行以下命令。*BootstrapBrokerStringSaslScram*替换为您在本主题的步骤 1 中检索到的引导代理字符串。 ``` /bin/kafka-console-producer.sh --broker-list BootstrapBrokerStringSaslScram --topic ExampleTopicName --producer.config client_sasl.properties ``` 1. 要使用您创建的主题，在您的客户端计算机上运行以下命令。*BootstrapBrokerStringSaslScram*替换为您在本主题的步骤 1 中获得的引导代理字符串。 ``` /bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic ExampleTopicName --from-beginning --consumer.config client_sasl.properties ``` ## 排除连接问题 运行 Kafka 客户端命令时，可能会遇到 Java 堆内存错误，尤其是在使用大型主题或数据集时。出现这些错误的原因是，Kafka 工具作为 Java 应用程序运行，其默认内存设置可能不足以供工作负载使用。 要解决 `Out of Memory Java Heap` 错误，可通过修改 `KAFKA_OPTS` 环境变量包含内存设置来增大 Java 堆大小。 以下示例将最大堆大小设置为 1GB (`-Xmx1G`)。此值可根据可用系统内存和要求进行调整。 ``` export KAFKA_OPTS="-Djava.security.auth.login.config=/users_jaas.conf -Xmx1G" ``` 要使用大型主题，可考虑使用基于时间或基于偏移量的参数（而不是 `--from-beginning`）来限制内存使用量： ``` /bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic ExampleTopicName --max-messages 1000 --consumer.config client_sasl.properties ``` # 使用用户 **创建用户：**您在密钥中以键值对的形式创建用户。在 Secrets Manager 控制台中使用**明文**选项时，应按以下格式指定登录凭证数据。 ``` { "username": "alice", "password": "alice-secret" } ``` **撤销用户访问权限：**要撤销用户访问集群的凭证，建议您先在集群上移除或强制执行 ACL，然后取消与该密钥的关联。这是因为： + 移除用户并不能关闭现有连接。 + 对密钥的更改最多需要 10 分钟才能传播。 有关将 ACL 与 Amazon MSK 结合使用的更多信息，请参阅 [阿帕奇 Kafka ACLs](msk-acls.md)。 对于使用 ZooKeeper 模式的集群，我们建议您限制对 ZooKeeper 节点的访问以防止用户进行修改 ACLs。有关更多信息，请参阅 [控制对亚马逊 MSK ZooKeeper 集群中 Apache 节点的访问权限](zookeeper-security.md)。 # 使用 SCRAM 密钥时的限制 使用 SCRAM 密钥时请注意以下限制： + Amazon MSK 仅支持 SCRAM-SHA-512 身份验证。 + 一个 Amazon MSK 集群最多可拥有 1000 个用户。 + 你必须在你的密钥中 AWS KMS key 使用。您不能将使用默认 Secrets Manager 加密密钥的密钥与 Amazon MSK 一起使用。有关创建 KMS 密钥的信息，请参阅 [Creating symmetric encryption KMS keys](https://docs.aws.amazon.com/kms/latest/developerguide/create-keys.html#create-symmetric-cmk)。 + 您无法在 Secrets Manager 中使用非对称 KMS 密钥。 + 使用该[ BatchAssociateScramSecret](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn-scram-secrets.html#BatchAssociateScramSecret)操作，您一次最多可以将 10 个密钥与一个集群关联。 + 与 Amazon MSK 集群关联的密钥的名称必须带有前缀 **AmazonMSK\$1**。 + 与 Amazon MSK 集群关联的密钥必须与集群位于相同的 Amazon Web Services 账户和 AWS 区域中。 # 阿帕奇 Kafka ACLs Apache Kafka 有一个可插拔的授权器，并附带了授权器实现。 out-of-boxAmazon MSK 在代理上的 `server.properties` 文件中启用此授权方。 Apache Kafka ACLs 的格式为 “主体 P 是 [允许/拒绝] 来自主机 H 对任何与 RP 匹配的资源 R 的操作 O”。 ResourcePattern 如果 RP 与特定资源 R 不匹配，则 R 没有关联 ACLs，因此除了超级用户之外，不允许任何其他人访问 R。要更改此 Apache Kafka 行为，请将该属性`allow.everyone.if.no.acl.found`设置为 true。默认情况下，Amazon MSK 会将其设置为 true。这意味着， ACLs 在 Amazon MSK 集群中，如果您没有明确设置资源，则所有委托人都可以访问此资源。如果您在资源 ACLs 上启用，则只有授权的委托人才能访问该资源。如果要限制对主题的访问并使用 TLS 双向身份验证对客户端进行授权，请 ACLs 使用 Apache Kafka 授权器 CLI 进行添加。有关添加、删除和列出的更多信息 ACLs，请参阅 [Kafka 授权命令行界面](https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Authorization+Command+Line+Interface)。 由于 Amazon MSK 将代理配置为超级用户，因此代理可访问所有主题。这有助于代理从主分区复制消息，无论是否为集群配置定义了 `allow.everyone.if.no.acl.found` 属性。 **添加或删除对主题的读写访问权** 1. 将您的代理添加到 ACL 表中，以允许他们读取所有已 ACLs 存在的主题。要授予代理对主题的读取访问权限，请在可与 MSK 集群通信的客户端计算机上运行以下命令。 *Distinguished-Name*替换为集群中任何引导程序代理的 DNS，然后将此可分辨名称中第一个句点之前的字符串替换为星号 () `*`。例如，如果集群的某个引导代理有 DNS`b-6.mytestcluster.67281x.c4.kafka.us-east-1.amazonaws.com`，则在以下命令*Distinguished-Name*中替换为。`*.mytestcluster.67281x.c4.kafka.us-east-1.amazonaws.com`有关如何获取引导代理的信息，请参阅[获取 Amazon MSK 集群的引导代理](msk-get-bootstrap-brokers.md)。 ``` /bin/kafka-acls.sh --bootstrap-server BootstrapServerString --add --allow-principal "User:CN=Distinguished-Name" --operation Read --group=* --topic Topic-Name ``` 1. 要授予客户端应用对主题的读访问权，请在客户端计算机上运行以下命令。如果您使用双向 TLS 身份验证，请使用与创建私钥时相同的*Distinguished-Name*身份验证。 ``` /bin/kafka-acls.sh --bootstrap-server BootstrapServerString --add --allow-principal "User:CN=Distinguished-Name" --operation Read --group=* --topic Topic-Name ``` 要删除读访问权，您可以运行相同的命令，并将 `--add` 替换为 `--remove`。 1. 要授予对主题的写访问权，请在客户端计算机上运行以下命令。如果您使用双向 TLS 身份验证，请使用与创建私钥时相同的*Distinguished-Name*身份验证。 ``` /bin/kafka-acls.sh --bootstrap-server BootstrapServerString --add --allow-principal "User:CN=Distinguished-Name" --operation Write --topic Topic-Name ``` 要删除写访问权，您可以运行相同的命令，并将 `--add` 替换为 `--remove`。 # 更改 Amazon MSK 集群的安全组 本页介绍如何更改现有 MSK 集群的安全组。您可能需要更改集群的安全组，以便为特定用户组提供访问权限或限制对集群的访问权限。有关安全组的信息，请参阅《Amazon VPC 用户指南》中的[您的 VPC 的安全组](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_SecurityGroups.html)。 1. 使用中的 [ListNodes](https://docs.amazonaws.cn/en_us/msk/1.0/apireference/clusters-clusterarn-nodes.html#ListNodes)API 或 [list-nodes](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kafka/list-nodes.html) 命令获取集群中代理的列表。 AWS CLI 此操作的结果包括与代理关联 IDs 的弹性网络接口 (ENIs)。 1. 登录 AWS 管理控制台 并打开 Amazon EC2 控制台，网址为[https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/)。 1. 使用屏幕右上角附近的下拉列表，选择部署集群的区域。 1. 在左侧窗格的**网络与安全**下，选择**网络接口**。 1. 选择您在第一步中获得的第一个 ENI。选择屏幕顶部的**操作**菜单，然后选择**更改安全组**。将新的安全组分配给此 ENI。对您在第一步中获得 ENIs 的每个内容重复此步骤。 **注意** 您使用 Amazon EC2 控制台对集群安全组所做的更改，不会反映在 MSK 控制台的**网络设置**下。 1. 配置新安全组的规则，确保您的客户端可以访问代理。有关设置安全组规则的信息，请参阅《Amazon VPC 用户指南》中的[添加、删除和更新规则](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_SecurityGroups.html?shortFooter=true#AddRemoveRules)。 **重要** 如果您更改与集群代理关联的安全组，然后向该集群添加新的代理，Amazon MSK 会将新代理与创建集群时与该集群关联的原始安全组关联。但是，要使集群正常运行，其所有代理都必须与同一个安全组关联。因此，如果您在更改安全组后添加新代理，则必须再次执行前面的步骤并更新新代理 ENIs 的代理。 # 控制对亚马逊 MSK ZooKeeper 集群中 Apache 节点的访问权限 出于安全考虑，您可以限制对属于您的 Amazon MSK ZooKeeper 集群的 Apache 节点的访问权限。要限制对节点的访问，您可以为节点分配单独的安全组。然后，您可以决定有权访问该安全组的人员。 **重要** 本节不适用于在 KRaft 模式下运行的集群。请参阅[KRaft 模式](metadata-management.md#kraft-intro)。 **Topics** + [将 Apache ZooKeeper 节点放在单独的安全组中](zookeeper-security-group.md) + [在 Apache 中使用 TLS 安全性 ZooKeeper](zookeeper-security-tls.md) # 将 Apache ZooKeeper 节点放在单独的安全组中 要限制对 Apache ZooKeeper 节点的访问，您可以为其分配单独的安全组。您可以通过设置安全组规则来选择谁有权访问此新安全组。 1. 获取您的集群的 Apache ZooKeeper 连接字符串。要了解如何操作，请参阅[ZooKeeper 模式](metadata-management.md#msk-get-connection-string)。连接字符串包含您的 Apache ZooKeeper 节点的 DNS 名称。 1. 使用 `host` 或 `ping` 等工具将您在上一步中获得的 DNS 名称转换为 IP 地址。稍后您需要在此过程中使用这些 IP 地址，因此请保存这些地址。 1. 登录 AWS 管理控制台 并打开 Amazon EC2 控制台，网址为[https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/)。 1. 在左侧窗格的 **Network & Security (网络与安全性)** 下，选择 **Network Interfaces (网络接口)**。 1. 在网络接口表上方的搜索字段中，键入集群名称，然后键入 return。这会将表中显示的网络接口数限制为与您的集群关联的接口。 1. 选中与列表中的第一个网络接口对应的行开头处的复选框。 1. 在页面底部的详细信息窗格中，查找**主私 IPv4 有 IP**。如果此 IP 地址与您在本过程第一步中获得的 IP 地址相匹配，则表示该网络接口已分配给集群中的一个 Apache ZooKeeper 节点。否则，取消选中此网络接口旁边的复选框，然后选择列表中的下一个网络接口。选择网络接口的顺序无关紧要。在接下来的步骤中，您将对分配给 Apache ZooKeeper 节点的所有网络接口逐一执行相同的操作。 1. 当您选择与 Apache ZooKeeper 节点对应的网络接口时，请选择页面顶部的**操作**菜单，然后选择**更改安全组**。将新安全组分配给此网络接口。有关创建安全组的信息，请参阅 Amazon VPC 文档中的[创建安全组](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_SecurityGroups.html?shortFooter=true#CreatingSecurityGroups)。 1. 重复上一步为与集群的 Apache ZooKeeper 节点关联的所有网络接口分配相同的新安全组。 1. 现在，您可以选择有权访问此新安全组的人员。有关设置安全组规则的信息，请参阅 Amazon VPC 文档中的[添加、删除和更新规则](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_SecurityGroups.html?shortFooter=true#AddRemoveRules)。 # 在 Apache 中使用 TLS 安全性 ZooKeeper 您可以使用 TLS 安全性在客户端和 Apache ZooKeeper 节点之间传输时进行加密。要在 Apache ZooKeeper 节点上实现 TLS 安全，请执行以下操作： + 集群必须使用 Apache Kafka 版本 2.5.1 或更高版本才能在 Apache 中使用 TLS 安全性。 ZooKeeper + 在创建或配置集群时启用 TLS 安全。使用 Apache Kafka 版本 2.5.1 或更高版本创建并启用 TLS 的集群会自动对 Apache 终端节点使用 TLS 安全性。 ZooKeeper 有关设置 TLS 安全的信息，请参阅[Amazon MSK 加密入门](msk-working-with-encryption.md)。 + 使用[DescribeCluster ](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn.html#DescribeCluster)操作检索 TLS Apache ZooKeeper 端点。 + 创建 Apache ZooKeeper 配置文件，以便与`kafka-configs.sh`和[https://kafka.apache.org/documentation/#security_authz_cli](https://kafka.apache.org/documentation/#security_authz_cli)工具或 ZooKeeper 外壳一起使用。对于每个工具，您都使用`--zk-tls-config-file`参数来指定 Apache ZooKeeper 配置。 以下示例显示了一个典型的 Apache ZooKeeper 配置文件： ``` zookeeper.ssl.client.enable=true zookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty zookeeper.ssl.keystore.location=kafka.jks zookeeper.ssl.keystore.password=test1234 zookeeper.ssl.truststore.location=truststore.jks zookeeper.ssl.truststore.password=test1234 ``` + 对于其他命令（例如`kafka-topics`），必须使用`KAFKA_OPTS`环境变量来配置 Apache ZooKeeper 参数。以下示例说明如何配置`KAFKA_OPTS`环境变量以将 Apache ZooKeeper 参数传递给其他命令： ``` export KAFKA_OPTS=" -Dzookeeper.clientCnxnSocket=org.apache.zookeeper.ClientCnxnSocketNetty -Dzookeeper.client.secure=true -Dzookeeper.ssl.trustStore.location=/home/ec2-user/kafka.client.truststore.jks -Dzookeeper.ssl.trustStore.password=changeit" ``` 配置 `KAFKA_OPTS` 环境变量后，您便可正常使用 CLI 命令。以下示例使用环境变量中的 Apache ZooKeeper 配置创建 Apache Kafka 主题：`KAFKA_OPTS` ``` /bin/kafka-topics.sh --create --zookeeper ZooKeeperTLSConnectString --replication-factor 3 --partitions 1 --topic AWSKafkaTutorialTopic ``` **注意** 您在 Apache ZooKeeper 配置文件中使用的参数名称与您在`KAFKA_OPTS`环境变量中使用的参数名称不一致。注意在配置文件和 `KAFKA_OPTS` 环境变量中与参数一起使用的名称。 有关使用 TLS 访问您的 Apache ZooKeeper 节点的更多信息，请参阅 [KIP-515：启用 ZK 客户端使用新的 TLS 支持的身份验证](https://cwiki.apache.org/confluence/display/KAFKA/KIP-515%3A+Enable+ZK+client+to+use+the+new+TLS+supported+authentication)。 # Amazon Managed Streaming for Apache Kafka 的合规性验证 作为 AWS 合规性计划的一部分，第三方审计员将评估 Amazon Managed Streaming for Apache Kafka 的安全性与合规性。其中包括 PCI 和 HIPAA BAA。 有关特定合规计划范围内的 AWS 服务列表，请参阅[按合规计划提供的范围内的亚马逊服务 Amazon Web Ser](https://aws.amazon.com/compliance/services-in-scope/) 。有关一般信息，请参阅[AWS 合规计划AWS](https://aws.amazon.com/compliance/programs/)。 您可以使用下载第三方审计报告 AWS Artifact。有关更多信息，请参阅中的 “[下载报告” 中的 “ AWS Artifact](https://docs.aws.amazon.com/artifact/latest/ug/downloading-documents.html)。 您在使用 Amazon MSK 时的合规责任取决于您的数据的敏感性、贵公司的合规目标以及适用的法律和法规。 AWS 提供了以下资源来帮助实现合规性： + [安全性与合规性快速入门指南](https://aws.amazon.com/quickstart/?awsf.quickstart-homepage-filter=categories%23security-identity-compliance) - 这些部署指南讨论了架构注意事项，并提供了在 AWS上部署基于安全性和合规性的基准环境的步骤。 + [HIPAA 安全与合规架构白皮书 — 本白皮书](https://docs.aws.amazon.com/whitepapers/latest/architecting-hipaa-security-and-compliance-on-aws/architecting-hipaa-security-and-compliance-on-aws.html)描述了公司如何使用来 AWS 创建符合 HIPAA 标准的应用程序。 + [AWS 合规资源AWS](https://aws.amazon.com/compliance/resources/) — 此工作簿和指南集可能适用于您所在的行业和所在地区。 + [使用*AWS Config 开发人员指南*中的规则评估资源](https://docs.aws.amazon.com/config/latest/developerguide/evaluate-config.html) — 该 AWS Config 服务评估您的资源配置在多大程度上符合内部实践、行业准则和法规。 + [AWS Security Hub CSPM](https://docs.aws.amazon.com/securityhub/latest/userguide/what-is-securityhub.html)— 此 AWS 服务可全面了解您的安全状态 AWS ，帮助您检查是否符合安全行业标准和最佳实践。 # Amazon Managed Streaming for Apache Kafka 中的恢复能力 AWS 全球基础设施是围绕 AWS 区域和可用区构建的。 AWS 区域提供多个物理隔离和隔离的可用区，这些可用区通过低延迟、高吞吐量和高度冗余的网络相连。利用可用区，您可以设计和操作在可用区之间无中断地自动实现失效转移的应用程序和数据库。与传统的单个或多个数据中心基础设施相比，可用区具有更高的可用性、容错能力和可扩展性。 有关 AWS 区域和可用区的更多信息，请参阅[AWS 全球基础设施](https://aws.amazon.com/about-aws/global-infrastructure/)。 # Amazon Managed Streaming for Apache Kafka 中的基础设施安全性 作为一项托管服务，适用于 Apache Kafka 的亚马逊托管流媒体受到 AWS 《[亚马逊网络服务：安全流程概述》白皮书中描述的全球网络安全程序的](https://d0.awsstatic.com/whitepapers/Security/AWS_Security_Whitepaper.pdf)保护。 您可以使用 AWS 已发布的 API 调用通过网络访问 Amazon MSK。客户端必须支持传输层安全性协议（TLS）1.0 或更高版本。建议使用 TLS 1.2 或更高版本。客户端还必须支持具有完全向前保密（PFS）的密码套件，例如 Ephemeral Diffie-Hellman（DHE）或 Elliptic Curve Ephemeral Diffie-Hellman（ECDHE）。大多数现代系统（如 Java 7 及更高版本）都支持这些模式。 此外，必须使用访问密钥 ID 和与 IAM 主体关联的秘密访问密钥来对请求进行签名。或者，您可以使用 [AWS Security Token Service](https://docs.aws.amazon.com/STS/latest/APIReference/Welcome.html)（AWS STS）生成临时安全凭证来对请求进行签名。