跨AWS账户连接Amazon MSK遇IAM访问拒绝问题求助

我在一个AWS账户下部署了基于Amazon MSK的Kafka集群，通过AWS Transit Gateway实现跨VPC/跨账户连接。集群同VPC内的客户端使用指定IAM角色可正常连接，但从另一个AWS账户的客户端连接时，收到错误提示：

[SaslAuthenticator-AWS_MSK_IAM] [xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx]: Access denied

配置信息 #

集群端 #

• 认证控制方式：IAM
• Apache Kafka版本：3.4.0
• 公网访问：关闭
• 安全组：入站规则允许所有来源的所有流量
• 访问控制IAM角色（下称the-role-giving-access-to-cluster）：包含MSK客户端IAM角色所需权限，信任策略如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<client account id>:root"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

客户端 #

• 使用客户端：KafkaJS
• 客户端配置：

import { createMechanism } from '@jm18457/kafkajs-msk-iam-authentication-mechanism';
...

const config = {
  ssl: true,
  sasl: createMechanism({
    region: 'eu-west-1',
    credentials: provideAuthentication,
  }),
  clientId: 'my-client',
  brokers: ['my-broker-1:9098', 'my-broker-2:9098'],
  retry: {
    retries: 10,
  },
}

• 身份凭证获取函数provideAuthentication实现：

const sts_client = new STSClient();

export async function provideAuthentication(): Promise<AwsCredentialIdentity> {
  const command = new AssumeRoleCommand({
      RoleArn: 'arn:aws:iam::<cluster-account-id>:role/the-role-giving-access-to-cluster',
      RoleSessionName: 'some-role-session-name',
  });
  const { Credentials: credentials } = await sts_client.send(command);

  return {
    accessKeyId: credentials.AccessKeyId!,
    secretAccessKey: credentials.SecretAccessKey!,
    sessionToken: credentials.SessionToken!,
  };
}

• 运行客户端的ECS任务权限：

{
  "Action": "sts:AssumeRole",
  "Resource": "arn:aws:iam::<cluster-account-id>:role/the-role-giving-access-to-cluster",
  "Effect": "Allow"
}

排查与解决步骤 #

1. 修正IAM角色信任策略范围 #

当前信任策略仅允许客户端账户的root用户扮演角色，但ECS任务实际使用的是任务执行角色/任务角色，并非root用户。需修改信任策略，指定客户端账户中ECS任务使用的角色ARN：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<client account id>:role/<ecs-task-role-name>"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

若需允许客户端账户内多个角色访问，可通过条件限制缩小范围：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::<client account id>:root"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringLike": {
                    "aws:PrincipalArn": "arn:aws:iam::<client account id>:role/*ecs*"
                }
            }
        }
    ]
}

2. 确认集群访问角色权限完整性 #

确保the-role-giving-access-to-cluster拥有MSK集群的必要访问权限，示例权限如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:DescribeCluster",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:WriteData",
                "kafka-cluster:ReadData"
            ],
            "Resource": [
                "arn:aws:kafka:<region>:<cluster-account-id>:cluster/<cluster-name>/<cluster-uuid>",
                "arn:aws:kafka:<region>:<cluster-account-id>:topic/<cluster-name>/<cluster-uuid>/*",
                "arn:aws:kafka:<region>:<cluster-account-id>:group/<cluster-name>/<cluster-uuid>/*"
            ]
        }
    ]
}

替换<region>、<cluster-account-id>、<cluster-name>、<cluster-uuid>为实际值。

3. 校验STS凭证有效性 #

在客户端代码中添加日志，确认provideAuthentication函数获取的AccessKeyId、SecretAccessKey、SessionToken均非空且未过期。若凭证获取失败，会直接导致MSK认证失败。

4. 验证跨VPC网络连通性 #

• 确认客户端VPC与集群VPC的路由表已配置正确的Transit Gateway条目
• 集群安全组需允许客户端VPC的CIDR段访问9098端口（MSK IAM认证端口）
• 测试客户端能否正常解析MSK broker域名（可通过nslookup验证）

5. 检查KafkaJS IAM机制兼容性 #

确保@jm18457/kafkajs-msk-iam-authentication-mechanism版本与KafkaJS版本兼容，且配置的region参数与MSK集群所在区域完全一致。

内容的提问来源于stack exchange，提问作者Christian